Real-Time Updates: MQTT, Express, and Server-Sent Events Explained

Modern web applications frequently demand real-time data updates. Think of live sports scores, stock tickers, chat applications, or IoT dashboards. Delivering data to users the instant it changes is a core requirement, and for this, you need efficient, scalable tools.

This article delves into building real-time update systems using Message Queuing Telemetry Transport (MQTT) – a lightweight publish-subscribe messaging protocol – combined with Express.js as your backend framework. We'll explore how to set up an MQTT broker like Mosquitto, integrate it with Express, and push updates to web browsers using Server-Sent Events (SSE). Our example will be a football (soccer) sports update system, complete with an admin interface for score input and a viewer interface for live updates.

Understanding the Architecture

Our real-time system is composed of three primary parts:

Admin Interface: A web page enabling match creation, score updates, and event logging (goals, cards).
Express Server: This Node.js backend handles HTTP requests from the admin, publishes data to the MQTT broker, subscribes to MQTT topics, and streams real-time updates to connected viewers via Server-Sent Events.
Viewer Interface: A separate web page that connects to the Express server, receiving and displaying live scores and match events without requiring page refreshes.

The data flow is straightforward: when an admin submits an update, the Express server publishes a message to the MQTT broker. Since the same Express server also subscribes to these topics, it receives the message back. It then forwards this data to all connected viewer clients through Server-Sent Events, ensuring instant UI updates.

Why MQTT for Real-Time?

MQTT (Message Queuing Telemetry Transport) is a publish-subscribe protocol ideal for low-bandwidth and unreliable networks, making it excellent for IoT and various real-time broadcasting scenarios. Its advantages include:

Low Overhead: Efficient, small messages minimize network traffic.
Quality of Service (QoS): Offers delivery guarantees (at most once, at least once, exactly once).
Topic-Based Routing: Messages are organized hierarchically by topics (e.g., sports/football/match/123), allowing subscribers to receive only relevant data.
Broker-Based: A central broker (like Mosquitto) manages message distribution, simplifying application logic.

MQTT Topic Design and QoS

For our football update system, we'll use specific topics:

sports/football/match/{id}: Dedicated topic per match, broadcasting the full match object state.
sports/football/scores: For score-related notifications, including a type field (e.g., match_created, score_update).
sports/football/events: For specific match events like goals or cards.

Wildcards (# for all levels below, + for single level) enable flexible subscriptions. For instance, sports/football/# subscribes to all topics under sports/football. When publishing, we'll use QoS level 1 ("at least once"), ensuring the broker retries delivery until acknowledgement, mitigating data loss during brief connection interruptions.

Server-Sent Events (SSE) vs. WebSockets

While both SSE and WebSockets can push data to browsers, SSE is chosen for this viewer-centric system due to its simplicity for one-way communication:

Server-Sent Events: One-way (server to client), built on standard HTTP, features automatic browser reconnection, and is simpler to implement without extra libraries.
WebSockets: Two-way, uses a different protocol, more flexible but also more complex.

For a scenario where the client only needs to receive updates, SSE is a more natural and lightweight fit. If interactive client-to-server messaging (e.g., filtering by league) were required on the same channel, WebSockets would be more appropriate, or a separate HTTP API could be added.

Project Setup and Dependencies

To begin, create a project directory and initialize npm:

bash mkdir mqtt-football-scores cd mqtt-football-scores npm init -y

Install the necessary dependencies:

bash npm install express cors mqtt uuid

express: The web framework for our HTTP server.
cors: Enables Cross-Origin Resource Sharing.
mqtt: The Node.js MQTT client library.
uuid: Generates unique identifiers.

Ensure your package.json includes "type": "module" to use ES module syntax (import/export).

Setting Up the MQTT Broker

A running MQTT broker is essential. Docker is the recommended approach for setting up Mosquitto:

Create docker-compose.yml:

yaml version: "3.8" services: mosquitto: image: eclipse-mosquitto:2 container_name: mqtt-football-mosquitto ports: - "1883:1883" volumes: - ./mosquitto.conf:/mosquitto/config/mosquitto.conf restart: unless-stopped

Create mosquitto.conf:

plaintext listener 1883 protocol mqtt allow_anonymous true log_dest stdout log_type all

Then, start the broker:

bash docker-compose up -d

For local development, allow_anonymous true simplifies setup, but robust authentication should be used in production.

Building the Express Server

The Express server (located at server/index.js) acts as the central hub, handling HTTP requests, serving static files, and managing the MQTT-to-SSE bridge.

Here’s a simplified view of server/index.js:

javascript import express from 'express'; import cors from 'cors'; import mqtt from 'mqtt'; import { fileURLToPath } from 'url'; import { dirname, join } from 'path'; import { v4 as uuidv4 } from 'uuid'; import { matchRoutes } from './routes/matches.js'; import { setupSSE, addSSEClient } from './sse.js';

const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename);

