Reenvía mensajes MQTT a webhooks y APIs REST

Para reenviar mensajes MQTT a un webhook o a una API REST, ejecuta un pequeño consumidor que se suscriba a los topics relevantes y convierta cada mensaje en un POST HTTP. La suscripción es la parte fácil. Lo que hace que un reenviador esté listo para producción es lo que ocurre cuando el lado HTTP falla: timeouts, 503 Service Unavailable, 429 Too Many Requests, despliegues que cortan conexiones durante un minuto. Este tutorial construye un reenviador en Node.js con MQTT.js 5 que reintenta con backoff exponencial y jitter, respeta Retry-After, firma las peticiones, envía claves de idempotencia y aparca los mensajes que no se pueden entregar en un topic de dead-letter.

La arquitectura

devices ──MQTT──▶ broker ──▶ forwarder ──HTTPS POST──▶ webhook / REST API
                                 │
                                 └──▶ testmqtt/webhook/dlq (failed deliveries)

Los dispositivos publican eventos en testmqtt/webhook/<device>/events. El reenviador se suscribe con QoS 1, envía cada evento como JSON y decide, según el resultado HTTP, si lo da por terminado, lo reintenta o lo manda a dead-letter. Así los dispositivos no necesitan saber nada de HTTP y la API receptora no necesita saber nada de MQTT. Si el receptor es una función serverless, se aplica el mismo patrón; consulta MQTT con AWS Lambda para esa variante.

Reglas de reintento que funcionan

Usa backoff exponencial con full jitter: espera un tiempo aleatorio entre 0 y min(cap, base × 2^attempt). Sin jitter, todos los mensajes que fallaron durante una caída se reintentan en el mismo instante y vuelven a tumbar la API justo cuando se recupera.

El reenviador

Instala MQTT.js (npm install mqtt). Node 18+ incluye fetch y AbortSignal.timeout, así que no hace falta ninguna librería HTTP:

import mqtt from "mqtt";
import crypto from "node:crypto";

const WEBHOOK_URL = process.env.WEBHOOK_URL ?? "http://localhost:3000/hook";
const SECRET = process.env.WEBHOOK_SECRET ?? "change-me";
const MAX_ATTEMPTS = 6;
const CONCURRENCY = 8;
const DLQ_TOPIC = "testmqtt/webhook/dlq";

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "webhook-forwarder-1",
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
});

client.on("connect", () => {
  console.log("conectado");
  client.subscribe("testmqtt/webhook/+/events", { qos: 1 });
});
client.on("error", (err) => console.error("error mqtt:", err.message));

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const retryable = (status) => status === 408 || status === 429 || status >= 500;

async function deliver(job) {
  const body = JSON.stringify({
    topic: job.topic,
    payload: job.payload.toString("utf8"),
    receivedAt: new Date().toISOString(),
  });
  const signature = crypto.createHmac("sha256", SECRET).update(body).digest("hex");

  for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
    let retryAfterMs = 0;
    try {
      const res = await fetch(WEBHOOK_URL, {
        method: "POST",
        headers: {
          "content-type": "application/json",
          "idempotency-key": job.id,
          "x-signature-sha256": signature,
        },
        body,
        signal: AbortSignal.timeout(10_000),
      });
      await res.body?.cancel();                    // liberar la conexión
      if (res.ok) return { ok: true };
      if (!retryable(res.status)) return { ok: false, reason: `HTTP ${res.status}` };
      retryAfterMs = (Number(res.headers.get("retry-after")) || 0) * 1000;
      console.warn(`intento ${attempt}: HTTP ${res.status}`);
    } catch (err) {
      console.warn(`intento ${attempt}: ${err.name} ${err.message}`);
    }
    if (attempt < MAX_ATTEMPTS) {
      const backoff = Math.random() * Math.min(30_000, 500 * 2 ** attempt);
      await sleep(Math.max(backoff, retryAfterMs));
    }
  }
  return { ok: false, reason: "reintentos agotados" };
}

const queue = [];
let active = 0;

function drain() {
  while (active < CONCURRENCY && queue.length > 0) {
    const job = queue.shift();
    active++;
    deliver(job)
      .then((result) => {
        if (!result.ok) {
          console.error(`enviando a dead-letter ${job.id}: ${result.reason}`);
          client.publish(DLQ_TOPIC, JSON.stringify({
            topic: job.topic,
            payload: job.payload.toString("base64"),
            id: job.id,
            reason: result.reason,
          }), { qos: 1 });
        }
      })
      .finally(() => { active--; drain(); });
  }
}

client.on("message", (topic, payload) => {
  // clave estable: el mismo mensaje reentregado recibe la misma clave
  const id = crypto.createHash("sha256").update(topic).update(payload).digest("hex").slice(0, 32);
  queue.push({ topic, payload, id });
  drain();
});

Qué hace cada parte:

