Padrão Request Response MQTT com MQTT 5

O padrão request response MQTT permite que um client faça uma pergunta e receba de volta uma resposta específica, mesmo que o MQTT seja, em si, um protocolo publish/subscribe. O MQTT 5 adicionou duas propriedades, response topic e correlation data, que tornam isso limpo e interoperável. Neste guia você vai construir uma chamada no estilo RPC funcionando em Python e Node.js, testá-la em um broker público e aprender a lidar com timeouts e falhas.

Por que Request/Response sobre MQTT

A maior parte do tráfego MQTT é unidirecional: sensores publicam, dashboards assinam. Mas sistemas reais também precisam de perguntas com respostas:

"Qual versão de firmware você está rodando?"
"Leia o registrador 40001 e me diga o valor."
"A válvula realmente se moveu?"
"Execute um autoteste e informe o resultado."

Você poderia abrir um endpoint HTTP em cada dispositivo, mas os dispositivos costumam ficar atrás de NAT ou de redes celulares, onde conexões de entrada são impossíveis. Como o dispositivo já mantém uma conexão MQTT de saída, rotear a requisição e a resposta pelo broker é mais simples e mantém um único modelo de segurança. A comparação entre MQTT e HTTP aprofunda por que esse modelo somente de saída é importante.

Como o padrão request response MQTT funciona no MQTT 5

O padrão usa duas propriedades de PUBLISH definidas no MQTT 5:

Propriedade	Definida por	Finalidade
Response Topic	Requisitante	Topic no qual o respondente deve publicar a resposta
Correlation Data	Requisitante	Bytes opacos que o respondente devolve sem alteração

O fluxo é:

O requisitante assina o seu próprio topic de resposta, por exemplo connectmq/demo/k3p8/rpc/replies/<requester-id>.
Ele publica a requisição no topic de requisições do respondente, com o Response Topic apontando para esse topic de resposta e o Correlation Data definido com um ID único.
O respondente recebe a requisição, faz o trabalho e publica o resultado no Response Topic, copiando o Correlation Data.
O requisitante associa o Correlation Data recebido à requisição pendente e a conclui.

O broker não rastreia requisições de forma alguma. Ele apenas encaminha duas mensagens comuns. Tudo o que transforma isso em "RPC" vive nos clients, e é por isso que o padrão escala bem e funciona em qualquer broker MQTT 5.

Design de topics para Request/Response

Bons topics deixam o padrão fácil de proteger e depurar:

Requisições: <prefix>/rpc/<device-id>/req, para que cada dispositivo assine apenas as próprias requisições.
Respostas: <prefix>/rpc/replies/<requester-id>, para que cada chamador receba apenas as próprias respostas.
Nunca use um topic de resposta compartilhado entre vários chamadores. Cada chamador receberia todas as respostas e teria de descartar a maioria delas.

Para uma visão mais ampla de como estruturar nomes, veja convenções de nomenclatura de topics MQTT.

Respondente em Python

Este respondente faz o papel de um dispositivo. Ele escuta requisições e devolve um resultado usando o response topic e o correlation data da requisição. Ele usa paho-mqtt 2.x com MQTT 5 no servidor MQTT público (apenas para testes):

import json
import uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

BROKER = "broker.connectmq.com"
REQ_TOPIC = "connectmq/demo/k3p8/rpc/pump-07/req"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(REQ_TOPIC, qos=1)
    print("responder ready")

def on_message(client, userdata, msg):
    props = msg.properties
    response_topic = getattr(props, "ResponseTopic", None)
    correlation = getattr(props, "CorrelationData", None)
    if not response_topic:
        print("request without response topic, ignoring")
        return

    request = json.loads(msg.payload)
    if request.get("method") == "getStatus":
        result = {"ok": True, "rpm": 1480, "running": True}
    else:
        result = {"ok": False, "error": "unknown method"}

    reply_props = Properties(PacketTypes.PUBLISH)
    if correlation is not None:
        reply_props.CorrelationData = correlation
    client.publish(response_topic, json.dumps(result), qos=1, properties=reply_props)

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=f"responder-{uuid.uuid4().hex[:8]}",
    protocol=mqtt.MQTTv5,
)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_forever()

Requisitante em Node.js com timeouts

O requisitante mantém um mapa de chamadas pendentes indexado pelo ID de correlação e rejeita qualquer chamada que não receba resposta dentro do timeout. Ele usa MQTT.js v5 sobre TLS:

const mqtt = require("mqtt");
const { randomUUID } = require("crypto");

const clientId = "requester-" + randomUUID().slice(0, 8);
const replyTopic = `connectmq/demo/k3p8/rpc/replies/${clientId}`;
const pending = new Map();

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId,
  protocolVersion: 5,
});

function call(deviceId, method, params = {}, timeoutMs = 5000) {
  const id = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(id);
      reject(new Error(`timeout calling ${method} on ${deviceId}`));
    }, timeoutMs);
    pending.set(id, { resolve, timer });
    client.publish(
      `connectmq/demo/k3p8/rpc/${deviceId}/req`,
      JSON.stringify({ method, params }),
      {
        qos: 1,
        properties: {
          responseTopic: replyTopic,
          correlationData: Buffer.from(id),
          messageExpiryInterval: Math.ceil(timeoutMs / 1000),
        },
      }
    );
  });
}

client.on("message", (topic, payload, packet) => {
  const corr = packet.properties && packet.properties.correlationData;
  if (!corr) return;
  const id = corr.toString();
  const entry = pending.get(id);
  if (!entry) return; // resposta atrasada ou desconhecida
  clearTimeout(entry.timer);
  pending.delete(id);
  entry.resolve(JSON.parse(payload.toString()));
});