const MQTT_BROKER = process.env.MQTT_BROKER || 'mqtt://localhost:1883'; const PORT = process.env.PORT || 3000;

const app = express(); app.use(cors()); app.use(express.json()); app.use(express.static(join(__dirname, '../public')));

function connectMQTT() { const mqttClient = mqtt.connect(MQTT_BROKER, { clientId: football-scores-${uuidv4().slice(0, 8)}, reconnectPeriod: 3000, connectTimeout: 10000, });

mqttClient.on('connect', () => { console.log('Connected to MQTT broker at', MQTT_BROKER); mqttClient.subscribe('sports/football/#', { qos: 1 }, (err) => { if (err) console.error('Subscribe error:', err); }); });

mqttClient.on('error', (err) => { console.error('MQTT error:', err.message); });

// ... other MQTT event handlers ...

return mqttClient; }

const mqttClientInstance = connectMQTT(); const { publishMatch, publishScoreUpdate, publishEvent, getMatches } = matchRoutes(mqttClientInstance);

setupSSE(mqttClientInstance); // Setup MQTT message forwarding to SSE clients

app.get('/api/events', (req, res) => addSSEClient(res)); // SSE endpoint app.post('/api/matches', publishMatch); app.patch('/api/matches/:id/score', publishScoreUpdate); app.post('/api/matches/:id/events', publishEvent); app.get('/api/matches', getMatches);

