Node.js Express MQTT Köprüsü: REST'ten MQTT'ye ve SSE

Bir REST'ten MQTT'ye köprü, herhangi bir HTTP istemcisinin — mobil uygulama, webhook, cron job, curl — MQTT konuşmadan cihazlara komut göndermesini sağlar. Express ve MQTT.js 5 ile bu küçük bir servistir: uzun ömürlü tek bir MQTT bağlantısı, yayın yapan bir POST endpoint'i ve cihaz güncellemelerini geldikleri anda tarayıcıya ileten bir Server-Sent Events (SSE) endpoint'i.

Bu eğitim servisi adım adım kuruyor, MQTT 5 correlation data ile bir istek/yanıt endpoint'i ekliyor ve her yolun terminalden nasıl test edileceğini gösteriyor.

Mimari

Kim kiminle konuşuyor

HTTP client ──POST /api/devices/:id/commands──▶ Express ──publish──▶ testmqtt/express/devices/<id>/cmd ──▶ device
Browser     ◀──── GET /api/stream (SSE) ─────── Express ◀─subscribe── testmqtt/express/devices/+/state ◀── device

Temel tasarım kararları:

Süreç başına bir MQTT istemcisi, açılışta oluşturulur; asla istek başına değil.
Path parametrelerini doğrulayın: bir topic'e koymadan önce kontrol edin ki kimse +, # veya / enjekte edemesin.
Bellekte dağıtın: tek bir MQTT aboneliği tüm SSE istemcilerini besler.
Hızlı hata verin: broker'a ulaşılamadığında 503 döndürün.

MQTT.js'i hatırlamak isterseniz JavaScript ile MQTT yazısıyla başlayın. Uçtan uca MQTT yerine bir köprünün ne zaman daha mantıklı olduğunu MQTT vs HTTP anlatıyor.

Proje kurulumu

Kurulum

mkdir mqtt-rest-bridge && cd mqtt-rest-bridge
npm init -y && npm pkg set type=module
npm install express@5 mqtt@5

Express 5, async handler'lardan reddedilen promise'leri hata işleyicisine iletir; bu yüzden her await'i sarmalamanız gerekmez.

MQTT bağlantısı

mqtt.js

import mqtt from 'mqtt';

export const PREFIX = 'testmqtt/express';
export const CLIENT_ID = `rest-bridge-${process.pid}`;
export const REPLY_TOPIC = `${PREFIX}/replies/${CLIENT_ID}`;

export const client = mqtt.connect('mqtts://broker.connectmq.com:8883', {
  clientId: CLIENT_ID,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
  reconnectPeriod: 2000,
  connectTimeout: 10_000,
});

client.on('connect', async () => {
  console.log('MQTT connected');
  try {
    await client.subscribeAsync([`${PREFIX}/devices/+/state`, REPLY_TOPIC], { qos: 1 });
  } catch (err) {
    console.error('subscribe failed:', err.message);
  }
});
client.on('reconnect', () => console.warn('MQTT reconnecting…'));
client.on('error', (err) => console.error('MQTT error:', err.message));

connect handler'ı içinde abone olmak, clean session olsa bile aboneliklerin her yeniden bağlantıdan sonra geri yüklenmesini sağlar. Süreç başına benzersiz bir client ID, ölçeklendirirken "iki instance'ın birbirini sürekli düşürmesi" sorununu önler; iyi formatlar için client ID oluşturucu aracına bakın.

Endpoint'ler: yayınlama, durum okuma, akış

server.js

import express from 'express';
import { randomUUID } from 'node:crypto';
import { client, PREFIX, REPLY_TOPIC } from './mqtt.js';

const ID_RE = /^[A-Za-z0-9_-]{1,64}$/;
const lastState = new Map();   // deviceId -> son payload (string)
const streams = new Set();     // açık SSE yanıtları
const pending = new Map();     // correlationId -> { resolve, timer }

const sse = (event, data) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;

client.on('message', (topic, payload, packet) => {
  if (topic === REPLY_TOPIC) {
    const id = packet.properties?.correlationData?.toString();
    const waiter = pending.get(id);
    if (waiter) {
      clearTimeout(waiter.timer);
      pending.delete(id);
      waiter.resolve(payload.toString());
    }
    return;
  }
  const deviceId = topic.split('/')[3];          // testmqtt/express/devices/<id>/state
  const text = payload.toString();
  lastState.set(deviceId, text);
  const frame = sse('state', { deviceId, payload: text, ts: Date.now() });
  for (const res of streams) res.write(frame);
});

const app = express();
app.use(express.json({ limit: '16kb' }));

app.param('id', (req, res, next, id) =>
  ID_RE.test(id) ? next() : res.status(400).json({ error: 'invalid device id' }));

