Patrón solicitud respuesta MQTT con MQTT 5

El patrón solicitud respuesta MQTT permite que un cliente haga una pregunta y reciba una respuesta concreta, aunque MQTT en sí sea un protocolo publish/subscribe. MQTT 5 añadió dos propiedades, response topic y correlation data, que hacen que esto sea limpio e interoperable. En esta guía construirás una llamada de tipo RPC funcional en Python y Node.js, la probarás en el servidor MQTT público gratuito y aprenderás a gestionar los timeouts y los fallos.

Por qué solicitud/respuesta sobre MQTT

La mayor parte del tráfico MQTT es unidireccional: los sensores publican y los dashboards se suscriben. Pero los sistemas reales también necesitan preguntas con respuesta:

"¿Qué versión de firmware estás ejecutando?"
"Lee el registro 40001 y dime su valor."
"¿Se ha movido realmente la válvula?"
"Ejecuta un autodiagnóstico e informa del resultado."

Podrías abrir un endpoint HTTP en cada dispositivo, pero los dispositivos suelen estar detrás de NAT o en redes celulares donde las conexiones entrantes son imposibles. Como el dispositivo ya mantiene una conexión MQTT saliente, enrutar la solicitud y la respuesta a través del broker es más sencillo y mantiene un único modelo de seguridad. La comparativa entre MQTT y HTTP profundiza en por qué importa ese modelo solo saliente.

Cómo funciona el patrón solicitud respuesta en MQTT 5

El patrón usa dos propiedades de PUBLISH definidas en MQTT 5:

Propiedad	La establece	Propósito
Response Topic	Solicitante	Topic en el que el respondedor debe publicar la respuesta
Correlation Data	Solicitante	Bytes opacos que el respondedor devuelve sin modificar

El flujo es:

El solicitante se suscribe a su propio topic de respuesta, por ejemplo connectmq/demo/k3p8/rpc/replies/<requester-id>.
Publica la solicitud en el topic de solicitudes del respondedor con Response Topic apuntando a ese topic de respuesta y Correlation Data con un ID único.
El respondedor recibe la solicitud, hace el trabajo y publica el resultado en el Response Topic, copiando la Correlation Data.
El solicitante empareja la Correlation Data entrante con la solicitud pendiente y la completa.

El broker no hace ningún seguimiento de las solicitudes. Simplemente reenvía dos mensajes normales. Todo lo que lo convierte en "RPC" vive en los clientes, y por eso el patrón escala bien y funciona en cualquier broker MQTT 5.

Diseño de topics para solicitud/respuesta

Unos buenos topics hacen que el patrón sea fácil de asegurar y depurar:

Solicitudes: <prefix>/rpc/<device-id>/req, para que cada dispositivo se suscriba solo a sus propias solicitudes.
Respuestas: <prefix>/rpc/replies/<requester-id>, para que cada llamante reciba solo sus propias respuestas.
No uses nunca un topic de respuesta compartido para muchos llamantes. Cada llamante recibiría todas las respuestas y tendría que descartar la mayoría.

Si quieres una visión más amplia de cómo estructurar los nombres, consulta las convenciones de nombres de topics MQTT.

Respondedor en Python

Este respondedor hace el papel de un dispositivo. Escucha solicitudes y devuelve un resultado usando el response topic y la correlation data de la solicitud. Usa paho-mqtt 2.x con MQTT 5 en el servidor MQTT público gratuito:

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()

Solicitante en Node.js con timeouts

El solicitante mantiene un mapa de llamadas pendientes indexado por ID de correlación y rechaza cualquier llamada que no reciba respuesta dentro del timeout. 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; // respuesta tardía o desconocida
  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();
    }
  });
});

Arranca el respondedor en Python y luego ejecuta el solicitante en Node.js. Deberías ver reply: { ok: true, rpm: 1480, running: true }. Detén el respondedor y vuelve a ejecutar el solicitante para ver el camino del timeout.

Fíjate en que la solicitud también lleva una expiración de mensaje igual al timeout. Si el dispositivo está desconectado con una sesión persistente, la solicitud se descarta en cuanto el llamante se ha rendido, en lugar de ejecutarse minutos después. Los detalles están en nuestro artículo sobre el message expiry interval de MQTT.

Checklist de diseño de solicitud/respuesta

Suscríbete al topic de respuesta antes de enviar la primera solicitud, o podrías perderte una respuesta rápida.
Usa un ID de correlación único por solicitud, como un UUID.
Implementa siempre un timeout y limpia las entradas pendientes.
Establece una expiración en las solicitudes para que las llamadas obsoletas nunca se ejecuten.
Devuelve errores estructurados (ok: false, código de error) en lugar de quedarte callado.
Haz que las acciones sean idempotentes siempre que puedas, porque QoS 1 puede entregar una solicitud dos veces. Consulta mensajes duplicados en MQTT.
No retengas ni las solicitudes ni las respuestas.

