Forward MQTT Messages to Webhooks and REST APIs

To forward MQTT messages to a webhook or REST API, run a small consumer that subscribes to the relevant topics and turns every message into an HTTP POST. The subscription is the easy part. What makes a forwarder production-ready is what happens when the HTTP side misbehaves: timeouts, 503 Service Unavailable, 429 Too Many Requests, deploys that drop connections for a minute. This tutorial builds a Node.js forwarder with MQTT.js 5 that retries with exponential backoff and jitter, honours Retry-After, signs requests, sends idempotency keys and parks undeliverable messages on a dead-letter topic.

The architecture

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

Devices publish events to testmqtt/webhook/<device>/events. The forwarder subscribes with QoS 1, posts each event as JSON and decides — based on the HTTP outcome — whether to finish, retry or dead-letter it. This keeps HTTP knowledge out of the devices and keeps the receiving API free from MQTT. If the receiver is a serverless function, the same pattern applies; see MQTT with AWS Lambda for that variant.

Retry rules that work

Use exponential backoff with full jitter: wait a random time between 0 and min(cap, base × 2^attempt). Without the jitter, every message that failed during an outage retries at the same instant and knocks the API over again the moment it recovers.

The forwarder

Install MQTT.js (npm install mqtt). Node 18+ ships fetch and AbortSignal.timeout, so no HTTP library is needed:

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("connected");
  client.subscribe("testmqtt/webhook/+/events", { qos: 1 });
});
client.on("error", (err) => console.error("mqtt error:", 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();                    // free the connection
      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(`attempt ${attempt}: HTTP ${res.status}`);
    } catch (err) {
      console.warn(`attempt ${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 exhausted" };
}

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(`dead-lettering ${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) => {
  // stable key: the same message redelivered gets the same key
  const id = crypto.createHash("sha256").update(topic).update(payload).digest("hex").slice(0, 32);
  queue.push({ topic, payload, id });
  drain();
});

What each part does:

• Timeouts on every request. AbortSignal.timeout(10_000) stops one hanging endpoint from tying up all workers.
• Bounded concurrency. Eight requests in flight at once; tune it to what the API can take. Without a limit, a burst of MQTT traffic becomes a burst of HTTP traffic.
• HMAC signature. The receiver recomputes HMAC-SHA256(secret, body) and rejects requests that do not match — standard webhook practice.
• Idempotency key. Derived from topic and payload, so a redelivered message carries the same key. This assumes payloads contain a timestamp or event ID; if two genuinely different events can be byte-identical, put an ID in the payload and use it instead.
• Dead-letter topic. Undeliverable messages go to testmqtt/webhook/dlq instead of disappearing. Subscribe to it from an alerting job, or replay it after fixing the API.

Delivery guarantees and their limits

By default, MQTT.js acknowledges a QoS 1 message as soon as your message handler returns. The in-memory queue above is therefore lost if the process crashes. That is acceptable for many notification-style webhooks. When it is not, you have two options.

Acknowledge after delivery. Instead of the message handler and in-memory queue, override MQTT.js's handleMessage(packet, callback); the PUBACK is sent only when you call callback. Messages are then processed one at a time per client, which also preserves order:

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 only now
};

Combine it with clean: false (MQTT 3.1.1) or a session expiry interval (MQTT 5) so the broker queues messages while the forwarder is down. The broker must keep persistent sessions for this — use your own private broker instance; the shared test endpoint keeps no offline sessions. See MQTT persistent sessions.

Put a durable queue in between. Write messages to Redis Streams, SQS or a database table, and let HTTP workers drain it. More moving parts, but retries survive restarts.

Either way the result is at-least-once, so the receiving API must tolerate duplicates. Handling duplicate MQTT messages covers deduplication patterns.

Scaling out

One forwarder handles a lot of webhooks, because the work is waiting on HTTP. When you need more, run several instances with a shared subscription such as $share/webhook/testmqtt/webhook/+/events. The broker distributes messages between instances, and if one dies its share moves to the others. Give each instance a unique client ID.

Test the retries

Run a receiver that fails 40 % of requests with 503 and logs duplicates:

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) ? "DUPLICATE" : "accepted ", key, body);
    seen.add(key);
    res.writeHead(204).end();
  });
}).listen(3000, () => console.log("receiver on :3000"));

Start node receiver.mjs, then node forwarder.mjs, and publish a burst of events:

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

Every event should eventually show up as accepted, with attempt N: HTTP 503 warnings in the forwarder log. Then test the other paths:

• Change the receiver to return 400 — messages must land on the dead-letter topic immediately, without retries. Watch it with mosquitto_sub -t testmqtt/webhook/dlq -v or in the online MQTT client.
• Stop the receiver for 30 seconds while publishing — retries should back off and then succeed after it returns.

Watch events and the dead-letter topic in your browser

A webhook forwarder is a few dozen lines of code, and almost all of them are about failure. Classify responses, back off with jitter, send idempotency keys, never drop a message silently, and run the failing-receiver test before you trust it with production traffic. For an API you control, also consider building the HTTP side in Express next to the forwarder.

Frequently asked questions