// 1. "Fire-and-forget" komut
app.post('/api/devices/:id/commands', async (req, res) => {
  if (!client.connected) return res.status(503).json({ error: 'broker unavailable' });
  const topic = `${PREFIX}/devices/${req.params.id}/cmd`;
  await client.publishAsync(topic, JSON.stringify(req.body ?? {}), { qos: 1 });
  res.status(202).json({ topic, accepted: true });
});

// 2. Bilinen son durum
app.get('/api/devices/:id/state', (req, res) => {
  const payload = lastState.get(req.params.id);
  if (payload === undefined) return res.status(404).json({ error: 'no state received yet' });
  res.json({ deviceId: req.params.id, payload });
});

// 3. Tarayıcılar için canlı akış
app.get('/api/stream', (req, res) => {
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
    'X-Accel-Buffering': 'no',              // nginx buffering'i kapatır
  });
  res.write('retry: 3000\n\n');
  for (const [deviceId, payload] of lastState) res.write(sse('state', { deviceId, payload }));
  streams.add(res);
  req.on('close', () => streams.delete(res));
});

setInterval(() => { for (const res of streams) res.write(': ping\n\n'); }, 15_000);

app.use(express.static('public'));
app.listen(3000, () => console.log('listening on http://localhost:3000'));

Production'da önemli olan birkaç ayrıntı:

202 Accepted dürüst bir yanıttır: QoS 1, cihazın mesajı çalıştırdığını değil, broker'ın onayladığını gösterir. Cihazın yanıtına ihtiyacınız varsa aşağıdaki RPC endpoint'ini kullanın.
Her 15 saniyede bir gönderilen : ping yorumu, proxy'lerin ve load balancer'ların boştaki SSE bağlantılarını kapatmasını önler.
Yeni SSE istemcileri önbellekteki son durumu anında alır; böylece dashboard'lar hiçbir zaman boş kalmaz. Seyrek yayın yapan cihazların durumlarını retained mesaj olarak yayınlamasını sağlayın ki köprü yeniden başladıktan sonra da bu bilgiye sahip olsun.
/api'yi dışarı açmadan önce önüne kimlik doğrulama (API anahtarı veya oturum middleware'i) koyun; köprü kendi broker kimlik bilgileriyle yayın yapar.

MQTT 5 ile istek/yanıt

Bazen HTTP üzerinden çağıran tarafın cihazın yanıtına ihtiyacı olur. MQTT 5 bunu yerleşik olarak sunar: istek bir response topic ve correlation data taşır, cihaz da yanıtında correlation data'yı geri gönderir. Bu endpoint'i server.js dosyasına ekleyin:

server.js (devamı)

app.post('/api/devices/:id/rpc', async (req, res) => {
  if (!client.connected) return res.status(503).json({ error: 'broker unavailable' });
  const correlationId = randomUUID();
  const reply = new Promise((resolve, reject) => {
    const timer = setTimeout(() => { pending.delete(correlationId); reject(new Error('timeout')); }, 5000);
    pending.set(correlationId, { resolve, timer });
  });
  reply.catch(() => {});                      // publish başarısız olursa unhandled rejection'ı önler

  await client.publishAsync(`${PREFIX}/devices/${req.params.id}/rpc`, JSON.stringify(req.body ?? {}), {
    qos: 1,
    properties: { responseTopic: REPLY_TOPIC, correlationData: Buffer.from(correlationId) },
  });

  try {
    res.json({ deviceId: req.params.id, reply: await reply });
  } catch {
    res.status(504).json({ error: 'device did not answer within 5 s' });
  }
});

Test için bir cihaz simülatörü:

device.js

import mqtt from 'mqtt';

const c = mqtt.connect('mqtts://broker.connectmq.com:8883', {
  username: process.env.MQTT_USER, password: process.env.MQTT_PASS, protocolVersion: 5,
});
c.on('connect', () => c.subscribe('testmqtt/express/devices/lamp1/rpc', { qos: 1 }));
c.on('message', (topic, payload, packet) => {
  const { responseTopic, correlationData } = packet.properties ?? {};
  if (!responseTopic) return;
  c.publish(responseTopic, JSON.stringify({ ok: true, echo: JSON.parse(payload) }), {
    qos: 1, properties: { correlationData },
  });
});

Desen ve tuzakları MQTT ile istek/yanıt yazısında ayrıntılı olarak ele alınıyor.

Tarayıcı tarafı

public/index.html

<!doctype html>
<ul id="log"></ul>
<script>
  const log = document.getElementById('log');
  const es = new EventSource('/api/stream');
  es.addEventListener('state', (e) => {
    const { deviceId, payload } = JSON.parse(e.data);
    const li = document.createElement('li');
    li.textContent = `${deviceId}: ${payload}`;
    log.prepend(li);
  });
  es.onerror = () => console.warn('stream interrupted, browser will retry');
</script>