client.on("connect", () => {
  client.subscribe(replyTopic, { qos: 1 }, async (err) => {
    if (err) return console.error(err);
    try {
      const status = await call("pump-07", "getStatus");
      console.log("reply:", status);
    } catch (e) {
      console.error(e.message);
    } finally {
      client.end();
    }
  });
});

Inicie o respondente em Python e depois execute o requisitante em Node.js. Você deve ver reply: { ok: true, rpm: 1480, running: true }. Pare o respondente e execute o requisitante novamente para ver o caminho de timeout.

Repare que a requisição também carrega um message expiry igual ao timeout. Se o dispositivo estiver offline com uma sessão persistente, a requisição é descartada assim que o chamador desiste, em vez de ser executada minutos depois. Os detalhes estão no nosso artigo sobre o message expiry interval do MQTT.

Checklist de design de Request/Response

Assine o topic de resposta antes de enviar a primeira requisição, ou você pode perder uma resposta rápida.
Use um ID de correlação único por requisição, como um UUID.
Sempre implemente um timeout e limpe as entradas pendentes.
Defina message expiry nas requisições para que chamadas obsoletas nunca sejam executadas.
Retorne erros estruturados (ok: false, código de erro) em vez de ficar em silêncio.
Torne as ações idempotentes sempre que possível, porque o QoS 1 pode entregar uma requisição duas vezes. Veja mensagens duplicadas no MQTT.
Não use retain em requisições nem em respostas.

Desenhando o payload da resposta

O transporte é só metade do contrato; o formato da resposta importa tanto quanto quando várias equipes chamam os mesmos dispositivos. Um envelope pequeno e consistente mantém os chamadores simples:

{"ok": false, "code": "BUSY", "message": "self-test already running", "retryAfter": 10}

Algumas convenções que se sustentam bem na prática:

Sempre responda, mesmo em caso de falha. Um respondente silencioso é indistinguível de um offline, e o chamador só pode esperar o timeout.
Use códigos legíveis por máquina, como BUSY, INVALID_PARAMS ou UNKNOWN_METHOD, além de uma mensagem legível por humanos para os logs.
Versione seus métodos. getStatus.v2 ou um campo version permite mudar o formato da resposta sem quebrar chamadores mais antigos.
Separe aceitação de conclusão. Para qualquer coisa que leve mais de um ou dois segundos, responda imediatamente com {"ok": true, "accepted": true, "jobId": "..."} e publique o progresso e o resultado final em um topic de job. O timeout do chamador passa a cobrir apenas a aceitação.
Indique o content type. Definir a propriedade Content Type do MQTT 5 como application/json em requisições e respostas ajuda ferramentas genéricas a exibi-las corretamente.

Trabalhando com clients MQTT 3.1.1

Se alguns dispositivos só falam MQTT 3.1.1, as propriedades não estão disponíveis. A solução comum é colocar a mesma informação no payload, por exemplo {"replyTo": "...", "corrId": "...", "method": "getStatus"}. Funciona, mas todas as partes precisam concordar com os nomes dos campos, enquanto as propriedades do MQTT 5 são padronizadas e mantêm o payload limpo. Frotas mistas costumam rodar as duas abordagens até que o firmware antigo seja aposentado.

Solução de problemas

Nenhuma resposta chega. Verifique se o respondente realmente enxerga o Response Topic. Se qualquer um dos lados se conectar com MQTT 3.1.1, as propriedades são descartadas. Confirme também que o requisitante assinou exatamente o topic de resposta.

As respostas vão para o chamador errado. Seu topic de resposta provavelmente está compartilhado. Inclua nele o client ID do requisitante.

O correlation data não bate. Ele é binário. Compare bytes, ou decodifique os dois lados de forma consistente, como os exemplos fazem com strings UTF-8.

As chamadas travam sob carga. Um respondente single-thread trata uma requisição por vez. Operações longas devem responder com um status "accepted" e publicar o resultado final depois.

Rodando RPC no ConnectMQ

O ConnectMQ suporta response topic, correlation data e message expiry do MQTT 5.0, então o código acima roda sem alterações no seu próprio broker. Troque o host pelo do seu broker e adicione username: "YOUR_USERNAME", password: "YOUR_PASSWORD" e clientId: "YOUR_CLIENT_ID" obtidos no portal. Um broker MQTT gerenciado elimina a parte operacional para que você possa se concentrar no contrato das requisições. Para ter a visão geral, comece pelo tutorial de MQTT.

Comece a usar o ConnectMQ

Crie um broker gratuito em app.connectmq.com no plano Serverless, sem cartão de crédito. Rode o respondente em um dispositivo ou notebook, chame-o a partir de um script e acompanhe requisições e respostas fluindo no explorador de topics ao vivo do portal ConnectMQ.

Perguntas frequentes

O broker associa requisições e respostas?

Não. O broker encaminha duas mensagens independentes. O requisitante as associa usando o correlation data que ele mesmo escolheu.

O correlation data é obrigatório?

Não estritamente, mas sem ele um client com várias requisições em aberto não consegue saber qual resposta pertence a qual chamada. Sempre o defina.

Posso usar QoS 0 para request/response?

Pode, mas requisições perdidas só vão aparecer como timeouts. QoS 1 com handlers idempotentes é a escolha usual.

O request/response no MQTT substitui APIs HTTP?

Para comandos e consultas a dispositivos atrás de NAT, muitas vezes é uma opção melhor. Para transferências de arquivos grandes ou APIs públicas, o HTTP costuma ser mais simples.

Padrão Request/Response no MQTT com Response Topic e Correlation Data