From 72cbd67e17f126865317ab243b4836ab244d0da3 Mon Sep 17 00:00:00 2001 From: Harvey Zuccon Date: Tue, 21 Jul 2026 12:51:08 +1000 Subject: [PATCH] Update readme --- README.md | 553 ++++++++++++++++++++++++------------------------------ 1 file changed, 241 insertions(+), 312 deletions(-) diff --git a/README.md b/README.md index 45f5552..97b3c4d 100644 --- a/README.md +++ b/README.md @@ -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`. 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` 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` — 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`. + +| Option | Type | Default | +| ------------------ | ------------------------------------------------------- | ----------------------------------------------------------- | +| `method` | `"GET" \| "POST"` | `"GET"` | +| `headers` | `Record` | `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` | Global `fetch` | +| `onRequest` | `(init: RequestInit) => Promise` | 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(); +- `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` 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.