@forgeframework/streaming

Server-Sent Events (SSE) and real-time streaming utilities for the Forge Framework.

Part of the Forge Framework — a TypeScript framework for backend microservices and web applications.

Installation

npm install @forgeframework/streaming

Usage

import {
  ForgeSSEServer,
  ForgeEventStream,
  BackpressureStrategy,
  createSSEMiddleware,
  type StreamMessage,
  type SSEConfig,
} from '@forgeframework/streaming';
import { createServer } from '@forgeframework/node-core';

const sseConfig: SSEConfig = {
  heartbeatIntervalMs: 30000,
  retryMs: 3000,
  maxClients: 5000,
  compression: false,
};

const sseServer = new ForgeSSEServer(sseConfig);

const ordersChannel = sseServer.createChannel('orders', {
  maxClients: 1000,
  replayBufferSize: 50,
  backpressure: {
    strategy: BackpressureStrategy.DROP_OLDEST,
    bufferSize: 200,
    onOverflow: (dropped) => console.warn(`Dropped ${dropped.length} messages`),
  },
});

ordersChannel.onConnect((client) => {
  console.log(`Client ${client.id} connected to orders channel`);
});

ordersChannel.onDisconnect((client) => {
  console.log(`Client ${client.id} disconnected from orders channel`);
});

interface OrderEvents {
  'order.created': { orderId: string; total: number };
  'order.updated': { orderId: string; status: string };
}

const orderStream = new ForgeEventStream<OrderEvents>();
orderStream.pipe(ordersChannel);

const middleware = createSSEMiddleware({
  server: sseServer,
  pathPrefix: '/events',
  channelResolver: (req) => {
    const url = new URL(req.url!, `http://${req.headers.host}`);
    return url.pathname.split('/').pop() ?? 'default';
  },
});

const server = createServer({ port: 3000 });
server.use(middleware);

orderStream.emit('order.created', { orderId: 'ORD-001', total: 99.99 });

const notification: StreamMessage = {
  id: 'evt-001',
  event: 'order.updated',
  data: { orderId: 'ORD-001', status: 'shipped' },
};
sseServer.broadcastToChannel('orders', notification);

sseServer.sendToClient('client-42', {
  event: 'order.updated',
  data: { orderId: 'ORD-002', status: 'delivered' },
});

const stats = sseServer.getStats();
console.log(`Active: ${stats.totalClients} clients across ${stats.totalChannels} channels`);

process.on('SIGTERM', async () => {
  await sseServer.shutdown();
});

Documentation

Full API reference and guides: https://carbonforge.io/framework/docs/events/streaming

(Documentation site launching soon.)

License

MIT © Carbon Forge