Node.js-Express-MQTT-Bridge: REST zu MQTT mit SSE

Eine REST-zu-MQTT-Bridge erlaubt jedem HTTP-Client – einer Mobile-App, einem Webhook, einem Cronjob, curl –, Befehle an Geräte zu senden, ohne MQTT zu sprechen. Mit Express und MQTT.js 5 ist das ein kleiner Dienst: eine langlebige MQTT-Verbindung, ein POST-Endpunkt, der publiziert, und ein Server-Sent-Events-Endpunkt (SSE), der Geräte-Updates sofort an den Browser schiebt.

Dieses Tutorial baut den Dienst Schritt für Schritt auf, ergänzt einen Request/Response-Endpunkt mit MQTT-5-Correlation-Data und zeigt, wie du jeden Pfad im Terminal testest.

Architektur

Wer mit wem spricht

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

Die wichtigsten Designentscheidungen:

Ein MQTT-Client pro Prozess, beim Start erstellt – niemals einer pro Request.
Pfadparameter validieren, bevor sie in einem Topic landen, damit niemand +, # oder / einschleusen kann.
Fan-out im Speicher: Eine MQTT-Subscription versorgt alle SSE-Clients.
Schnell scheitern mit 503, wenn der Broker nicht erreichbar ist.

Wenn du MQTT.js auffrischen willst, starte mit MQTT in JavaScript. Wann eine Bridge sinnvoller ist als MQTT Ende-zu-Ende, beschreibt MQTT vs HTTP.

Projekt aufsetzen

Installation

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

Express 5 reicht abgelehnte Promises aus async-Handlern an den Error-Handler weiter, du brauchst also keinen Wrapper um jedes await.

Die MQTT-Verbindung

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));

Weil im connect-Handler abonniert wird, werden die Subscriptions nach jedem Reconnect wiederhergestellt, auch mit Clean Session. Eine eindeutige Client-ID pro Prozess verhindert beim Skalieren das klassische Problem "zwei Instanzen werfen sich gegenseitig raus" – der Client-ID-Generator zeigt gute Formate.

Endpunkte: publizieren, Zustand lesen, streamen

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 -> letzter Payload (String)
const streams = new Set();     // offene SSE-Responses
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-Befehl
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. Letzter bekannter Zustand
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. Live-Stream für Browser
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 abschalten
  });
  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'));

Ein paar Details, die in Produktion zählen:

202 Accepted ist ehrlich: QoS 1 heißt, der Broker hat die Nachricht bestätigt – nicht, dass das Gerät sie ausgeführt hat. Brauchst du die Antwort des Geräts, nimm den RPC-Endpunkt weiter unten.
Der Kommentar : ping alle 15 Sekunden verhindert, dass Proxys und Load Balancer inaktive SSE-Verbindungen schließen.
Neue SSE-Clients erhalten sofort den zwischengespeicherten letzten Zustand, Dashboards sind also nie leer. Lass selten publizierende Geräte ihren Zustand als Retained Messages senden, damit auch die Bridge ihn nach einem Neustart hat.
Setze Authentifizierung (API-Key oder Session-Middleware) vor /api, bevor du es veröffentlichst – die Bridge publiziert mit ihren eigenen Broker-Zugangsdaten.

Request/Response mit MQTT 5

Manchmal braucht der HTTP-Aufrufer die Antwort des Geräts. MQTT 5 bringt das mit: Der Request trägt ein Response Topic und Correlation Data, und das Gerät gibt die Correlation Data in seiner Antwort zurück. Ergänze diesen Endpunkt in server.js:

server.js (Fortsetzung)

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(() => {});                      // verhindert eine Unhandled Rejection, falls publish fehlschlägt

  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' });
  }
});

Ein Gerätesimulator zum Testen:

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 },
  });
});

Das Muster und seine Fallstricke behandelt MQTT Request/Response ausführlich.

Die Browser-Seite

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 verbindet sich automatisch neu und nutzt dabei den retry-Wert, den der Server geschickt hat. Ein Haken: Über HTTP/1.1 erlauben Browser nur etwa sechs Verbindungen pro Origin, und jeder offene SSE-Stream belegt eine; liefere die App über HTTP/2 aus, wenn Nutzer viele Tabs öffnen. Einen tieferen Vergleich der Push-Optionen bietet MQTT vs SSE und Long Polling.

