MQTT Mesajlarını Webhook'lara ve REST API'lere Yönlendirin

MQTT mesajlarını bir webhook'a veya REST API'ye yönlendirmek için ilgili topic'lere subscribe olan ve her mesajı bir HTTP POST isteğine dönüştüren küçük bir consumer çalıştırın. Subscribe olmak işin kolay kısmı. Bir yönlendiriciyi production'a hazır yapan şey, HTTP tarafı sorun çıkardığında ne olduğudur: timeout'lar, 503 Service Unavailable, 429 Too Many Requests, bağlantıları bir dakika boyunca koparan deploy'lar. Bu eğitimde MQTT.js 5 ile; üstel backoff ve jitter ile yeniden deneyen, Retry-After'a uyan, istekleri imzalayan, idempotency key gönderen ve teslim edilemeyen mesajları bir dead-letter topic'ine park eden bir Node.js yönlendiricisi kuruyoruz.

Mimari

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

Cihazlar olayları testmqtt/webhook/<device>/events topic'ine publish eder. Yönlendirici QoS 1 ile subscribe olur, her olayı JSON olarak gönderir ve HTTP sonucuna göre işi bitirmeye, yeniden denemeye ya da dead-letter'a göndermeye karar verir. Böylece cihazların HTTP bilmesi, alıcı API'nin de MQTT bilmesi gerekmez. Alıcı bir serverless fonksiyonsa aynı desen geçerlidir; bu varyant için AWS Lambda ile MQTT yazısına bakın.

İşe yarayan yeniden deneme kuralları

Full jitter ile üstel backoff kullanın: 0 ile min(cap, base × 2^attempt) arasında rastgele bir süre bekleyin. Jitter olmadan, bir kesinti sırasında başarısız olan tüm mesajlar aynı anda yeniden denenir ve API tam toparlandığı anda onu tekrar devirir.

Yönlendirici

MQTT.js'i kurun (npm install mqtt). Node 18+ fetch ve AbortSignal.timeout ile birlikte gelir, dolayısıyla bir HTTP kütüphanesine gerek yok:

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("bağlandı");
  client.subscribe("testmqtt/webhook/+/events", { qos: 1 });
});
client.on("error", (err) => console.error("mqtt hatası:", 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();                    // bağlantıyı serbest bırak
      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(`deneme ${attempt}: HTTP ${res.status}`);
    } catch (err) {
      console.warn(`deneme ${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: "deneme hakkı bitti" };
}

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-letter'a gönderiliyor ${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) => {
  // sabit key: yeniden iletilen aynı mesaj aynı key'i alır
  const id = crypto.createHash("sha256").update(topic).update(payload).digest("hex").slice(0, 32);
  queue.push({ topic, payload, id });
  drain();
});

Her parçanın görevi:

• Her istekte timeout. AbortSignal.timeout(10_000), takılan tek bir endpoint'in tüm worker'ları kilitlemesini önler.
• Sınırlı eşzamanlılık. Aynı anda sekiz istek; API'nin kaldırabileceğine göre ayarlayın. Sınır olmadan bir MQTT trafik patlaması bir HTTP trafik patlamasına dönüşür.
• HMAC imzası. Alıcı HMAC-SHA256(secret, body) değerini yeniden hesaplar ve eşleşmeyen istekleri reddeder; standart webhook pratiğidir.
• Idempotency key. Topic ve payload'dan türetilir; böylece yeniden iletilen bir mesaj aynı key'i taşır. Bu, payload'ların bir timestamp veya olay ID'si içerdiğini varsayar; gerçekten farklı iki olay bayt bayt aynı olabiliyorsa payload'a bir ID koyun ve onu kullanın.
• Dead-letter topic. Teslim edilemeyen mesajlar kaybolmak yerine testmqtt/webhook/dlq topic'ine gider. Bir alarm job'ından bu topic'e subscribe olun ya da API'yi düzelttikten sonra mesajları yeniden oynatın.

Teslimat garantileri ve sınırları

MQTT.js varsayılan olarak bir QoS 1 mesajını message handler'ınız döner dönmez onaylar. Bu yüzden yukarıdaki bellek içi kuyruk, süreç çökerse kaybolur. Bildirim tarzındaki birçok webhook için bu kabul edilebilir. Değilse iki seçeneğiniz var.

Teslimattan sonra onaylayın. message handler'ı ve bellek içi kuyruk yerine MQTT.js'in handleMessage(packet, callback) metodunu ezin; PUBACK yalnızca callback'i çağırdığınızda gönderilir. Mesajlar istemci başına tek tek işlenir; bu da sıralamayı korur:

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 ancak şimdi
};

Yönlendirici kapalıyken broker'ın mesajları kuyruğa alması için bunu clean: false (MQTT 3.1.1) veya bir session expiry interval (MQTT 5) ile birleştirin. Bunun için broker'ın kalıcı session'ları tutması gerekir: kendi özel broker instance'ınızı kullanın, çünkü paylaşımlı test endpoint'i çevrimdışı session saklamaz. MQTT kalıcı session'lar yazısına bakın.

Araya kalıcı bir kuyruk koyun. Mesajları Redis Streams'e, SQS'e veya bir veritabanı tablosuna yazın ve HTTP worker'larının bunu tüketmesine izin verin. Daha fazla parça demektir, ama yeniden denemeler restart'lardan sağ çıkar.

Her iki durumda da sonuç at-least-once'tır; alıcı API kopyaları tolere etmelidir. Kopya MQTT mesajlarını yönetme yazısı tekilleştirme desenlerini anlatıyor.

Yatayda ölçekleme

Tek bir yönlendirici çok sayıda webhook'u kaldırır, çünkü işin büyük kısmı HTTP'yi beklemektir. Daha fazlası gerektiğinde $share/webhook/testmqtt/webhook/+/events gibi bir shared subscription ile birden çok instance çalıştırın. Broker mesajları instance'lar arasında dağıtır; biri ölürse onun payı diğerlerine geçer. Her instance'a benzersiz bir client ID verin.

Yeniden denemeleri test edin

İsteklerin %40'ını 503 ile reddeden ve kopyaları loglayan bir alıcı çalıştırın:

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) ? "KOPYA" : "kabul", key, body);
    seen.add(key);
    res.writeHead(204).end();
  });
}).listen(3000, () => console.log("alıcı :3000 portunda"));

Önce node receiver.mjs, sonra node forwarder.mjs başlatın ve bir dizi olay publish edin:

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

Her olay sonunda kabul olarak görünmeli; yönlendirici log'unda da deneme N: HTTP 503 uyarıları olmalıdır. Ardından diğer yolları test edin:

• Alıcının 400 döndürmesini sağlayın: mesajlar yeniden denenmeden hemen dead-letter topic'ine düşmelidir. Bunu mosquitto_sub -t testmqtt/webhook/dlq -v ile veya online MQTT istemcisinde izleyin.
• Publish ederken alıcıyı 30 saniye durdurun: yeniden denemeler aralıklarını açmalı ve alıcı geri geldiğinde başarılı olmalıdır.

Olayları ve dead-letter topic'ini tarayıcıda izleyin

Bir webhook yönlendiricisi birkaç düzine satırdır ve neredeyse hepsi hatalarla ilgilidir. Yanıtları sınıflandırın, jitter'lı backoff uygulayın, idempotency key gönderin, hiçbir mesajı sessizce atmayın ve production trafiğini emanet etmeden önce hata veren alıcı testini çalıştırın. API sizinse, HTTP tarafını yönlendiricinin yanında Express ile kurmayı da düşünün.

Sıkça sorulan sorular