app.listen(PORT, () => { console.log(Football Scores Server running at http://localhost:${PORT}); });

Key aspects:

connectMQTT: Establishes a connection to the MQTT broker with a unique client ID and subscription to sports/football/# (QoS 1) upon successful connection.
Middleware: cors() for cross-origin requests, express.json() for parsing JSON bodies, and express.static() to serve frontend assets from the public folder.
SSE Integration: The /api/events endpoint uses addSSEClient to register new browser connections. setupSSE registers a listener on the MQTT client to push incoming messages to all registered SSE clients.
API Routes: POST/PATCH/GET endpoints delegate to handlers for managing matches, scores, and events.

Implementing Match Routes

Match-related logic resides in server/routes/matches.js, handling the creation, update, and retrieval of match data. For simplicity, match data is stored in an in-memory Map; for production, a persistent database like PostgreSQL or MongoDB would be used.

javascript import { v4 as uuidv4 } from 'uuid';

const TOPIC_MATCH = 'sports/football/match'; const TOPIC_SCORES = 'sports/football/scores'; const TOPIC_EVENTS = 'sports/football/events';

const matches = new Map(); // In-memory store

export function matchRoutes(mqttClient) { return { publishMatch: (req, res) => { // ... logic to create match object ... const match = { /* ... match data ... */ id: uuidv4() }; matches.set(match.id, match); publish(mqttClient, ${TOPIC_MATCH}/${match.id}, match); publish(mqttClient, TOPIC_SCORES, { type: 'match_created', match }); res.status(201).json(match); },

publishScoreUpdate: (req, res) => {
  const { id } = req.params;
  const match = matches.get(id);
  if (!match) return res.status(404).json({ error: 'Match not found' });

  // ... logic to update scores/minute/status ...
  publish(mqttClient, `${TOPIC_MATCH}/${id}`, match);
  publish(mqttClient, TOPIC_SCORES, { type: 'score_update', matchId: id, /* ... scores ... */ });
  res.json(match);
},

publishEvent: (req, res) => {
  const { id } = req.params;
  const match = matches.get(id);
  if (!match) return res.status(404).json({ error: 'Match not found' });

  const event = { /* ... event data ... */ id: uuidv4().slice(0, 8) };
  match.events.push(event);
  if (event.type === 'goal') { /* ... increment score ... */ }
  
  publish(mqttClient, `${TOPIC_MATCH}/${id}`, match);
  publish(mqttClient, TOPIC_EVENTS, { type: 'match_event', matchId: id, event });
  res.status(201).json({ match, event });
},

getMatches: (req, res) => {
  const list = Array.from(matches.values()).sort(
    (a, b) => new Date(b.createdAt) - new Date(a.createdAt)
  );
  res.json(list);
},

}; }

Each route handler performs the following:

publishMatch: Creates a new match, stores it, and publishes the full match object to its specific topic (sports/football/match/{id}) and a notification to the general scores topic.
publishScoreUpdate: Updates an existing match's scores, minute, and status, then publishes the full match object and a score update notification.
publishEvent: Adds a new event to a match, potentially incrementing scores for goals, and publishes both the updated match state and an event notification.
getMatches: Retrieves all stored matches, sorts them, and returns them as JSON.

The publish helper ensures the MQTT client is connected before attempting to send messages, using QoS 1 and converting payloads to JSON.

Bridging MQTT to the Browser with Server-Sent Events

The server/sse.js module, although not fully detailed in the source, plays a crucial role by bridging the MQTT message stream to browser clients via SSE. The index.js file demonstrates its usage:

setupSSE(mqttClientInstance): This function registers a listener on the MQTT client's message event. Whenever a new MQTT message arrives, setupSSE's listener processes it and forwards the payload to all currently connected SSE clients.
addSSEClient(res): When a browser connects to the /api/events endpoint, addSSEClient is invoked. It configures the HTTP response headers for SSE (e.g., Content-Type: text/event-stream), ensures the connection remains open, and adds the response object to a collection of active SSE clients. This allows setupSSE to iterate through these client connections and push data. Each message sent through SSE is typically prefixed with data:, followed by the JSON payload, and terminated by two newline characters.

This setup effectively translates the lightweight, broker-centric MQTT messages into an HTTP-compatible stream that modern web browsers can easily consume, enabling dynamic, real-time updates on the viewer interface without complex client-side libraries.

Practical Takeaways and Production Considerations

This architecture provides a solid foundation for real-time systems. For production deployment, you would enhance it by:

Persistent Storage: Replace the in-memory Map with a robust database (e.g., PostgreSQL, MongoDB) to store match data persistently.
Authentication & Authorization: Secure your admin interface and API endpoints. Implement user authentication and role-based authorization.
MQTT Security: Configure your Mosquitto broker with username/password authentication and TLS/SSL encryption for secure communication.
Scalability: Consider load balancing for your Express server and potentially clustering your MQTT broker for high availability.

FAQ

Q: Why use MQTT instead of simply having the Express server publish directly to WebSockets?

A: MQTT introduces a decoupled architecture via a broker. This means your Express server doesn't need to directly manage connections for every client or know about all subscribers. It publishes messages to the broker, which handles distribution. This approach is more scalable, robust, and allows multiple publishers and subscribers (e.g., other services) to easily integrate without direct server-to-server communication.

Q: What happens if the Express server restarts? Do clients lose updates?

A: When the Express server restarts, existing Server-Sent Events connections from viewers will automatically attempt to reconnect, a built-in browser feature for SSE. Once reconnected, the viewer will resume receiving updates. However, any match data stored only in the Express server's in-memory Map would be lost. This underscores the need for a persistent database in production to ensure data integrity across server restarts.

Q: How does MQTT's Quality of Service (QoS) level 1 guarantee delivery?

A: QoS 1, "at least once," ensures that the message is delivered to the broker or subscriber at least once. When a publisher sends a message with QoS 1, it expects a PUBACK (Publish Acknowledgment) from the broker. If PUBACK isn't received within a certain timeframe, the publisher will re-send the message. Similarly, when a broker forwards a QoS 1 message to a subscriber, it expects a PUBACK from the subscriber; otherwise, it retries. This mechanism prevents messages from being lost due to transient network issues, though it might result in duplicate messages (which your application should handle idempotently if strict "exactly once" behavior is required).