• Timeout en cada petición. AbortSignal.timeout(10_000) evita que un endpoint colgado bloquee a todos los workers.
• Concurrencia limitada. Ocho peticiones en vuelo a la vez; ajústalo a lo que aguante la API. Sin límite, una ráfaga de tráfico MQTT se convierte en una ráfaga de tráfico HTTP.
• Firma HMAC. El receptor recalcula HMAC-SHA256(secret, body) y rechaza las peticiones que no coinciden: práctica estándar en webhooks.
• Clave de idempotencia. Se deriva del topic y del payload, así que un mensaje reentregado lleva la misma clave. Esto supone que los payloads incluyen un timestamp o un ID de evento; si dos eventos realmente distintos pueden ser idénticos byte a byte, pon un ID en el payload y úsalo en su lugar.
• Topic de dead-letter. Los mensajes que no se pueden entregar van a testmqtt/webhook/dlq en lugar de desaparecer. Suscríbete desde un job de alertas o reprodúcelos tras arreglar la API.

Garantías de entrega y sus límites

Por defecto, MQTT.js confirma un mensaje QoS 1 en cuanto termina tu handler de message. Por eso la cola en memoria anterior se pierde si el proceso se cae. Eso es aceptable para muchos webhooks de tipo notificación. Cuando no lo es, tienes dos opciones.

Confirmar después de entregar. En lugar del handler de message y la cola en memoria, sobrescribe handleMessage(packet, callback) de MQTT.js; el PUBACK solo se envía cuando llamas a callback. Los mensajes se procesan entonces de uno en uno por cliente, lo que además conserva el orden:

client.handleMessage = (packet, callback) => {
  const id = crypto.createHash("sha256").update(packet.topic).update(packet.payload).digest("hex").slice(0, 32);
  deliver({ topic: packet.topic, payload: packet.payload, id })
    .then((result) => { if (!result.ok) console.error("dead-letter:", id, result.reason); })
    .finally(() => callback());   // PUBACK solo ahora
};

Combínalo con clean: false (MQTT 3.1.1) o con un session expiry interval (MQTT 5) para que el broker encole los mensajes mientras el reenviador está caído. Para ello el broker debe conservar sesiones persistentes: usa tu propia instancia privada de broker, porque el endpoint de pruebas compartido no guarda sesiones offline. Consulta sesiones persistentes en MQTT.

Poner una cola duradera en medio. Escribe los mensajes en Redis Streams, SQS o una tabla de base de datos y deja que los workers HTTP la vacíen. Más piezas, pero los reintentos sobreviven a los reinicios.

En ambos casos el resultado es at-least-once, así que la API receptora debe tolerar duplicados. Gestionar mensajes MQTT duplicados cubre patrones de deduplicación.

Escalar horizontalmente

Un solo reenviador gestiona muchos webhooks, porque la mayor parte del trabajo es esperar a HTTP. Cuando necesites más, ejecuta varias instancias con una suscripción compartida como $share/webhook/testmqtt/webhook/+/events. El broker reparte los mensajes entre las instancias y, si una muere, su parte pasa a las demás. Da a cada instancia un client ID único.

Prueba los reintentos

Ejecuta un receptor que falle el 40 % de las peticiones con 503 y registre los duplicados:

import http from "node:http";

const seen = new Set();
http.createServer((req, res) => {
  let body = "";
  req.on("data", (chunk) => (body += chunk));
  req.on("end", () => {
    if (Math.random() < 0.4) {
      res.writeHead(503, { "retry-after": "1" });
      return res.end();
    }
    const key = req.headers["idempotency-key"];
    console.log(seen.has(key) ? "DUPLICADO" : "aceptado ", key, body);
    seen.add(key);
    res.writeHead(204).end();
  });
}).listen(3000, () => console.log("receptor en :3000"));

Arranca node receiver.mjs, después node forwarder.mjs, y publica una ráfaga de eventos:

for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
    -t testmqtt/webhook/door-1/events -m "{\"event\":\"open\",\"n\":$i,\"ts\":$(date +%s)}"
done

Todos los eventos deberían acabar apareciendo como aceptado, con avisos intento N: HTTP 503 en el log del reenviador. Después prueba los otros caminos:

• Cambia el receptor para que devuelva 400: los mensajes deben ir al topic de dead-letter de inmediato, sin reintentos. Obsérvalo con mosquitto_sub -t testmqtt/webhook/dlq -v o en el cliente MQTT online.
• Detén el receptor durante 30 segundos mientras publicas: los reintentos deberían espaciarse y tener éxito cuando vuelva.

Observa los eventos y el topic de dead-letter en el navegador

Un reenviador de webhooks son unas pocas decenas de líneas, y casi todas tratan de fallos. Clasifica las respuestas, aplica backoff con jitter, envía claves de idempotencia, no descartes nunca un mensaje en silencio y ejecuta la prueba con el receptor que falla antes de confiarle tráfico de producción. Si controlas la API, considera también construir el lado HTTP con Express junto al reenviador.

Preguntas frecuentes