Jeden Pfad testen

Starte die Bridge und in separaten Terminals den Stream und einen MQTT-Subscriber:

Terminal 1 — starten

MQTT_USER=username MQTT_PASS=password node server.js

Terminal 2 — SSE-Stream beobachten

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

Terminal 3 — Befehle am Broker beobachten

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

Jetzt jede Richtung durchspielen:

Terminal 4 — testen

# HTTP -> MQTT: 202 erwartet, Nachricht erscheint in Terminal 3
curl -i -X POST http://localhost:3000/api/devices/lamp1/commands \
  -H 'content-type: application/json' -d '{"action":"on"}'

# MQTT -> SSE: ein "event: state"-Frame erscheint in Terminal 2
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}'

# Validierung: 400 erwartet
curl -i -X POST 'http://localhost:3000/api/devices/%23/commands' -d '{}'

# RPC: zuerst device.js starten, {"reply": ...} erwartet; ohne Simulator 504 nach 5 s
curl -s -X POST http://localhost:3000/api/devices/lamp1/rpc \
  -H 'content-type: application/json' -d '{"method":"getStatus"}'

Statt mosquitto_pub kannst du die State-Nachricht auch aus dem Online-MQTT-Client publizieren und zusehen, wie sie in deiner Seite erscheint. Zum Schluss trennst du die Broker-Verbindung (WLAN aus oder Port blockieren) und prüfst, dass POST 503 liefert und die Bridge sich danach selbstständig neu verbindet und neu abonniert.

Test-Zustände aus dem Browser publizieren

Horizontal skalieren

Mehrere Bridge-Instanzen hinter einem Load Balancer funktionieren mit zwei Anpassungen. Jede Instanz braucht eine eigene Client-ID (das Suffix process.pid reicht auf einem Host; in Containern nimm Hostname oder Pod-Namen) und ein eigenes Reply-Topic, das der Code bereits aus der Client-ID ableitet. Soll nur eine Instanz jede eingehende State-Nachricht verarbeiten – etwa um sie in eine Datenbank zu schreiben –, stelle diese Subscription auf eine Shared Subscription um ($share/bridge/testmqtt/express/devices/+/state) und lass das SSE-Fan-out auf einer normalen Subscription, damit jede Instanz ihre eigenen Browser bedienen kann. Für viel Traffic teste gegen deine eigene Broker-Instanz: Geteilte Test-Endpunkte sind ratenbegrenzt und für Funktionstests gedacht, nicht für Durchsatz.

Eine Express-Bridge sind ein paar Dutzend Zeilen Code, aber sie ist das Teil, mit dem der Rest deines Stacks MQTT komplett ignorieren kann. Halte eine Verbindung pro Prozess, validiere alles, was in einem Topic landet, gib ehrliche Statuscodes zurück und teste beide Richtungen mit curl und einem Subscriber, bevor jemand darauf aufbaut. Wenn du denselben Client-Code in einer anderen Sprache brauchst, liefert der MQTT-Code-Generator einen Ausgangspunkt.

Häufig gestellte Fragen

Sollte ich pro HTTP-Request eine neue MQTT-Verbindung aufbauen?

Nein. Öffne beim Start des Express-Prozesses einen langlebigen MQTT-Client und nutze ihn für alle Requests. Eine Verbindung pro Request kostet bei jedem Aufruf einen TCP/TLS-Handshake plus CONNECT-Roundtrip und schöpft die Verbindungslimits des Brokers schnell aus.

Warum Server-Sent Events statt MQTT über WebSockets im Browser?

SSE hält die Broker-Zugangsdaten auf dem Server, läuft als normales HTTP durch die meisten Proxys und braucht keine Client-Bibliothek. MQTT über WebSockets ist besser, wenn Browser häufig publizieren oder viele dynamische Topics abonnieren müssen.

Was sollte der REST-Endpunkt zurückgeben, wenn der Broker nicht erreichbar ist?

Schnell mit 503 Service Unavailable scheitern, statt MQTT.js die Nachricht still in die Warteschlange legen zu lassen. So weiß der Aufrufer, dass der Befehl nicht gesendet wurde, und kann es erneut versuchen.

MQTT mit Node.js und Express: eine REST-zu-MQTT-Bridge bauen