EventSource, sunucunun gönderdiği retry değerini kullanarak otomatik olarak yeniden bağlanır. Bir uyarı: HTTP/1.1 üzerinde tarayıcılar origin başına yalnızca altı kadar bağlantıya izin verir ve açık her SSE akışı bunlardan birini kullanır; kullanıcılar çok sayıda sekme açıyorsa uygulamayı HTTP/2 üzerinden sunun. Push seçeneklerinin daha derin bir karşılaştırması için bkz. MQTT vs SSE ve long polling.

Her yolu test edin

Köprüyü başlatın; ayrı terminallerde akışı ve bir MQTT subscriber'ı açın:

Terminal 1 — çalıştır

MQTT_USER=username MQTT_PASS=password node server.js

Terminal 2 — SSE akışını izle

curl -N http://localhost:3000/api/stream

Terminal 3 — broker'a ulaşan komutları izle

mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -t 'testmqtt/express/devices/+/cmd' -v

Şimdi her yönü deneyin:

Terminal 4 — test et

# HTTP -> MQTT: 202 ve terminal 3'te mesaj beklenir
curl -i -X POST http://localhost:3000/api/devices/lamp1/commands \
  -H 'content-type: application/json' -d '{"action":"on"}'

# MQTT -> SSE: terminal 2'de bir "event: state" çerçevesi beklenir
mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -t 'testmqtt/express/devices/lamp1/state' -m '{"on":true}'

# Doğrulama: 400 beklenir
curl -i -X POST 'http://localhost:3000/api/devices/%23/commands' -d '{}'

# RPC: önce device.js'i çalıştırın, {"reply": ...} beklenir; o olmadan 5 sn sonra 504
curl -s -X POST http://localhost:3000/api/devices/lamp1/rpc \
  -H 'content-type: application/json' -d '{"method":"getStatus"}'

mosquitto_pub yerine durum mesajını online MQTT istemcisinden yayınlayıp sayfanızda belirdiğini görebilirsiniz. Son olarak broker bağlantısını kesin (Wi-Fi'ı kapatın veya portu engelleyin) ve POST'un 503 döndürdüğünü, ardından köprünün kendiliğinden yeniden bağlanıp yeniden abone olduğunu kontrol edin.

Tarayıcıdan test durumları yayınlayın

Yatay ölçeklendirme

Bir load balancer arkasında birden fazla köprü instance'ı çalıştırmak iki ayarla mümkündür. Her instance'ın kendi client ID'sine (tek bir host'ta process.pid soneki yeterlidir; container'larda hostname veya pod adını kullanın) ve kodun zaten client ID'den türettiği kendi yanıt topic'ine ihtiyacı vardır. Gelen her durum mesajını yalnızca bir instance'ın işlemesini istiyorsanız — örneğin veritabanına yazmak için — o aboneliği bir shared subscription'a ($share/bridge/testmqtt/express/devices/+/state) çevirin; SSE dağıtımını ise normal bir abonelikte tutun ki her instance kendi tarayıcılarına hizmet verebilsin. Yoğun trafik için kendi broker instance'ınıza karşı test edin: paylaşımlı test endpoint'leri hız sınırlıdır ve throughput için değil, işlevsel kontroller için tasarlanmıştır.

Bir Express köprüsü birkaç düzine satırlık koddur, ama stack'inizin geri kalanının MQTT'yi tamamen görmezden gelmesini sağlayan parçadır. Süreç başına tek bağlantı tutun, bir topic'e giren her şeyi doğrulayın, dürüst durum kodları döndürün ve kimse üzerine bir şey inşa etmeden önce iki yönü de curl ve bir subscriber ile test edin. Aynı istemci kodunu başka bir dilde istiyorsanız MQTT kod oluşturucu iyi bir başlangıç noktası sunar.

Sıkça sorulan sorular

Her HTTP isteği için yeni bir MQTT bağlantısı açmalı mıyım?

Hayır. Express süreci başlarken uzun ömürlü tek bir MQTT istemcisi açın ve tüm isteklerde onu yeniden kullanın. İstek başına bağlanmak her çağrıya bir TCP/TLS el sıkışması ve CONNECT gidiş-dönüşü ekler ve broker'ın bağlantı limitlerini hızla tüketir.

Tarayıcıda neden MQTT over WebSockets yerine Server-Sent Events kullanmalı?

SSE broker kimlik bilgilerini sunucuda tutar, sıradan HTTP olarak çoğu proxy'den geçer ve istemci kütüphanesi gerektirmez. Tarayıcının sık yayın yapması veya çok sayıda dinamik topic'e abone olması gerekiyorsa MQTT over WebSockets daha iyi bir seçimdir.

Broker çalışmadığında REST endpoint'i ne döndürmeli?

MQTT.js'in mesajı sessizce kuyruğa almasına izin vermek yerine 503 Service Unavailable ile hızlıca hata verin. Böylece çağıran taraf komutun gönderilmediğini bilir ve yeniden deneyebilir.

Node.js ve Express ile MQTT: REST'ten MQTT'ye Köprü Kurun