Diseñar el payload de respuesta

El transporte es solo la mitad del contrato; el formato de la respuesta importa igual en cuanto varios equipos llaman a los mismos dispositivos. Un sobre pequeño y coherente mantiene sencillos a los llamantes:

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

Algunas convenciones que funcionan bien en la práctica:

Responde siempre, incluso si falla. Un respondedor silencioso es indistinguible de uno desconectado, y el llamante solo puede agotar el timeout.
Usa códigos legibles por máquina como BUSY, INVALID_PARAMS o UNKNOWN_METHOD, más un mensaje legible por humanos para los logs.
Versiona tus métodos. getStatus.v2 o un campo version te permite cambiar la forma de la respuesta sin romper a los llamantes antiguos.
Separa la aceptación de la finalización. Para cualquier cosa que tarde más de uno o dos segundos, responde de inmediato con {"ok": true, "accepted": true, "jobId": "..."} y publica el progreso y el resultado final en un topic de trabajo. Así el timeout del llamante solo cubre la aceptación.
Indica el tipo de contenido. Establecer la propiedad Content Type de MQTT 5 a application/json en solicitudes y respuestas ayuda a que las herramientas genéricas las muestren correctamente.

Trabajar con clientes MQTT 3.1.1

Si algunos dispositivos solo hablan MQTT 3.1.1, las propiedades no están disponibles. La solución habitual es meter la misma información en el payload, por ejemplo {"replyTo": "...", "corrId": "...", "method": "getStatus"}. Funciona, pero todas las partes deben ponerse de acuerdo en los nombres de los campos, mientras que las propiedades de MQTT 5 son estándar y mantienen limpio el payload. Las flotas mixtas suelen usar ambos enfoques hasta que se retira el firmware antiguo.

Resolución de problemas

No llega ninguna respuesta. Comprueba que el respondedor ve realmente el Response Topic. Si cualquiera de los dos lados se conecta con MQTT 3.1.1, las propiedades se pierden. Confirma también que el solicitante está suscrito exactamente al topic de respuesta.

Las respuestas van al llamante equivocado. Probablemente tu topic de respuesta es compartido. Incluye en él el client ID del solicitante.

La correlation data no coincide. Es binaria. Compara bytes, o decodifica ambos lados de forma coherente, como hacen los ejemplos con cadenas UTF-8.

Las llamadas se quedan colgadas bajo carga. Un respondedor de un solo hilo atiende una solicitud cada vez. Las operaciones largas deberían responder con un estado "accepted" y publicar el resultado final más tarde.

RPC en ConnectMQ

ConnectMQ admite response topic, correlation data y message expiry de MQTT 5.0, así que el código anterior funciona sin cambios contra tu propio broker. Sustituye el host por el de tu broker y añade username: "YOUR_USERNAME", password: "YOUR_PASSWORD" y clientId: "YOUR_CLIENT_ID" del portal. Un broker MQTT gestionado elimina la parte operativa para que puedas centrarte en el contrato de las solicitudes. Para la visión de conjunto, empieza por el tutorial de MQTT.

Empieza con ConnectMQ

Crea un broker gratuito en app.connectmq.com con el plan Serverless, sin tarjeta de crédito. Ejecuta el respondedor en un dispositivo o en un portátil, llámalo desde un script y observa cómo fluyen las solicitudes y respuestas en el explorador de topics en vivo del portal de ConnectMQ.

Preguntas frecuentes

¿Empareja el broker las solicitudes con las respuestas?

No. El broker reenvía dos mensajes independientes. El solicitante los empareja usando la correlation data que eligió.

¿Es obligatoria la correlation data?

No estrictamente, pero sin ella un cliente con varias solicitudes pendientes no puede saber qué respuesta corresponde a cada llamada. Establécela siempre.

¿Puedo usar QoS 0 para solicitud/respuesta?

Puedes, pero entonces las solicitudes perdidas solo se manifiestan como timeouts. Lo habitual es QoS 1 con handlers idempotentes.

¿Sustituye la solicitud/respuesta de MQTT a las API HTTP?

Para comandos y consultas a dispositivos detrás de un NAT, a menudo encaja mejor. Para transferencias de archivos grandes o API públicas, HTTP suele ser más sencillo.

Patrón de solicitud/respuesta en MQTT con response topic y correlation data