Sse Nodejs

Q: How does backpressure work in Node.js SSE?

When `res.write()` returns false, the internal buffer is full. Listen for the `drain` event to resume writing.

DodaTech 6 min read

title: "Advanced SSE in Node.js" description: "Learn advanced Server-Sent Event patterns in Node.js including connection pooling, event multiplexing, backpressure handling, and cluster support." weight: 17 date: 2026-06-28 lastmod: 2026-06-28 tags: ["apis", "sse"]


Node.js is well-suited for SSE due to its event-driven, non-blocking architecture. This lesson covers advanced patterns for building production-grade SSE servers.

## What You'll Learn

- Connection pooling and management
- Event filtering and multiplexing
- Backpressure handling
- Clustering for horizontal scaling
- Redis pub/sub for cross-instance events

## Why It Matters

Production SSE servers must handle thousands of connections, manage memory efficiently, and scale horizontally. Advanced patterns ensure your SSE infrastructure is reliable and performant.

## Real-World Use

A real-time analytics platform uses clustered Node.js SSE servers behind a load balancer. Redis pub/sub synchronizes events across instances, and each connection is tracked with metadata for targeted event delivery.

## Flow Chart

```mermaid
flowchart LR
    A[Load Balancer] --> B[Node Cluster 1]
    A --> C[Node Cluster 2]
    A --> D[Node Cluster 3]
    B --> E[Redis Pub/Sub]
    C --> E
    D --> E
    E --> F[Event Sources]
    F --> G[Application Events]

Code Examples

Example 1: Connection Pool with Backpressure

const http = require('http');

class SSEPool {
  constructor() {
    this.clients = new Map();
    this.writeQueue = new Map();
    this.maxQueued = 100;
  }

  addClient(req, res) {
    const id = `${req.socket.remoteAddress}-${Date.now()}-${Math.random()}`;
    
    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    });

    this.clients.set(id, res);
    this.writeQueue.set(id, []);
    
    res.on('close', () => {
      this.clients.delete(id);
      this.writeQueue.delete(id);
    });

    return id;
  }

  sendToClient(id, data) {
    const res = this.clients.get(id);
    if (!res) return false;

    const queue = this.writeQueue.get(id);
    
    // Check for backpressure
    if (queue.length >= this.maxQueued) {
      console.warn(`Client ${id} backpressure detected, closing`);
      res.end();
      this.clients.delete(id);
      this.writeQueue.delete(id);
      return false;
    }

    const message = `data: ${JSON.stringify(data)}\n\n`;
    
    const canWrite = res.write(message);
    if (!canWrite) {
      // Backpressure: queue the message
      queue.push(data);
      res.once('drain', () => this.flushQueue(id));
    }

    return true;
  }

  flushQueue(id) {
    const queue = this.writeQueue.get(id);
    const res = this.clients.get(id);
    if (!queue || !res) return;

    while (queue.length > 0) {
      const data = queue.shift();
      const message = `data: ${JSON.stringify(data)}\n\n`;
      
      if (!res.write(message)) {
        // Still backpressure, wait for next drain
        res.once('drain', () => this.flushQueue(id));
        break;
      }
    }
  }

  broadcast(data) {
    this.clients.forEach((res, id) => {
      this.sendToClient(id, data);
    });
  }

  getStats() {
    return {
      totalClients: this.clients.size,
      queuedMessages: Array.from(this.writeQueue.values())
        .reduce((sum, q) => sum + q.length, 0),
    };
  }
}

const pool = new SSEPool();

const server = http.createServer((req, res) => {
  if (req.url === '/events') {
    pool.addClient(req, res);
  } else if (req.url === '/stats') {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify(pool.getStats()));
  } else {
    res.writeHead(404);
    res.end();
  }
});

server.listen(3000);

Expected output: SSE connection pool with backpressure handling, queue management, and monitoring stats.

Example 2: Event Multiplexing

const http = require('http');
const { EventEmitter } = require('events');

class SSEMultiplexer extends EventEmitter {
  constructor() {
    super();
    this.connections = new Map();
  }

  createConnection(req, res) {
    const channels = new Set(req.url.split('?channels=')[1]?.split(',') || ['*']);
    
    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
    });

    const id = `${Date.now()}-${Math.random()}`;
    this.connections.set(id, { res, channels });

    // Send available channels
    res.write(`event: channels\ndata: ${JSON.stringify(Array.from(channels))}\n\n`);

    res.on('close', () => this.connections.delete(id));
    return id;
  }

  publish(channel, event, data) {
    const message = event
      ? `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`
      : `data: ${JSON.stringify(data)}\n\n`;

    this.connections.forEach((conn) => {
      if (conn.channels.has('*') || conn.channels.has(channel)) {
        conn.res.write(message);
      }
    });
  }
}

const multiplexer = new SSEMultiplexer();

// Server
http.createServer((req, res) => {
  if (req.url.startsWith('/events')) {
    multiplexer.createConnection(req, res);
  } else {
    res.writeHead(404);
    res.end();
  }
}).listen(3000);

// Event sources
const events = [
  { channel: 'stocks', event: 'price', data: { symbol: 'AAPL', price: 175 } },
  { channel: 'news', event: 'headline', data: { title: 'Market Update' } },
  { channel: 'weather', event: 'alert', data: { type: 'storm' } },
  { channel: 'stocks', event: 'price', data: { symbol: 'GOOG', price: 140 } },
];

let index = 0;
setInterval(() => {
  const evt = events[index % events.length];
  multiplexer.publish(evt.channel, evt.event, evt.data);
  index++;
}, 2000);

Expected output: SSE multiplexer allows clients to subscribe to specific channels and receive only matching events.

Example 3: Clustered SSE with Redis

const http = require('http');
const { Worker } = require('cluster');
const Redis = require('ioredis');

if (require('cluster').isMaster) {
  const numCPUs = require('os').cpus().length;
  for (let i = 0; i < numCPUs; i++) {
    require('cluster').fork();
  }

  require('cluster').on('exit', (worker) => {
    console.log(`Worker ${worker.process.pid} died, restarting`);
    require('cluster').fork();
  });
} else {
  const redis = new Redis();
  const sub = new Redis();

  const clients = new Map();

  http.createServer((req, res) => {
    if (req.url === '/events') {
      const id = `${process.pid}-${Date.now()}`;
      
      res.writeHead(200, {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
      });

      clients.set(id, res);
      res.on('close', () => clients.delete(id));
    }
  }).listen(3000);

  // Listen for cross-instance events
  sub.subscribe('sse-events');
  sub.on('message', (channel, message) => {
    const data = JSON.parse(message);
    clients.forEach((res) => {
      res.write(`data: ${JSON.stringify(data)}\n\n`);
    });
  });

  // Publish local events to all instances
  function publishToAll(data) {
    redis.publish('sse-events', JSON.stringify(data));
  }

  // Simulate events
  setInterval(() => {
    publishToAll({
      message: `Event from worker ${process.pid}`,
      time: Date.now(),
    });
  }, 5000);

  console.log(`Worker ${process.pid} started`);
}

Expected output: Clustered SSE with Redis pub/sub synchronizing events across all worker processes.

Common Mistakes

Mistake	Explanation
Not handling backpressure	Without backpressure handling, slow clients cause memory growth from buffered writes
Using process.memoryUsage without monitoring	SSE connections consume memory; monitor and set connection limits
Broadcasting to all without filtering	Always filter events per client to avoid overwhelming clients and wasting bandwidth
Not using cluster mode for multiple cores	Node.js single-thread limits parallelism; use cluster or worker_threads for scale
Forgetting Redis error handling	Redis connection failures break cross-instance communication; implement fallbacks

Practice Questions

How do you detect and handle backpressure in Node.js SSE?
How do you implement event channel multiplexing?
How does clustering improve SSE server performance?
How do you use Redis to synchronize events across SSE server instances?
What metrics should you monitor for SSE connections?

Challenge

Build a production-grade SSE server with clustering, Redis pub/sub for cross-instance events, per-client event filtering, backpressure handling, and Prometheus metrics for monitoring connection counts and throughput.

FAQ

How many SSE connections can a single Node.js process handle?

A Node.js process can handle 10,000-50,000 idle SSE connections. Active connections with frequent events will be limited by CPU and bandwidth.

How does backpressure work in Node.js SSE?

When res.write() returns false, the internal buffer is full. Listen for the drain event to resume writing.

Should I use cluster or worker_threads for SSE?

Cluster is simpler for horizontal scaling across CPU cores. Worker threads are better for sharing memory with the main process.

How do I handle SSE authentication in clustered mode?

Authenticate in the HTTP request handler before establishing SSE. Use a shared Redis session store for cross-instance auth.

What is the memory cost per SSE connection?

Each connection uses ~50-100KB for the socket buffer and connection metadata. 10,000 connections use approximately 500MB-1GB.

How do I implement SSE with HTTP/2?

Http2Session supports multiple streams. Create a library that maps SSE channels to separate HTTP/2 streams for multiplexing efficiency.

Mini Project

Build a horizontally scalable SSE infrastructure for a live notification system. Use Node.js cluster mode, Redis pub/sub for cross-instance communication, per-user event filtering, backpressure handling, and health monitoring endpoints.

What's Next

Learn about SSE named event types

← Previous Sse Last Event Id Next → Sse Project

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Sse