Testando request/response no MQTT 5 com Response Topic e Correlation Data

MQTT é publish/subscribe, mas muitos sistemas reais precisam de uma pergunta e uma resposta: "leia esta configuração", "execute este comando e me diga se funcionou". Request/response no MQTT 5 padroniza isso com duas propriedades do PUBLISH: o solicitante define um Response Topic (para onde enviar a resposta) e Correlation Data (um ID opaco que liga a resposta à pergunta). O responder copia a correlation data na resposta e publica no response topic.

Este guia implementa o padrão em Python e JavaScript, mostra um teste de uma linha no terminal e lista os casos de teste que pegam os bugs que realmente chegam à produção: IDs que não batem, timeouts ausentes, requisições concorrentes e ACLs que bloqueiam a resposta. Para o lado de design do padrão, veja request/response no MQTT.

Como o fluxo funciona

O broker não impõe nada aqui: ele apenas encaminha as propriedades. Isso é ótimo para portabilidade (qualquer broker MQTT 5 funciona) e ruim para a correção: se o seu responder esquecer de copiar a correlation data, nada reclama até o solicitante estourar o timeout.

Existem duas propriedades relacionadas no CONNECT/CONNACK. Um cliente pode definir Request Response Information no CONNECT, e o broker pode responder com Response Information no CONNACK — um prefixo de topic que o cliente deveria usar para as respostas. O suporte varia entre brokers, então a maioria das aplicações monta o response topic a partir do client ID. De qualquer forma, o response topic é um nome de topic: não pode conter + nem #.

Teste primeiro pela linha de comando

O Mosquitto inclui o mosquitto_rr, um cliente de request/response que usa MQTT 5 por padrão. Inicie um responder falso em um terminal — um subscriber que imprime as propriedades — para ver o que chega:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t 'testmqtt/rr/req/#' \
  -F 'topic=%t response=%R corr=%D payload=%p'

Em um segundo terminal, envie uma requisição com mosquitto_rr. Ele se inscreve no response topic, publica a requisição e espera uma resposta:

mosquitto_rr -h broker.connectmq.com -p 1883 \
  -t testmqtt/rr/req/device-7 \
  -e testmqtt/rr/resp/cli-$$ \
  -m '{"op":"get","key":"firmware"}' -W 5

O terminal 1 mostra o response topic e a correlation data. Para responder manualmente, publique no response topic impresso com a mesma correlation data a partir de um terceiro terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t testmqtt/rr/resp/cli-12345 \
  -D publish correlation-data '<valor impresso no terminal 1>' \
  -m '{"firmware":"2.4.1"}'

O mosquitto_rr imprime a resposta e encerra; com -W 5 ele desiste após cinco segundos, que é o seu primeiro teste de timeout. Você também pode acompanhar os dois topics no cliente MQTT online: inscreva-se em testmqtt/rr/# e cada requisição e resposta aparece lado a lado.

Um responder e um solicitante em Python

O código a seguir usa paho-mqtt 2.x com MQTT 5. O responder se inscreve nas requisições e devolve a correlation data:

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

HOST = "broker.connectmq.com"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe("testmqtt/rr/req/+", qos=1)

def on_message(client, userdata, msg):
    resp_topic = getattr(msg.properties, "ResponseTopic", None)
    corr = getattr(msg.properties, "CorrelationData", None)
    if not resp_topic:
        print("requisição sem Response Topic, ignorada:", msg.topic)
        return
    request = json.loads(msg.payload)
    reply = {"ok": True, "echo": request}
    props = Properties(PacketTypes.PUBLISH)
    if corr is not None:
        props.CorrelationData = corr
    client.publish(resp_topic, json.dumps(reply), qos=1, properties=props)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, 1883)
client.loop_forever()

O solicitante mantém uma tabela de requisições pendentes indexada pelo correlation ID, para que muitas requisições fiquem em voo ao mesmo tempo e as respostas cheguem em qualquer ordem:

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

HOST = "broker.connectmq.com"
CLIENT_ID = f"rr-test-{uuid.uuid4().hex[:8]}"
RESP_TOPIC = f"testmqtt/rr/resp/{CLIENT_ID}"
pending = {}  # correlation id -> [Event, resposta]
ready = threading.Event()

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(RESP_TOPIC, qos=1)

def on_subscribe(client, userdata, mid, reason_codes, properties):
    ready.set()

def on_message(client, userdata, msg):
    corr = getattr(msg.properties, "CorrelationData", b"")
    slot = pending.get(corr)
    if slot is None:
        print("resposta inesperada ou atrasada:", corr)
        return
    slot[1] = json.loads(msg.payload)
    slot[0].set()

