express4-ssevent

Server-Sent Events implementation for Express with channel multiplexing and dynamic subscriptions.

Installation

npm install express4-ssevent

Quick Start

Server

import express from 'express';
import { sse } from 'express4-ssevent';

const app = express();
app.use(express.json());

app.all('/events', sse.init);

sse.send('notifications', { message: 'Hello' });
sse.sendTo('client-uuid', { private: 'message' });
sse.broadcast({ announcement: 'Server update' });

app.listen(3000);

Browser

import { connectSSE } from 'express4-ssevent';

const client = connectSSE('/events');

client.on('connected', ({ data }) => {
  client.subscribe('notifications', 'updates');
});

client.on('message', ({ data }) => {
  console.log(data);
});

CDN

<script src="https://unpkg.com/express4-ssevent/dist/ssevent.min.js"></script>
<script>
  const client = SSEvent.connect('/events');
  client.on('message', e => console.log(e.data));
</script>

API Reference

Server

sse.setup(options)

sse.setup({
  timeout: 300000,
  heartbeat: true,
  heartbeatInterval: 30000,
  maxPayloadSize: 65536,
  maxChannelsPerClient: 20,
  maxTotalConnections: 1000,
  allowDefaultChannel: true,
  cleanupInterval: 0,
  
  onConnect: async (req, res) => {
    return !!req.session?.userId;
  },
  
  onSubscribe: async (clientId, channel, req) => {
    return true;
  }
});

sse.init(req, res)

Express handler for SSE endpoint.

app.all('/events', sse.init);

sse.send(channel, data)

Send message to channel subscribers.

sse.send('notifications', { title: 'New message', body: 'Content' });

sse.sendTo(clientId, data, event?)

Send private message to specific client.

sse.sendTo('client-uuid', { alert: 'Private' });

sse.broadcast(data, event?)

Send message to all connected clients.

sse.broadcast({ announcement: 'Maintenance in 5 min' });

sse.subscribe(clientId, ...channels)

Server-side subscription (bypasses onSubscribe hook).

sse.subscribe('client-uuid', 'news', 'sports');

sse.unsubscribe(clientId, ...channels?)

Unsubscribe from channels or disconnect.

sse.unsubscribe('client-uuid', 'news');
sse.unsubscribe('client-uuid'); // disconnect

sse.getClientCount()

Get active connection count.

const count = sse.getClientCount();

sse.getClientsByChannel(channel)

Get client IDs subscribed to channel.

const clients = sse.getClientsByChannel('notifications');

sse.clearClients()

Disconnect all clients.

sse.clearClients();

Client

connectSSE(url, options?)

const client = connectSSE('/events', {
  autoReconnect: true,
  maxReconnectAttempts: Infinity,
  reconnectDelay: 3000
});

client.on(event, callback)

client.on('message', ({ data, raw }) => {});
client.on('connected', ({ data }) => {});
client.on('open', () => {});
client.on('error', ({ error }) => {});

client.off(event, callback?)

client.off('message', handler);
client.off('message'); // remove all

client.subscribe(...channels)

await client.subscribe('news', 'updates');

client.unsubscribe(...channels)

await client.unsubscribe('news');

client.getClientId()

const id = client.getClientId();

client.isConnected()

if (client.isConnected()) { /* ... */ }

client.getReadyState()

const state = client.getReadyState(); // 0, 1, or 2

client.close()

client.close();

Configuration Options

| Option | Type | Default | Description | |--------|------|---------|-------------| | timeout | number | 300000 | Client disconnect timeout (ms) | | heartbeat | boolean | true | Enable heartbeat messages | | heartbeatInterval | number | 30000 | Heartbeat interval (ms) | | maxPayloadSize | number | 65536 | Max message size (bytes) | | maxChannelsPerClient | number | 20 | Max channels per client | | maxTotalConnections | number | 1000 | Max concurrent connections | | allowDefaultChannel | boolean | true | Auto-subscribe to default | | cleanupInterval | number | 0 | Stale connection cleanup (ms) | | trustOrigins | (string|RegExp)[] | [] | Allowed Origins/Referers | | onConnect | function | () => true | Validate connections | | onSubscribe | function | () => true | Validate subscriptions |

Examples

JWT Authentication

import jwt from 'jsonwebtoken';

sse.setup({
  onConnect: async (req, res) => {
    const token = req.headers.authorization?.replace('Bearer ', '');
    try {
      req.user = jwt.verify(token, process.env.JWT_SECRET);
      return true;
    } catch {
      return false;
    }
  },
  
  onSubscribe: async (clientId, channel, req) => {
    // Note: 'req' here is the POST request. 
    // Ensure you have auth middleware running on POST /events too.
    const token = req.headers.authorization?.replace('Bearer ', '');
    // ... verify token ...
    if (channel.startsWith('user-')) {
       // example validation
       return true;
    }
    return true;
  }
});

Rate Limiting

import rateLimit from 'express-rate-limit';

const limiter = rateLimit({
  windowMs: 60000,
  max: 10
});

app.use('/events', limiter);
app.all('/events', sse.init);

User Notifications

// Server
function notifyUser(userId, notification) {
  sse.send(`user-${userId}`, notification);
}

// Client
client.on('connected', async () => {
  await client.subscribe(`user-${currentUserId}`);
});

client.on('message', ({ data }) => {
  if (data.channel.startsWith('user-')) {
    showNotification(data.payload);
  }
});

Security

CORS

import cors from 'cors';

app.use(cors({
  origin: process.env.ALLOWED_ORIGINS.split(','),
  credentials: true
}));

HTTPS

if (process.env.NODE_ENV === 'production' && !req.secure) {
  return res.redirect(`https://${req.headers.host}${req.url}`);
}

Sanitization

import DOMPurify from 'dompurify';

client.on('message', ({ data }) => {
  element.innerHTML = DOMPurify.sanitize(data.payload);
});

License

ISC