Transférer des messages MQTT vers des webhooks et des API REST

Pour transférer des messages MQTT vers un webhook ou une API REST, exécutez un petit consommateur qui s'abonne aux topics concernés et transforme chaque message en POST HTTP. L'abonnement est la partie facile. Ce qui rend un forwarder prêt pour la production, c'est ce qui se passe quand le côté HTTP se comporte mal : timeouts, 503 Service Unavailable, 429 Too Many Requests, déploiements qui coupent les connexions pendant une minute. Ce tutoriel construit un forwarder Node.js avec MQTT.js 5 qui réessaie avec un backoff exponentiel et du jitter, respecte Retry-After, signe les requêtes, envoie des clés d'idempotence et met de côté les messages non livrables sur un topic dead-letter.

L'architecture

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

Les appareils publient des événements sur testmqtt/webhook/<device>/events. Le forwarder s'abonne en QoS 1, poste chaque événement en JSON et décide, selon le résultat HTTP, de le terminer, de le réessayer ou de l'envoyer en dead-letter. Les appareils n'ont ainsi rien à savoir de HTTP, et l'API réceptrice rien à savoir de MQTT. Si le récepteur est une fonction serverless, le même schéma s'applique ; voir MQTT avec AWS Lambda pour cette variante.

Des règles de retry qui fonctionnent

Utilisez un backoff exponentiel avec full jitter : attendez un temps aléatoire entre 0 et min(cap, base × 2^attempt). Sans jitter, tous les messages qui ont échoué pendant une panne réessaient au même instant et font retomber l'API au moment même où elle se rétablit.

Le forwarder

Installez MQTT.js (npm install mqtt). Node 18+ fournit fetch et AbortSignal.timeout, aucune bibliothèque HTTP n'est donc nécessaire :

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("connecté");
  client.subscribe("testmqtt/webhook/+/events", { qos: 1 });
});
client.on("error", (err) => console.error("erreur 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();                    // libérer la connexion
      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(`tentative ${attempt} : HTTP ${res.status}`);
    } catch (err) {
      console.warn(`tentative ${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: "retries épuisés" };
}

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(`envoi en 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) => {
  // clé stable : un même message relivré reçoit la même clé
  const id = crypto.createHash("sha256").update(topic).update(payload).digest("hex").slice(0, 32);
  queue.push({ topic, payload, id });
  drain();
});

Le rôle de chaque partie :

• Un timeout sur chaque requête. AbortSignal.timeout(10_000) empêche un endpoint bloqué d'immobiliser tous les workers.
• Concurrence bornée. Huit requêtes en vol à la fois ; ajustez selon ce que l'API supporte. Sans limite, une rafale de trafic MQTT devient une rafale de trafic HTTP.
• Signature HMAC. Le récepteur recalcule HMAC-SHA256(secret, body) et rejette les requêtes qui ne correspondent pas : pratique standard pour les webhooks.
• Clé d'idempotence. Dérivée du topic et du payload, si bien qu'un message relivré porte la même clé. Cela suppose que les payloads contiennent un timestamp ou un ID d'événement ; si deux événements réellement différents peuvent être identiques octet pour octet, mettez un ID dans le payload et utilisez-le.
• Topic dead-letter. Les messages non livrables vont sur testmqtt/webhook/dlq au lieu de disparaître. Abonnez-y un job d'alerte, ou rejouez-les après avoir corrigé l'API.

Garanties de livraison et leurs limites

Par défaut, MQTT.js acquitte un message QoS 1 dès que votre handler message rend la main. La file en mémoire ci-dessus est donc perdue si le processus plante. C'est acceptable pour beaucoup de webhooks de type notification. Sinon, vous avez deux options.

Acquitter après la livraison. À la place du handler message et de la file en mémoire, surchargez handleMessage(packet, callback) de MQTT.js ; le PUBACK n'est envoyé que lorsque vous appelez callback. Les messages sont alors traités un par un par client, ce qui préserve aussi l'ordre :

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 seulement maintenant
};

Combinez cela avec clean: false (MQTT 3.1.1) ou un session expiry interval (MQTT 5) pour que le broker mette les messages en file pendant l'arrêt du forwarder. Le broker doit pour cela conserver des sessions persistantes : utilisez votre propre instance de broker privée, car l'endpoint de test partagé ne garde aucune session hors ligne. Voir les sessions persistantes MQTT.

Placer une file durable entre les deux. Écrivez les messages dans Redis Streams, SQS ou une table de base de données, et laissez des workers HTTP la vider. Plus de pièces, mais les retries survivent aux redémarrages.

Dans les deux cas, le résultat est at-least-once : l'API réceptrice doit tolérer les doublons. Gérer les messages MQTT en double présente des modèles de déduplication.

Monter en charge

Un seul forwarder gère beaucoup de webhooks, car l'essentiel du travail consiste à attendre HTTP. Quand il en faut plus, lancez plusieurs instances avec un abonnement partagé comme $share/webhook/testmqtt/webhook/+/events. Le broker répartit les messages entre les instances et, si l'une tombe, sa part passe aux autres. Donnez à chaque instance un client ID unique.

Tester les retries

Lancez un récepteur qui fait échouer 40 % des requêtes en 503 et journalise les doublons :

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) ? "DOUBLON" : "accepté", key, body);
    seen.add(key);
    res.writeHead(204).end();
  });
}).listen(3000, () => console.log("récepteur sur :3000"));

Démarrez node receiver.mjs, puis node forwarder.mjs, et publiez une rafale d'événements :

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

Chaque événement doit finir par apparaître comme accepté, avec des avertissements tentative N : HTTP 503 dans le journal du forwarder. Testez ensuite les autres chemins :

• Faites renvoyer 400 au récepteur : les messages doivent arriver immédiatement sur le topic dead-letter, sans retry. Observez-le avec mosquitto_sub -t testmqtt/webhook/dlq -v ou dans le client MQTT en ligne.
• Arrêtez le récepteur 30 secondes pendant que vous publiez : les retries doivent s'espacer puis réussir à son retour.

Observer les événements et le topic dead-letter dans le navigateur

Un forwarder de webhooks tient en quelques dizaines de lignes, et presque toutes concernent les échecs. Classez les réponses, appliquez un backoff avec jitter, envoyez des clés d'idempotence, ne jetez jamais un message en silence et lancez le test avec le récepteur défaillant avant de lui confier du trafic de production. Si l'API vous appartient, envisagez aussi de construire le côté HTTP avec Express, à côté du forwarder.

Questions fréquentes