Puente MQTT con Node.js y Express: REST a MQTT y SSE

Un puente REST a MQTT permite que cualquier cliente HTTP —una app móvil, un webhook, un cron, curl— envíe comandos a dispositivos sin hablar MQTT. Con Express y MQTT.js 5 es un servicio pequeño: una conexión MQTT de larga duración, un endpoint POST que publica y un endpoint Server-Sent Events (SSE) que empuja al navegador las actualizaciones de los dispositivos en cuanto llegan.

Este tutorial construye ese servicio paso a paso, añade un endpoint de petición/respuesta con correlation data de MQTT 5 y muestra cómo probar cada camino desde la terminal.

Arquitectura

Quién habla con quién

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

Las decisiones de diseño clave:

Un cliente MQTT por proceso, creado al arrancar; nunca uno por petición.
Valida los parámetros de ruta antes de meterlos en un topic, para que nadie pueda inyectar +, # o /.
Reparte en memoria: una sola suscripción MQTT alimenta a todos los clientes SSE.
Falla rápido con 503 cuando el broker no está accesible.

Si quieres repasar MQTT.js, empieza por MQTT en JavaScript. Para saber cuándo tiene más sentido un puente que hablar MQTT de extremo a extremo, consulta MQTT vs HTTP.

Preparar el proyecto

Instalación

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

Express 5 pasa las promesas rechazadas de los handlers async al manejador de errores, así que no necesitas envolver cada await.

La conexión 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));

Suscribirse dentro del handler connect hace que las suscripciones se restauren tras cada reconexión, incluso con clean session. Un client ID único por proceso evita el clásico problema de "dos instancias que se expulsan mutuamente" al escalar; el generador de client ID muestra buenos formatos.

Endpoints: publicar, leer estado, stream

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 -> último payload (string)
const streams = new Set();     // respuestas SSE abiertas
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. Comando "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. Último estado conocido
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. Stream en directo para navegadores
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',              // desactiva el 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'));

Algunos detalles que importan en producción:

202 Accepted es honesto: QoS 1 significa que el broker confirmó el mensaje, no que el dispositivo lo ejecutara. Si necesitas la respuesta del dispositivo, usa el endpoint RPC de más abajo.
El comentario : ping cada 15 segundos evita que proxies y balanceadores cierren conexiones SSE inactivas.
Los nuevos clientes SSE reciben al instante el último estado en caché, así que los dashboards nunca están vacíos. Para dispositivos que publican poco, haz que publiquen el estado como mensajes retenidos para que el puente también lo tenga tras reiniciarse.
Pon autenticación (una API key o un middleware de sesión) delante de /api antes de exponerlo: el puente publica con sus propias credenciales de broker.

Petición/respuesta con MQTT 5

A veces quien llama por HTTP necesita la respuesta del dispositivo. MQTT 5 lo trae de serie: la petición lleva un response topic y correlation data, y el dispositivo devuelve la correlation data en su respuesta. Añade este endpoint a server.js:

server.js (continuación)

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(() => {});                      // evita un unhandled rejection si falla el publish

  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 simulador de dispositivo para probarlo:

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

El patrón y sus trampas se explican a fondo en petición/respuesta con MQTT.

El lado del navegador

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 reconecta automáticamente usando el valor retry que envió el servidor. Una advertencia: con HTTP/1.1 los navegadores solo permiten unas seis conexiones por origen, y cada stream SSE abierto ocupa una; sirve la app por HTTP/2 si los usuarios abren muchas pestañas. Para una comparación más profunda de las opciones push, consulta MQTT vs SSE y long polling.

Prueba cada camino

Arranca el puente y, en terminales separadas, el stream y un suscriptor MQTT:

Terminal 1 — ejecutar

MQTT_USER=username MQTT_PASS=password node server.js

Terminal 2 — ver el stream SSE

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

Terminal 3 — ver los comandos que llegan al 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

Ahora prueba cada dirección:

Terminal 4 — probar

# HTTP -> MQTT: espera 202 y el mensaje en la terminal 3
curl -i -X POST http://localhost:3000/api/devices/lamp1/commands \
  -H 'content-type: application/json' -d '{"action":"on"}'

# MQTT -> SSE: espera una trama "event: state" en la 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}'

# Validación: espera 400
curl -i -X POST 'http://localhost:3000/api/devices/%23/commands' -d '{}'

# RPC: ejecuta antes device.js y espera {"reply": ...}; sin él, 504 tras 5 s
curl -s -X POST http://localhost:3000/api/devices/lamp1/rpc \
  -H 'content-type: application/json' -d '{"method":"getStatus"}'

En lugar de mosquitto_pub, puedes publicar el mensaje de estado desde el cliente MQTT online y ver cómo aparece en tu página. Por último, corta la conexión con el broker (desconecta el Wi-Fi o bloquea el puerto) y comprueba que POST devuelve 503 y que después el puente se reconecta y se vuelve a suscribir solo.

Publicar estados de prueba desde el navegador

Escalar horizontalmente

Ejecutar varias instancias del puente detrás de un balanceador funciona con dos ajustes. Cada instancia necesita su propio client ID (el sufijo process.pid vale en un solo host; en contenedores usa el hostname o el nombre del pod) y su propio topic de respuesta, que el código ya deriva del client ID. Si quieres que solo una instancia procese cada mensaje de estado entrante —por ejemplo, para guardarlo en una base de datos—, cambia esa suscripción a una shared subscription ($share/bridge/testmqtt/express/devices/+/state) y mantén el reparto SSE en una suscripción normal para que cada instancia sirva a sus propios navegadores. Para tráfico intenso, prueba contra tu propia instancia de broker: los endpoints de prueba compartidos tienen límite de tasa y sirven para comprobaciones funcionales, no de rendimiento.

Un puente Express son unas pocas docenas de líneas, pero es la pieza que permite que el resto de tu stack ignore MQTT por completo. Mantén una conexión por proceso, valida todo lo que acabe en un topic, devuelve códigos de estado honestos y prueba ambas direcciones con curl y un suscriptor antes de que nadie construya encima. Si quieres el mismo código cliente en otro lenguaje, el generador de código MQTT te da un punto de partida.

Preguntas frecuentes

¿Debo crear una conexión MQTT nueva por cada petición HTTP?

No. Abre un único cliente MQTT de larga duración al arrancar el proceso Express y reutilízalo en todas las peticiones. Conectar por petición añade un handshake TCP/TLS y un CONNECT a cada llamada y agota rápido los límites de conexiones del broker.

¿Por qué usar Server-Sent Events en lugar de MQTT sobre WebSockets en el navegador?

SSE mantiene las credenciales del broker en el servidor, atraviesa la mayoría de proxies como HTTP normal y no necesita librería cliente. MQTT sobre WebSockets es mejor cuando el navegador debe publicar a menudo o suscribirse a muchos topics dinámicos.

¿Qué debe devolver el endpoint REST si el broker está caído?

Falla rápido con 503 Service Unavailable en lugar de dejar que MQTT.js encole el mensaje en silencio. Así quien llama sabe que el comando no se envió y puede reintentar.

MQTT con Node.js y Express: crea un puente REST a MQTT