Harvmaster/sse-session
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update readme

Browse Source
This commit is contained in:
Harvey Zuccon
2026-07-21 12:51:08 +10:00
parent e12698ab6f
commit 72cbd67e17
1 changed files with 241 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

553
README.md
View File

@@ -1,309 +1,285 @@
# Server-Sent Events (SSE)
# sse-session
A fetch-based SSE client for browser and Node.js. It supports custom headers, POST bodies, exponential-backoff connect retries, automatic reconnect, optional tab-visibility handling, and `Last-Event-ID` resume semantics.
A fetch-based Server-Sent Events client for modern browsers and Node.js runtimes that provide the standard Fetch and Streams APIs.
Published from `@xo-cash/utils`:
`sse-session` supports GET and POST streams, custom request hooks, connection retries, automatic reconnects, async iteration, lifecycle events, optional Page Visibility handling, and opt-in `Last-Event-ID` headers.
## Installation
```sh
npm install sse-session
```
The package is authored as an ES module.
```ts
import {
SSESession,
SSEEventParser,
AsyncPushIterator,
ExponentialBackoff,
type SSEvent,
type SSESessionOptions,
} from "@xo-cash/utils";
} from "sse-session";
```
## Module layout
```
sse-session/
├── sse-session.ts # SSESession — main client
├── sse-event-parser.ts # Incremental SSE frame parser
├── async-push-iterator.ts # Push-based async iterable for messages
├── types.ts # Options, events, and SSEvent
└── index.ts # Public exports
```
`SSESession` is built on three internal utilities from the same package:
| Utility | Role |
|---------|------|
| `EventEmitter` | Typed pub/sub for session lifecycle and message events |
| `ExponentialBackoff` | Retries the initial HTTP connection until it succeeds |
| `tryAsync` | Runs callbacks safely without letting handler throws break the session |
---
The runtime must provide `fetch`, `Response`, `ReadableStream`, `AbortController`, `FormData`, and `TextDecoder`. A custom fetch-compatible function can be supplied through the `fetch` option.
## Quick start
`SSESession.create()` constructs a session and waits until its HTTP response body is available. Stream reading then continues in the background.
```ts
import { SSESession } from "sse-session";
const session = await SSESession.create("https://api.example.com/events");
try {
for await (const event of session.messages) {
console.log(event.event, event.data);
}
} finally {
await session.disconnect();
}
```
Messages are also emitted as events. Construct the session directly when listeners must be installed before the connection starts:
```ts
const session = new SSESession("https://api.example.com/events");
session.on("message", (event) => {
console.log(event.data);
});
await session.connect();
```
## Session API
### Creating and connecting
```ts
const session = new SSESession(url, options);
await session.connect();
```
or, as a convenience:
```ts
const session = await SSESession.create(url, options);
```
`connect()` is idempotent while a transport is active. It resolves after the fetch succeeds, the response has a non-null body, `onConnected` runs, and the `"connected"` event is emitted. It does not wait for the stream to finish.
Non-2xx responses and responses without a body are treated as connection failures. The response `Content-Type` is not validated.
### Receiving messages
Each parsed event is delivered through both of these interfaces:
```ts
const unsubscribe = session.on("message", (event) => {
handle(event);
});
for await (const event of session.messages) {
handle(event);
}
unsubscribe();
```
When finished:
`messages` is an `AsyncPushIterator<SSEvent>`. It buffers values until they are read, and breaking out of a loop does not cancel the underlying queue. Only one async iterator may consume a `messages` instance at a time; use `"message"` listeners when multiple concurrent consumers are needed.
Do not cache `session.messages` across a terminal close. A later `connect()` replaces a closed iterator with a new one.
### Stopping a session
`abort()` stops the active transport without ending the session:
- the active fetch is aborted;
- partial parser state is reset;
- `onDisconnected` and `"disconnected"` run;
- `messages` stays open for a later reconnect;
- `"closed"` and `"error"` are not emitted.
Calling `abort()` with no active transport is a no-op.
`disconnect()` is the terminal operation for the current session lifecycle:
- `messages` is closed;
- `"closed"` is emitted;
- an active transport is then aborted, which also produces `onDisconnected` and `"disconnected"`.
The same `SSESession` object can be connected again. When that happens, it creates a fresh `messages` iterator.
## Lifecycle and reconnect behavior
There are two separate reconnect controls:
| Situation | Option | Default | Behavior |
| ------------------------------------------ | ------------------ | ------- | --------------------------------------------------------------------- |
| The server ends the response body normally | `persistent` | `false` | Reconnect when `true`; otherwise close `messages`. |
| Reading the response body throws | `attemptReconnect` | `true` | Reconnect when `true`; otherwise emit the error and close `messages`. |
An intentional `abort()` does not trigger `attemptReconnect`. Automatic reconnects keep the existing `messages` iterator open.
Every call to `connect()`, including automatic reconnects, uses the configured retry policy while establishing the HTTP response. The session default is:
```ts
await session.disconnect();
```
---
## How it works
Each `SSESession` holds one HTTP streaming connection at a time. Bytes from the response body flow through `SSEEventParser`, which turns line-oriented SSE frames into `SSEvent` objects. Parsed events are delivered two ways:
1. **Events** — listen with `session.on("message", …)` (and other lifecycle events).
2. **Messages** — consume with `for await (const event of session.messages)`.
`connect()` resolves once the HTTP stream is open. Reading continues in the background; you do not need to await the full stream lifetime.
---
## SSESession
### Factory methods
| Method | Description |
|--------|-------------|
| `SSESession.create(url, options?)` | Creates a session and connects immediately. |
| `SSESession.withBrowserVisibility(url, options?)` | Same as `create`, plus tab visibility handling. Defers the first connect when the document is hidden. |
| `SSESession.addBrowserVisibilityHandler(session)` | Attaches visibility handling to an existing session. |
| `SSESession.addLastEventIdReconnect(session)` | Tracks the latest event `id` and sends `Last-Event-ID` on reconnect. |
### Instance methods
| Method | Description |
|--------|-------------|
| `connect()` | Opens or reopens the transport. |
| `abort()` | Stops the in-flight fetch without ending the session. Used for tab visibility. |
| `disconnect()` | Closes the session, closes `messages`, and emits `"closed"`. |
### Properties
| Property | Description |
|----------|-------------|
| `messages` | `AsyncPushIterator<SSEvent>` for `for await…of` consumption. |
| `onRequest` | Getter/setter for the pre-fetch hook (auth headers, `Last-Event-ID`, etc.). |
---
## Lifecycle
```
connect() ──► stream open ──► "connected"
│
▼
read & parse events ──► "message" + messages.push()
│
┌───────────┼───────────┐
▼ ▼ ▼
abort() server done read error
│ │ │
▼ ▼ ▼
"disconnected" "disconnected" "disconnected"
(messages open) │ │
│ "error" (unless aborted)
│ │
persistent? attemptReconnect?
│ │
└──── connect() again
│
disconnect() ──► close messages ──► "closed"
```
### `connect`
- Resets the event parser and reopens `messages` if it was previously closed.
- Retries the fetch via `ExponentialBackoff` (unlimited attempts by default).
- Emits `"connected"` once the stream is established.
- Starts background reading; the returned promise does not wait for the stream to end.
### `abort`
- Aborts the active fetch and invalidates the current read loop.
- Emits `"disconnected"` but **not** `"closed"`.
- Keeps `messages` open so an existing `for await` consumer resumes after reconnect.
- Resets the parser so partial frames from the abandoned transport are discarded.
### `disconnect`
- Closes `messages` and emits `"closed"`.
- Aborts any active transport.
- Detaches browser visibility handling until the next manual `connect()`.
### Automatic reconnect
Two independent flags control reconnect behaviour:
| Option | Triggers reconnect when… |
|--------|--------------------------|
| `persistent: true` | The **server** closes the stream normally. |
| `attemptReconnect: true` | A **transport error** occurs (not an intentional `abort()`). |
Both default to `true` and `false` respectively.
### Connection supersession
Each `connect()` or `abort()` bumps an internal `connectionId`. Background read loops capture their id at start and exit quietly when superseded, so stale transports never emit duplicate events or spurious errors.
---
## Events
| Event | Payload | When |
|-------|---------|------|
| `"connected"` | — | HTTP stream established. |
| `"message"` | `SSEvent` | A complete SSE frame was parsed. |
| `"disconnected"` | — | Active transport ended (including `abort()`). |
| `"error"` | `Error` | Fetch or read failure (not intentional abort). |
| `"closed"` | — | `disconnect()` was called. |
Listen with the standard `EventEmitter` API:
```ts
const off = session.on("message", (event) => { … });
off(); // unsubscribe
session.once("connected", () => { … });
```
---
## Messages iterator
`session.messages` is an `AsyncPushIterator<SSEvent>` — a push-based queue you consume with async iteration:
```ts
for await (const event of session.messages) {
process(event);
}
```
### When `messages` stays open
- `abort()` (tab hidden)
- Automatic reconnect (`persistent` / `attemptReconnect`)
### When `messages` closes
- `disconnect()`
- Server ends the stream and `persistent` is `false`
- Transport error with `attemptReconnect: false`
After a terminal close, call `connect()` again to open a fresh iterator. Read from `session.messages` directly rather than caching a reference across disconnects.
---
## Configuration
Pass a partial `SSESessionOptions` to `create()` or `withBrowserVisibility()`:
```ts
const session = await SSESession.create("/events", {
method: "POST",
headers: { Authorization: "Bearer …" },
body: JSON.stringify({ filter: "all" }),
onRequest: async (request) => {
// Mutate headers before each connect/reconnect
return request;
},
onConnected: () => console.log("connected"),
onDisconnected: () => console.log("disconnected"),
onError: (error) => console.error(error),
attemptReconnect: true,
persistent: false,
// Custom fetch (LibP2P, test doubles, etc.)
fetch: myFetch,
// Custom retry policy for the initial connection
retry: ExponentialBackoff.from({ baseDelay: 500, maxAttempts: 5 }),
// Custom parser (must implement parseEvents + reset)
eventParser: new SSEEventParser(),
new ExponentialBackoff({
baseDelay: 1_000,
maxDelay: 10_000,
maxAttempts: 0, // unlimited
growthRate: 1.3,
jitter: 0.3,
});
```
### Callbacks
| Callback | Purpose |
|----------|---------|
| `onRequest` | Transform `RequestInit` before each fetch (auth, `Last-Event-ID`). |
| `onConnected` | Stream is open; reading is about to begin. |
| `onDisconnected` | Transport ended (including `abort`). |
| `onError` | Unexpected failure. Not called for intentional aborts. |
### Default connect retry
The default `retry` is an `ExponentialBackoff` with unlimited attempts (`maxAttempts: 0`), `baseDelay: 1000`, `maxDelay: 10000`, `growthRate: 1.3`, and `jitter: 0.3`. This means `SSESession.create()` blocks until the first connection succeeds or the caller aborts/disconnects.
---
## Browser tab visibility
For browser clients that should pause SSE while a tab is in the background:
With the default unlimited policy, `create()` or `connect()` can remain pending indefinitely when the endpoint cannot be reached. Supply a finite policy when the caller needs connection failure to reject:
```ts
const session = await SSESession.withBrowserVisibility("/events");
const session = await SSESession.create(url, {
retry: ExponentialBackoff.from({
maxAttempts: 5,
baseDelay: 500,
}),
});
```
Behaviour:
When a finite policy is exhausted, `connect()` closes `messages`, invokes the disconnection and error notifications, and rejects with the retry policy's error (`AggregateError` for `ExponentialBackoff`).
- **Tab hidden** → `abort()` stops the fetch; `messages` stays open.
- **Tab visible** → `connect()` re-establishes the stream.
- **`disconnect()`** → removes the visibility listener until the next manual `connect()`.
## Events and callbacks
In Node.js (no `document`), visibility handling is a no-op.
`SSESession` extends the exported typed `EventEmitter`.
---
| Event | Payload | Emitted when |
| -------------- | --------- | ------------------------------------------------------------------------------------------------------------------ |
| `connected` | `void` | The response stream has been opened. |
| `message` | `SSEvent` | A complete event has been parsed. |
| `disconnected` | `void` | An established transport ends or `abort()` is called. It is also emitted after a connection attempt finally fails. |
| `error` | `Error` | A request hook, fetch, response, stream read, or `onConnected` callback fails. Intentional aborts are excluded. |
| `closed` | `void` | `disconnect()` is called. |
## Last-Event-ID resume
The emitter provides `on`, `once`, `off`, `emit`, `removeAllListeners`, and `waitFor`. `on` and `once` return an unsubscribe function and accept an optional debounce duration as their third argument.
To follow SSE resume semantics:
The corresponding option callbacks are invoked before their lifecycle event:
| Callback | Purpose |
| -------------------- | ------------------------------------------------------------------------------------------------------ |
| `onRequest(request)` | Returns the `RequestInit` used for each fetch. Use it for refreshed auth or other per-attempt changes. |
| `onConnected()` | Runs after the response body is available and before `"connected"`. |
| `onDisconnected()` | Runs before `"disconnected"`. |
| `onError(error)` | Runs before `"error"`. Defaults to logging the error. |
An error thrown by `onConnected` is reported through `onError` and `"error"`, after which the session still emits `"connected"` and starts reading. An error thrown by `onDisconnected` is passed to `onError`. If `onError` itself throws, that callback error is logged.
## Options
The constructor and `create()` accept `Partial<SSESessionOptions>`.
| Option | Type | Default |
| ------------------ | ------------------------------------------------------- | ----------------------------------------------------------- |
| `method` | `"GET" \| "POST"` | `"GET"` |
| `headers` | `Record<string, string>` | `Accept: text/event-stream` and `Cache-Control: no-cache` |
| `body` | `string \| FormData` | Empty `FormData`; sent only for POST requests |
| `fetch` | `(url: string, init: RequestInit) => Promise<Response>` | Global `fetch` |
| `onRequest` | `(init: RequestInit) => Promise<RequestInit>` | Returns the input unchanged |
| `onConnected` | `() => void` | No-op |
| `onDisconnected` | `() => void` | No-op |
| `onError` | `(error: Error) => void` | Logs to `console.error` |
| `retry` | Object with a `run()` method | Unlimited session-specific `ExponentialBackoff` shown above |
| `attemptReconnect` | `boolean` | `true` |
| `persistent` | `boolean` | `false` |
| `eventParser` | Object with `parseEvents()` and `reset()` | A new `SSEEventParser` per session |
Custom headers are merged over the two default headers. Each fetch also receives `cache: "no-store"`, the session's abort signal, and a `null` body for GET requests.
The resolved configuration is publicly available as `session.options`. It can be changed between connections, although passing options to the constructor is preferable when runtime mutation is not required.
Example POST stream with refreshed authentication:
```ts
const session = await SSESession.create("/events");
await SSESession.addLastEventIdReconnect(session);
const session = await SSESession.create("https://api.example.com/events", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ topic: "prices" }),
onRequest: async (request) => ({
...request,
headers: {
...request.headers,
Authorization: `Bearer ${await getAccessToken()}`,
},
}),
});
```
This tracks the most recent `event.id` from incoming messages and adds a `Last-Event-ID` header on every subsequent connect/reconnect. The existing `onRequest` callback is preserved and runs afterward.
The header is omitted until at least one event with an `id` field has been received.
---
## SSEvent
Parsed events match the SSE wire format:
The custom fetch signature deliberately accepts any string URL, which allows adapters for non-standard transports:
```ts
interface SSEvent {
data: string; // required — event payload
event?: string; // event type (default: "message")
id?: string; // last event id for resume
retry?: number; // server-suggested reconnect delay (ms)
const session = await SSESession.create("libp2p://peer/events", {
fetch: (url, init) => libp2pFetch(url, init),
});
```
## Page Visibility integration
`addBrowserVisibilityHandler()` attaches a `document.visibilitychange` listener to an existing session:
```ts
const session = new SSESession("/events");
SSESession.addBrowserVisibilityHandler(session);
await session.connect();
```
When the document becomes hidden, the handler calls `abort()`. When it becomes visible, it calls `connect()`. `disconnect()` removes the listener; connecting the same session later reattaches it after the connection succeeds. In runtimes without `document`, this method is a no-op.
The helper does not inspect the document's visibility when it is attached. To avoid the initial connection while the page is already hidden, gate it explicitly:
```ts
const session = new SSESession("/events");
SSESession.addBrowserVisibilityHandler(session);
if (document.visibilityState !== "hidden") {
await session.connect();
}
```
Multi-line `data:` fields are joined with `\n`. Trailing newlines on the payload are trimmed.
There is currently no `withBrowserVisibility()` factory.
---
## Last-Event-ID reconnects
Resume headers are opt-in:
```ts
const session = new SSESession("/events", { persistent: true });
await SSESession.addLastEventIdReconnect(session);
await session.connect();
```
The helper remembers the most recent non-`undefined` `id` received after it is attached and adds it as `Last-Event-ID` on subsequent fetches. The original `onRequest` hook runs afterward and can further modify the request. No header is sent until a non-empty ID has been observed.
Attach the helper before `connect()` if the first event must always be observed for later resume. The helper changes request headers only; it does not itself enable `persistent` or `attemptReconnect`.
## Parsed events
```ts
interface SSEvent {
data: string;
event?: string;
id?: string;
retry?: number;
}
```
`data` is required because frames without any `data` field are not emitted. Multiple `data` lines are joined with `\n`. The optional `event` field remains `undefined` when the server omits it; the parser does not fill in `"message"`.
The parser exposes a numeric `retry` field on the returned event, but `SSESession` does not apply it to its retry policy.
## SSEEventParser
Incremental parser for raw stream bytes. Used internally by `SSESession` but exported for testing or custom integrations.
`SSEEventParser` is the incremental parser used by the session and is also exported for direct use:
```ts
const parser = new SSEEventParser();
@@ -314,72 +290,25 @@ for (const chunk of chunks) {
}
}
// Clear partial state when abandoning a stream
parser.reset();
```
- Accepts arbitrary chunk boundaries; incomplete frames stay buffered.
- Handles `\r\n`, `\r`, and `\n` line endings.
- Supports `data`, `event`, `id`, and `retry` fields.
It buffers incomplete input until a blank line completes a frame, preserves multi-byte UTF-8 characters split across chunks, ignores comments and unknown fields, removes one optional space after a field's colon, and recognizes `data`, `event`, `id`, and numeric `retry` fields. It accepts `\n`, `\r`, and `\r\n` line endings.
---
Current parser limitation: a `\r\n` pair split between two byte chunks is not handled correctly. Keep those two bytes in the same chunk, or use `\n`-delimited streams, until that edge case is fixed.
## AsyncPushIterator
## Other exports
A push-based async iterable: producers call `push()`, consumers use `for await…of`.
The package currently exports all source utilities as part of its public entry point:
```ts
const stream = new AsyncPushIterator<SSEvent>();
- `AsyncPushIterator`
- `EventEmitter`
- `ExponentialBackoff`
- `tryAsync`
- the option, event-map, parser, retry, and backoff TypeScript types
stream.push({ data: "hello" });
stream.close(); // end iteration
```
These utilities are used internally by `SSESession` but can also be imported directly from `sse-session`.
Used by `SSESession.messages` to bridge callback-driven stream reading with pull-based async consumers.
## Current caveat for in-progress connection retries
---
## Supporting utilities
### EventEmitter
Lightweight typed event emitter used as the base class for `SSESession`. Provides `on`, `once`, `off`, and `emit` with optional debouncing.
### ExponentialBackoff
Retries a function with increasing delays and jitter. Used for the initial HTTP connection in `SSESession.connect()`.
```ts
const backoff = ExponentialBackoff.from({
baseDelay: 1000,
maxDelay: 10000,
maxAttempts: 0, // 0 = unlimited
});
await backoff.run(() => fetch(url));
```
### tryAsync
Executes an async function and routes failures to an optional error handler without rethrowing. Used internally when invoking user callbacks (`onConnected`, `onDisconnected`, `onError`) so a throwing handler does not crash the session.
---
## Custom fetch adapters
`fetch` is typed as `(url: string, options: RequestInit) => Promise<Response>` rather than the native `fetch` signature. This allows non-standard URL formats (for example LibP2P paths):
```ts
const session = await SSESession.create("libp2p://peer/events", {
fetch: async (url, options) => libp2pFetch(url, options),
});
```
---
## Notes
- Use `create()` or `withBrowserVisibility()` to construct sessions; the constructor is private.
- `connect()` is idempotent while already connected.
- Intentional aborts (tab visibility) do not emit `"error"` and do not trigger `attemptReconnect`.
- Each session instance owns its own `SSEEventParser` and `ExponentialBackoff` — instances do not share parser buffers.
`abort()` aborts the fetch signal, but it does not cancel the configured retry runner itself. If `connect()` is currently retrying failed fetches under the default unlimited policy, aborting or disconnecting the session does not guarantee that the pending `connect()` promise will settle. Use a finite retry policy when connection attempts need a bounded lifetime.