Ponte MQTT com Node.js e Express: REST para MQTT e SSE

Uma ponte REST para MQTT permite que qualquer cliente HTTP — um app mobile, um webhook, um cron job, curl — envie comandos a dispositivos sem falar MQTT. Com Express e MQTT.js 5 é um serviço pequeno: uma conexão MQTT de longa duração, um endpoint POST que publica e um endpoint Server-Sent Events (SSE) que empurra para o navegador as atualizações dos dispositivos assim que chegam.

Este tutorial constrói esse serviço passo a passo, adiciona um endpoint de request/response com correlation data do MQTT 5 e mostra como testar cada caminho pelo terminal.

Arquitetura

Quem fala com quem

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

As principais decisões de design:

Um cliente MQTT por processo, criado na inicialização — nunca um por requisição.
Valide os parâmetros de rota antes de colocá-los em um topic, para que ninguém consiga injetar +, # ou /.
Distribua em memória: uma única assinatura MQTT alimenta todos os clientes SSE.
Falhe rápido com 503 quando o broker estiver inacessível.

Se quiser relembrar o MQTT.js, comece por MQTT em JavaScript. Para saber quando uma ponte faz mais sentido do que falar MQTT de ponta a ponta, veja MQTT vs HTTP.

Preparando o projeto

Instalação

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

O Express 5 encaminha promises rejeitadas de handlers async para o tratador de erros, então você não precisa envolver cada await.

A conexão 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));

Assinar dentro do handler connect faz as assinaturas serem restauradas a cada reconexão, mesmo com clean session. Um client ID único por processo evita o clássico problema de "duas instâncias se derrubando" ao escalar; o gerador de client ID mostra bons formatos.

Endpoints: publicar, ler 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();     // respostas SSE abertas
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 conhecido
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 ao vivo 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',              // desativa o buffering do 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'));

Alguns detalhes que importam em produção:

202 Accepted é honesto: QoS 1 significa que o broker confirmou a mensagem, não que o dispositivo a executou. Se você precisa da resposta do dispositivo, use o endpoint RPC abaixo.
O comentário : ping a cada 15 segundos impede que proxies e load balancers fechem conexões SSE ociosas.
Novos clientes SSE recebem na hora o último estado em cache, então os dashboards nunca ficam vazios. Para dispositivos que publicam raramente, faça-os publicar o estado como mensagens retidas para que a ponte também o tenha após reiniciar.
Coloque autenticação (uma API key ou middleware de sessão) na frente de /api antes de expô-la: a ponte publica com as próprias credenciais de broker.

Request/response com MQTT 5

Às vezes quem chama via HTTP precisa da resposta do dispositivo. O MQTT 5 traz isso nativamente: a requisição leva um response topic e correlation data, e o dispositivo devolve a correlation data na resposta. Adicione este endpoint ao server.js:

server.js (continuação)

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 unhandled rejection se o publish falhar

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

Um simulador de dispositivo para testar:

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

O padrão e suas armadilhas são explicados em detalhe em request/response com MQTT.

O lado do 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>

O EventSource reconecta automaticamente usando o valor retry enviado pelo servidor. Um cuidado: em HTTP/1.1 os navegadores permitem só cerca de seis conexões por origem, e cada stream SSE aberto usa uma; sirva o app via HTTP/2 se os usuários abrirem muitas abas. Para uma comparação mais profunda das opções de push, veja MQTT vs SSE e long polling.

Teste cada caminho

Inicie a ponte e, em terminais separados, o stream e um assinante MQTT:

Terminal 1 — executar

MQTT_USER=username MQTT_PASS=password node server.js

Terminal 2 — ver o stream SSE

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

Terminal 3 — ver os comandos chegando ao 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

Agora exercite cada direção:

Terminal 4 — testar

# HTTP -> MQTT: espere 202 e a mensagem no terminal 3
curl -i -X POST http://localhost:3000/api/devices/lamp1/commands \
  -H 'content-type: application/json' -d '{"action":"on"}'

# MQTT -> SSE: espere um frame "event: state" no 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}'

# Validação: espere 400
curl -i -X POST 'http://localhost:3000/api/devices/%23/commands' -d '{}'

# RPC: rode device.js antes e espere {"reply": ...}; sem ele, 504 após 5 s
curl -s -X POST http://localhost:3000/api/devices/lamp1/rpc \
  -H 'content-type: application/json' -d '{"method":"getStatus"}'

Em vez do mosquitto_pub, você pode publicar a mensagem de estado pelo cliente MQTT online e vê-la aparecer na sua página. Por fim, corte a conexão com o broker (desligue o Wi-Fi ou bloqueie a porta) e confira que o POST retorna 503 e que depois a ponte reconecta e assina novamente sozinha.

Publicar estados de teste pelo navegador

Escalando horizontalmente

Rodar várias instâncias da ponte atrás de um load balancer funciona com dois ajustes. Cada instância precisa do próprio client ID (o sufixo process.pid serve em um único host; em containers use o hostname ou o nome do pod) e do próprio topic de resposta, que o código já deriva do client ID. Se você quer que só uma instância processe cada mensagem de estado recebida — por exemplo, para gravá-la em um banco de dados —, troque essa assinatura por uma shared subscription ($share/bridge/testmqtt/express/devices/+/state) e mantenha a distribuição SSE em uma assinatura normal, para que cada instância atenda aos próprios navegadores. Para tráfego pesado, teste contra a sua própria instância de broker: endpoints de teste compartilhados têm limite de taxa e servem para verificações funcionais, não de throughput.

Uma ponte Express tem poucas dezenas de linhas, mas é a peça que permite ao resto do seu stack ignorar o MQTT por completo. Mantenha uma conexão por processo, valide tudo que vai parar em um topic, retorne códigos de status honestos e teste as duas direções com curl e um assinante antes que alguém construa em cima. Se quiser o mesmo código cliente em outra linguagem, o gerador de código MQTT dá um ponto de partida.

Perguntas frequentes

Devo criar uma nova conexão MQTT a cada requisição HTTP?

Não. Abra um único cliente MQTT de longa duração quando o processo Express inicia e reutilize-o em todas as requisições. Conectar por requisição adiciona um handshake TCP/TLS e um CONNECT a cada chamada e esgota rapidamente os limites de conexão do broker.

Por que usar Server-Sent Events em vez de MQTT sobre WebSockets no navegador?

O SSE mantém as credenciais do broker no servidor, passa pela maioria dos proxies como HTTP comum e não precisa de biblioteca cliente. MQTT sobre WebSockets é melhor quando o navegador precisa publicar com frequência ou assinar muitos topics dinâmicos.

O que o endpoint REST deve retornar quando o broker está fora do ar?

Falhe rápido com 503 Service Unavailable em vez de deixar o MQTT.js enfileirar a mensagem em silêncio. Assim quem chamou sabe que o comando não foi enviado e pode tentar de novo.

MQTT com Node.js e Express: crie uma ponte REST para MQTT