Server-Sent Events#

Real-time streaming from server to client using SSE. One-way communication from server to browser - ideal for live updates, progress bars, and AI streaming.

Info

SSE uses a single persistent HTTP connection. Unlike WebSocket, SSE is simpler, works over standard HTTP, and automatically reconnects.

Basic Usage#

typescript

app.get("/events", (ctx) => {
  return ctx.sse(async (sse) => {
    sse.sendData("Connected!");

    // Send a named event
    sse.sendEvent("update", { count: 42 });

    // Listen for client disconnect
    sse.onClose(() => {
      console.log("Client disconnected");
    });
  });
});

SSE API#

Method	Description
sse.sendData(data)	Send raw data as a string
sse.sendEvent(event, data)	Send a named event with data (data is JSON-serialized)
sse.send(message)	Send a full SSEMessage object with optional id, event, and data
sse.sendWithId(id, data)	Send data with an explicit event ID (for resumption support)
sse.getLastEventId()	Get the Last-Event-ID header value from the client
sse.close()	Close the SSE connection
sse.onClose(callback)	Register a close handler

Options#

Option	Type	Default	Description
sendInitial	boolean	true	Send initial "connected" event
initialEvent	string	"connected"	Custom event name for initial
retry	number	undefined	Client reconnection timeout (ms)
maxConnections	number	0 (unlimited)	Max concurrent SSE connections. Returns 503 when exceeded.
onReconnect	(lastEventId: string) => Promise<SSEMessage[]>	undefined	Callback to replay missed events on reconnection

typescript

app.get("/events", (ctx) => {
  return ctx.sse(async (sse) => {
    // No auto "connected" event, retry every 5s
  }, {
    sendInitial: false,
    retry: 5000,
  });
});

Reconnection Support#

Use onReconnect to replay missed events when a client reconnects with a Last-Event-ID header:

typescript

// Store event history (in production, use Redis or database)
const eventHistory: SSEMessage[][] = [];

app.get("/events", (ctx) => {
  return ctx.sse(async (sse) => {
    // Send current state
    sse.sendEvent("state", { count: 42 });

    // Listen for updates
    emitter.on("update", (data) => {
      const message = { id: Date.now(), event: "update", data };
      sse.send(message);
      eventHistory.push([message]);
    });
  }, {
    onReconnect: async (lastEventId) => {
      // Return missed events since lastEventId
      return eventHistory
        .flat()
        .filter(msg => msg.id && msg.id > Number(lastEventId));
    },
  });
});

Connection Limits#

Limit concurrent connections to prevent resource exhaustion:

typescript

app.get("/events", (ctx) => {
  return ctx.sse(async (sse) => {
    // Handle events
  }, {
    maxConnections: 100, // Limit to 100 concurrent connections
  });
});

// Check active connections
console.log("Active connections:", SSE.activeConnections);

Broadcasting#

Use SSEBroadcaster to manage multiple connections and broadcast to all clients:

typescript

import { SSEBroadcaster } from "@buntok/core/sse";

const broadcaster = new SSEBroadcaster();

app.get("/events", (ctx) => {
  return ctx.sse(async (sse) => {
    broadcaster.add(sse);

    sse.onClose(() => {
      broadcaster.remove(sse);
    });

    // Keep connection alive
    await new Promise(() => {});
  });
});

// Broadcast to all clients
broadcaster.broadcast("update", { count: 42 });

// Broadcast with filter
broadcaster.broadcastWhere(
  (sse) => sse.isConnected,
  "notification",
  { message: "Hello!" }
);

// Check connected clients
console.log("Connected clients:", broadcaster.size);

Named Events#

Use named events so clients can listen for specific event types:

typescript

// Server
app.get("/notifications", (ctx) => {
  return ctx.sse(async (sse) => {
    emitter.on("user:created", (user) => {
      sse.sendEvent("user-created", user);
    });
    emitter.on("order:placed", (order) => {
      sse.sendEvent("order-placed", order);
    });
  });
});

// Client
const es = new EventSource("/notifications");
es.addEventListener("user-created", (e) => {
  const user = JSON.parse(e.data);
  console.log("New user:", user);
});
es.addEventListener("order-placed", (e) => {
  const order = JSON.parse(e.data);
  console.log("New order:", order);
});

AI Streaming#

Stream AI responses token-by-token:

typescript

import { streamAI } from "@buntok/core";

app.post("/chat", async (ctx) => {
  const { message } = await ctx.body();

  return ctx.sse(async (sse) => {
    await streamAI({
      messages: [{ role: "user", content: message }],
      onChunk: (chunk) => {
        sse.sendEvent("chunk", { content: chunk });
      },
      onDone: () => {
        sse.sendEvent("done", { message: "Complete" });
        sse.close();
      },
    });
  });
});

Live Updates#

typescript

// Push real-time data to clients
app.get("/live/metrics", (ctx) => {
  return ctx.sse(async (sse) => {
    const interval = setInterval(async () => {
      const metrics = await getMetrics();
      sse.sendEvent("metrics", metrics);
    }, 5000);

    sse.onClose(() => {
      clearInterval(interval);
    });
  });
});

SSE vs WebSocket#

Feature	SSE	WebSocket
Direction	Server → Client	Bidirectional
Protocol	HTTP	ws:// / wss://
Auto-reconnect	Yes	No (manual)
Binary data	No	Yes
Complexity	Simple	Moderate
Use case	Live feeds, notifications, AI	Chat, games, collaboration

Pluggable Infrastructure#

For production cluster deployments (multiple instances), replace in-memory PubSub and history with pluggable stores.

typescript

import {
  SSEBroadcaster,
  MemorySSEPubSub,     // default in-memory PubSub
  MemorySSEHistory,    // default in-memory ring buffer (1000 msg)
} from "@buntok/core/sse";

// Default (single instance)
const broadcaster = new SSEBroadcaster();

// Custom (e.g., Redis-backed for multi-instance)
const broadcaster = new SSEBroadcaster({
  pubSub: redisPubSub,      // implements SSEPubSub interface
  historyStore: redisHistory, // implements SSEHistoryStore interface
  channel: "events",        // pub/sub channel name
});

Interface	Methods	Default
SSEPubSub	publish, subscribe, unsubscribe	MemorySSEPubSub
SSEHistoryStore	add, getAfter, clear	MemorySSEHistory (1000 msg ring buffer)

Info

Additional SSE methods: sse.sendComment(comment) for custom heartbeats, sse.offClose(callback) to unregister close handlers, and SSE.closeAll() to gracefully close all connections.

Next Steps#

WebSocket - Bidirectional real-time communication
AI Module - AI streaming with SSE
Event Emitter - Decouple code with events

PreviousLogger Next WebSocket