Probar request/response en MQTT 5 con Response Topic y Correlation Data

MQTT es publish/subscribe, pero muchos sistemas reales necesitan una pregunta y una respuesta: "lee este ajuste", "ejecuta este comando y dime si funcionó". Request/response en MQTT 5 lo estandariza con dos propiedades del PUBLISH: el solicitante fija un Response Topic (dónde enviar la respuesta) y Correlation Data (un ID opaco que vincula la respuesta con la pregunta). El responder copia la correlation data en su respuesta y publica en el response topic.

Esta guía implementa el patrón en Python y JavaScript, muestra una prueba de una línea desde la terminal y enumera los casos de prueba que detectan los bugs que de verdad llegan a producción: IDs que no coinciden, timeouts ausentes, peticiones concurrentes y ACLs que bloquean la respuesta. Para la parte de diseño del patrón, consulta request/response en MQTT.

Cómo funciona el flujo

El broker no impone nada aquí: simplemente reenvía las propiedades. Es una buena noticia para la portabilidad (cualquier broker MQTT 5 sirve) y una mala para la corrección: si tu responder olvida copiar la correlation data, nada se queja hasta que el solicitante agota su timeout.

Hay dos propiedades relacionadas en CONNECT/CONNACK. Un cliente puede fijar Request Response Information en el CONNECT, y el broker puede contestar con Response Information en el CONNACK: un prefijo de topic que el cliente debería usar para las respuestas. El soporte varía entre brokers, así que la mayoría de aplicaciones construyen el response topic a partir de su client ID. En cualquier caso, el response topic es un nombre de topic: no puede contener + ni #.

Pruébalo primero desde la línea de comandos

Mosquitto incluye mosquitto_rr, un cliente de request/response que usa MQTT 5 por defecto. Arranca un responder falso en una terminal —un suscriptor que imprime las propiedades— para ver qué llega:

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

En una segunda terminal envía una petición con mosquitto_rr. Se suscribe al response topic, publica la petición y espera una respuesta:

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

La terminal 1 muestra el response topic y la correlation data. Para responder a mano, publica en el response topic impreso con la misma correlation data desde una tercera terminal:

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

mosquitto_rr imprime la respuesta y termina; con -W 5 se rinde a los cinco segundos, que es tu primera prueba de timeout. También puedes observar ambos topics en el cliente MQTT online: suscríbete a testmqtt/rr/# y verás cada petición y respuesta lado a lado.

Un responder y un solicitante en Python

Lo siguiente usa paho-mqtt 2.x con MQTT 5. El responder se suscribe a las peticiones y devuelve la 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("petición sin 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()

El solicitante mantiene una tabla de peticiones pendientes indexada por correlation ID, de modo que puede haber muchas peticiones en vuelo y las respuestas pueden llegar en cualquier orden:

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, respuesta]
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("respuesta inesperada o tardía:", 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)  # las peticiones caducadas se descartan
    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"sin respuesta de {device} en {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"respuesta en {ms:.1f} ms: {reply}")

Hay dos detalles importantes. El solicitante espera el SUBACK antes de enviar; si no, un responder rápido puede contestar antes de que exista la suscripción y la respuesta se pierde. Y la petición lleva un Message Expiry Interval igual al timeout, para que un responder que se conecte más tarde no ejecute un comando que el llamante ya dio por perdido (consulta message expiry).

El mismo patrón en MQTT.js

MQTT.js 5 expone las propiedades en las opciones de publicación y en el paquete recibido:

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 esperando a ${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();

Ejecútalo contra el responder de Python anterior: las dos librerías interoperan porque ambas usan solo propiedades estándar de MQTT 5. El generador de código puede crear la parte de conexión para otros lenguajes.

Casos de prueba que vale la pena automatizar

Una capa de request/response parece trivial hasta que falla en producción. Estos casos cubren la mayoría de bugs reales:

1. Camino feliz. Una petición, una respuesta; payload y correlation data coinciden.
2. Responder desconectado. El solicitante debe lanzar un timeout, no quedarse colgado. Mide que salta en el valor configurado.
3. Respuesta tardía. Haz que el responder duerma más que el timeout. La respuesta tardía debe ignorarse, no entregarse a la siguiente petición.
4. Concurrencia. Lanza 100 peticiones en paralelo con un responder que añade un retardo aleatorio; cada respuesta debe llegar al llamante correcto. Las respuestas fuera de orden son normales.
5. Propiedades ausentes. Envía una petición sin Response Topic. El responder debe registrarla y descartarla, no caerse.
6. Petición caducada. Detén el responder, envía una petición con 2 segundos de expiry, espera 5 segundos y reinicia el responder con sesión persistente. No debe ejecutar la petición.
7. Autorización. Con ACLs activadas, el responder necesita permiso de publish en los response topics y el solicitante permiso de subscribe en el suyo. Una respuesta denegada a menudo se descarta sin error; consulta ACL denegada en silencio.

Cuando la respuesta nunca llega

Un hábito útil: haz que el responder registre el response topic y la correlation data recibidos antes de procesar. Cuando una prueba falla, esa única línea de log te dice qué lado está mal.

Medir la latencia de ida y vuelta

Como el solicitante marca la hora de la petición y recibe la respuesta con el mismo reloj, request/response es un lugar fácil para medir la latencia extremo a extremo. Repite la llamada a request() unos cientos de veces y registra p50 y p99. El viaje de ida y vuelta incluye dos saltos por el broker, el tiempo de procesamiento del responder y dos trayectos de red, así que compáralo con una prueba de latencia de un solo salto para ver cuánto tiempo añade tu responder. Repite con QoS 0 y QoS 1: los PUBACK adicionales se notan directamente en los números.

Ver peticiones y respuestas en vivo

Request/response sobre MQTT 5 solo necesita dos propiedades, pero los detalles —suscribirse antes de enviar, timeouts, expiry, IDs de correlación únicos y ACLs correctas— deciden si es fiable. Guarda los scripts anteriores como suite de regresión y ejecútalos con cada cambio de broker o de permisos.

Preguntas frecuentes