def request(client, device, body, timeout=5.0):
    corr = uuid.uuid4().bytes
    slot = pending[corr] = [threading.Event(), None]
    props = Properties(PacketTypes.PUBLISH)
    props.ResponseTopic = RESP_TOPIC
    props.CorrelationData = corr
    props.MessageExpiryInterval = int(timeout)  # requisições vencidas são descartadas
    t0 = time.perf_counter()
    client.publish(f"testmqtt/rr/req/{device}", json.dumps(body), qos=1, properties=props)
    ok = slot[0].wait(timeout)
    pending.pop(corr, None)
    if not ok:
        raise TimeoutError(f"sem resposta de {device} em {timeout}s")
    return slot[1], (time.perf_counter() - t0) * 1000

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID, protocol=mqtt.MQTTv5)
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 1883)
client.loop_start()
ready.wait(5)

reply, ms = request(client, "device-7", {"op": "get", "key": "firmware"})
print(f"resposta em {ms:.1f} ms: {reply}")

Dois detalhes importam. O solicitante espera o SUBACK antes de enviar; caso contrário, um responder rápido pode responder antes de a inscrição existir e a resposta se perde. E a requisição leva um Message Expiry Interval igual ao timeout, para que um responder que fique online depois não execute um comando que o chamador já abandonou (veja message expiry).

O mesmo padrão no MQTT.js

O MQTT.js 5 expõe as propriedades nas opções de publish e no pacote recebido:

import mqtt from "mqtt";
import { randomUUID } from "node:crypto";

const id = `rr-js-${randomUUID().slice(0, 8)}`;
const respTopic = `testmqtt/rr/resp/${id}`;
const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
  protocolVersion: 5,
  clientId: id,
});
await client.subscribeAsync(respTopic, { qos: 1 });

const pending = new Map();
client.on("message", (topic, payload, packet) => {
  const corr = packet.properties?.correlationData?.toString();
  pending.get(corr)?.(JSON.parse(payload.toString()));
  pending.delete(corr);
});

function request(device, body, timeoutMs = 5000) {
  const corr = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(corr);
      reject(new Error(`timeout aguardando ${device}`));
    }, timeoutMs);
    pending.set(corr, (reply) => { clearTimeout(timer); resolve(reply); });
    client.publish(`testmqtt/rr/req/${device}`, JSON.stringify(body), {
      qos: 1,
      properties: { responseTopic: respTopic, correlationData: Buffer.from(corr) },
    });
  });
}

console.log(await request("device-7", { op: "get", key: "firmware" }));
await client.endAsync();

Rode-o contra o responder Python acima: as duas bibliotecas interoperam porque ambas usam apenas propriedades padrão do MQTT 5. O gerador de código cria a parte da conexão para outras linguagens.

Casos de teste que valem a automação

Uma camada de request/response parece trivial até falhar em produção. Estes casos cobrem a maioria dos bugs reais:

1. Caminho feliz. Uma requisição, uma resposta; payload e correlation data batem.
2. Responder offline. O solicitante deve lançar um timeout, não travar. Meça se ele dispara no valor configurado.
3. Resposta atrasada. Faça o responder dormir mais que o timeout. A resposta atrasada deve ser ignorada, não entregue à próxima requisição.
4. Concorrência. Dispare 100 requisições em paralelo com um responder que adiciona atraso aleatório; cada resposta deve chegar ao chamador certo. Respostas fora de ordem são normais.
5. Propriedades ausentes. Envie uma requisição sem Response Topic. O responder deve registrar e descartar, não quebrar.
6. Requisição vencida. Pare o responder, envie uma requisição com expiry de 2 segundos, espere 5 segundos e reinicie o responder com sessão persistente. Ele não deve executar a requisição.
7. Autorização. Com ACLs ativas, o responder precisa de permissão de publish nos response topics e o solicitante de subscribe no seu. Uma resposta negada costuma ser descartada sem erro — veja ACL negada silenciosamente.

Quando a resposta nunca chega

Um hábito útil: faça o responder registrar o response topic e a correlation data recebidos antes de processar. Quando um teste falha, essa única linha de log diz qual lado está errado.

Medindo a latência de ida e volta

Como o solicitante marca o horário da requisição e recebe a resposta no mesmo relógio, request/response é um lugar fácil para medir latência ponta a ponta. Repita a chamada request() algumas centenas de vezes e registre p50 e p99. A ida e volta inclui dois saltos pelo broker, o tempo de processamento do responder e dois caminhos de rede, então compare com um teste de latência de um único salto para ver quanto tempo o seu responder adiciona. Repita com QoS 0 e QoS 1: os PUBACKs extras aparecem diretamente nos números.

Ver requisições e respostas ao vivo

Request/response sobre MQTT 5 precisa de apenas duas propriedades, mas os detalhes — inscrever-se antes de enviar, timeouts, expiry, IDs de correlação únicos e ACLs corretas — decidem se ele é confiável. Guarde os scripts acima como suíte de regressão e rode-os a cada mudança de broker ou de permissões.

Perguntas frequentes