Node.js Express MQTT Bridge: REST to MQTT with SSE

A REST to MQTT bridge lets any HTTP client — a mobile app, a webhook, a cron job, curl — send commands to devices without speaking MQTT. With Express and MQTT.js 5 it is a small service: one long-lived MQTT connection, a POST endpoint that publishes, and a Server-Sent Events (SSE) endpoint that pushes device updates to the browser as they arrive.

This tutorial builds that service step by step, adds a request/response endpoint using MQTT 5 correlation data, and shows how to test every path from the terminal.

Architecture

Who talks to whom

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

The key design decisions:

One MQTT client per process, created at startup — never one per request.
Validate path parameters before putting them in a topic, so nobody can inject +, # or /.
Fan out in memory: one MQTT subscription feeds all SSE clients.
Fail fast with 503 when the broker is unreachable.

If you want a refresher on MQTT.js itself, start with MQTT in JavaScript. For when a bridge makes more sense than talking MQTT end to end, see MQTT vs HTTP.

Project setup

Install

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

Express 5 forwards rejected promises from async handlers to the error handler, so you do not need a wrapper around every await.

The MQTT connection

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

Subscribing inside the connect handler means subscriptions are restored after every reconnect, even with a clean session. A unique client ID per process avoids the classic "two instances keep kicking each other off" problem when you scale out — the client ID generator shows good formats.

Endpoints: publish, read state, 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 -> last payload (string)
const streams = new Set();     // open 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 command
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. Last known state
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 for browsers
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',              // disable nginx buffering
  });
  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'));

A few details that matter in production:

202 Accepted is honest: QoS 1 means the broker acknowledged the message, not that the device executed it. If you need the device's answer, use the RPC endpoint below.
The : ping comment every 15 seconds keeps proxies and load balancers from closing idle SSE connections.
New SSE clients immediately receive the cached last state, so dashboards are never empty. For devices that publish rarely, make them publish state as retained messages so the bridge also has it after a restart.
Put authentication (an API key or session middleware) in front of /api before exposing it — the bridge publishes with its own broker credentials.

Request/response with MQTT 5

Sometimes the HTTP caller needs the device's reply. MQTT 5 has this built in: the request carries a response topic and correlation data, and the device echoes the correlation data in its reply. Add this endpoint to server.js:

server.js (continued)

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(() => {});                      // avoid an unhandled rejection if publish fails

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

A device simulator for testing it:

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

The pattern and its pitfalls are covered in depth in MQTT request/response.

The browser side

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 reconnects automatically using the retry value the server sent. One caveat: over HTTP/1.1 browsers allow only about six connections per origin, and each open SSE stream uses one; serve the app over HTTP/2 if users open many tabs. For a deeper comparison of push options, see MQTT vs SSE and long polling.

Test every path

Start the bridge and, in separate terminals, the stream and an MQTT subscriber:

Terminal 1 — run

MQTT_USER=username MQTT_PASS=password node server.js

Terminal 2 — watch the SSE stream

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

Terminal 3 — watch commands reaching the 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

Now exercise each direction:

Terminal 4 — drive it

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

# MQTT -> SSE: expect an "event: state" frame 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}'

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

# RPC: run device.js first, expect {"reply": ...}; without it expect 504 after 5 s
curl -s -X POST http://localhost:3000/api/devices/lamp1/rpc \
  -H 'content-type: application/json' -d '{"method":"getStatus"}'

Instead of mosquitto_pub, you can publish the state message from the online MQTT client and watch it appear in your page. Finally, stop the broker connection (disconnect Wi-Fi or block the port) and check that POST returns 503, then that the bridge reconnects and resubscribes on its own.

Publish test states from the browser

Scaling out

Running several bridge instances behind a load balancer works with two adjustments. Each instance needs its own client ID (the process.pid suffix is fine on one host; use the hostname or pod name in containers) and its own reply topic, which the code already derives from the client ID. If you want only one instance to process each incoming state message — for example to write it to a database — switch that subscription to a shared subscription ($share/bridge/testmqtt/express/devices/+/state), while keeping the SSE fan-out on a normal subscription so every instance can serve its own browsers. For heavy traffic, test against your own broker instance: shared test endpoints are rate-limited and are meant for functional checks, not throughput.

An Express bridge is a few dozen lines of code, but it is the piece that lets the rest of your stack ignore MQTT entirely. Keep one connection per process, validate everything that ends up in a topic, return honest status codes, and test both directions with curl and a subscriber before anyone builds on it. If you want the same client code in another language, the MQTT code generator gives you a starting point.

Frequently asked questions

Should I create a new MQTT connection per HTTP request?

No. Open one long-lived MQTT client when the Express process starts and reuse it for every request. Connecting per request adds a TCP/TLS handshake and a CONNECT round trip to each call and quickly exhausts broker connection limits.

Why use Server-Sent Events instead of MQTT over WebSockets in the browser?

SSE keeps broker credentials on the server, works through most proxies as plain HTTP and needs no client library. MQTT over WebSockets is the better choice when browsers must publish often or subscribe to many dynamic topics.

What should the REST endpoint return when the broker is down?

Fail fast with 503 Service Unavailable instead of letting MQTT.js queue the message silently. The caller then knows the command was not sent and can retry.

MQTT with Node.js and Express: Build a REST to MQTT Bridge