Passerelle MQTT Node.js Express : REST vers MQTT et SSE

Une passerelle REST vers MQTT permet à n'importe quel client HTTP — une app mobile, un webhook, une tâche cron, curl — d'envoyer des commandes à des appareils sans parler MQTT. Avec Express et MQTT.js 5, c'est un petit service : une connexion MQTT longue durée, un endpoint POST qui publie et un endpoint Server-Sent Events (SSE) qui pousse au navigateur les mises à jour des appareils dès leur arrivée.

Ce tutoriel construit ce service pas à pas, ajoute un endpoint requête/réponse basé sur la correlation data de MQTT 5 et montre comment tester chaque chemin depuis le terminal.

Architecture

Qui parle à qui

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

Les choix de conception clés :

Un client MQTT par processus, créé au démarrage — jamais un par requête.
Validez les paramètres de route avant de les insérer dans un topic, pour que personne ne puisse injecter +, # ou /.
Diffusez en mémoire : un seul abonnement MQTT alimente tous les clients SSE.
Échouez vite avec 503 quand le broker est injoignable.

Pour revoir les bases de MQTT.js, commencez par MQTT en JavaScript. Pour savoir quand une passerelle est plus pertinente que MQTT de bout en bout, voir MQTT vs HTTP.

Mise en place du projet

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 transmet les promesses rejetées des handlers async au gestionnaire d'erreurs : inutile d'envelopper chaque await.

La connexion MQTT

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

S'abonner dans le handler connect garantit que les abonnements sont rétablis après chaque reconnexion, même en clean session. Un client ID unique par processus évite le problème classique des « deux instances qui s'éjectent mutuellement » lors d'une montée en charge ; le générateur de client ID propose de bons formats.

Endpoints : publier, lire l'état, diffuser

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 -> dernier payload (string)
const streams = new Set();     // réponses SSE ouvertes
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. Commande "fire-and-forget"
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. Dernier état connu
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. Flux en direct pour les navigateurs
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',              // désactive le buffering de nginx
  });
  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'));

Quelques détails qui comptent en production :

202 Accepted est honnête : QoS 1 signifie que le broker a acquitté le message, pas que l'appareil l'a exécuté. Si vous avez besoin de la réponse de l'appareil, utilisez l'endpoint RPC ci-dessous.
Le commentaire : ping toutes les 15 secondes empêche proxys et load balancers de fermer les connexions SSE inactives.
Les nouveaux clients SSE reçoivent immédiatement le dernier état en cache : les dashboards ne sont jamais vides. Pour les appareils qui publient rarement, faites-leur publier leur état en messages retenus afin que la passerelle le retrouve aussi après un redémarrage.
Placez une authentification (clé d'API ou middleware de session) devant /api avant de l'exposer : la passerelle publie avec ses propres identifiants de broker.

Requête/réponse avec MQTT 5

Parfois l'appelant HTTP a besoin de la réponse de l'appareil. MQTT 5 le prévoit nativement : la requête transporte un response topic et une correlation data, et l'appareil renvoie la correlation data dans sa réponse. Ajoutez cet endpoint à server.js :

server.js (suite)

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(() => {});                      // évite un unhandled rejection si le publish échoue

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

Un simulateur d'appareil pour le tester :

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

Le modèle et ses pièges sont détaillés dans requête/réponse avec MQTT.

Côté navigateur

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 se reconnecte automatiquement en utilisant la valeur retry envoyée par le serveur. Une réserve : en HTTP/1.1, les navigateurs n'autorisent qu'environ six connexions par origine, et chaque flux SSE ouvert en consomme une ; servez l'application en HTTP/2 si vos utilisateurs ouvrent beaucoup d'onglets. Pour une comparaison approfondie des options de push, voir MQTT vs SSE et long polling.

Tester chaque chemin

Lancez la passerelle puis, dans des terminaux séparés, le flux et un abonné MQTT :

Terminal 1 — lancer

MQTT_USER=username MQTT_PASS=password node server.js

Terminal 2 — observer le flux SSE

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

Terminal 3 — observer les commandes arrivant au broker

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

Testez maintenant chaque sens :

Terminal 4 — piloter

# HTTP -> MQTT : attendu 202 et le message dans le terminal 3
curl -i -X POST http://localhost:3000/api/devices/lamp1/commands \
  -H 'content-type: application/json' -d '{"action":"on"}'

# MQTT -> SSE : attendu une trame "event: state" dans le 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}'

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

# RPC : lancez d'abord device.js, attendu {"reply": ...} ; sans lui, 504 après 5 s
curl -s -X POST http://localhost:3000/api/devices/lamp1/rpc \
  -H 'content-type: application/json' -d '{"method":"getStatus"}'

Au lieu de mosquitto_pub, vous pouvez publier le message d'état depuis le client MQTT en ligne et le voir apparaître dans votre page. Enfin, coupez la connexion au broker (désactivez le Wi-Fi ou bloquez le port) et vérifiez que POST renvoie 503, puis que la passerelle se reconnecte et se réabonne d'elle-même.

Publier des états de test depuis le navigateur

Montée en charge horizontale

Faire tourner plusieurs instances derrière un load balancer fonctionne moyennant deux ajustements. Chaque instance a besoin de son propre client ID (le suffixe process.pid suffit sur une seule machine ; en conteneur, utilisez le hostname ou le nom du pod) et de son propre topic de réponse, que le code dérive déjà du client ID. Si une seule instance doit traiter chaque message d'état entrant — par exemple pour l'écrire en base —, passez cet abonnement en shared subscription ($share/bridge/testmqtt/express/devices/+/state) et gardez la diffusion SSE sur un abonnement normal, afin que chaque instance serve ses propres navigateurs. Pour un trafic important, testez contre votre propre instance de broker : les endpoints de test partagés sont limités en débit et servent aux vérifications fonctionnelles, pas aux mesures de throughput.

Une passerelle Express tient en quelques dizaines de lignes, mais c'est la pièce qui permet au reste de votre stack d'ignorer complètement MQTT. Gardez une connexion par processus, validez tout ce qui finit dans un topic, renvoyez des codes de statut honnêtes et testez les deux sens avec curl et un abonné avant que quiconque ne bâtisse dessus. Pour obtenir le même code client dans un autre langage, le générateur de code MQTT offre un point de départ.

Questions fréquentes

Faut-il ouvrir une nouvelle connexion MQTT à chaque requête HTTP ?

Non. Ouvrez un seul client MQTT longue durée au démarrage du processus Express et réutilisez-le pour toutes les requêtes. Se connecter à chaque requête ajoute une poignée de main TCP/TLS et un aller-retour CONNECT à chaque appel et épuise vite les limites de connexions du broker.

Pourquoi utiliser Server-Sent Events plutôt que MQTT sur WebSockets dans le navigateur ?

SSE garde les identifiants du broker côté serveur, traverse la plupart des proxys comme du HTTP classique et ne demande aucune bibliothèque cliente. MQTT sur WebSockets est préférable quand le navigateur doit publier souvent ou s'abonner à de nombreux topics dynamiques.

Que doit renvoyer l'endpoint REST quand le broker est indisponible ?

Échouez vite avec 503 Service Unavailable au lieu de laisser MQTT.js mettre le message en file d'attente en silence. L'appelant sait alors que la commande n'est pas partie et peut réessayer.

MQTT avec Node.js et Express : créer une passerelle REST vers MQTT