Add SSE Session class
This commit is contained in:
@@ -1,3 +1,23 @@
|
||||
/**
|
||||
* Error thrown when a response body is null
|
||||
*/
|
||||
export class ResponseBodyNullError extends Error {
|
||||
constructor() {
|
||||
super('Response body is null');
|
||||
this.name = 'ResponseBodyNullError';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Error thrown when an HTTP error occurs
|
||||
*/
|
||||
export class HTTPError extends Error {
|
||||
constructor(status: number, message: string) {
|
||||
super(`HTTP error! Status: ${status} - ${message}`);
|
||||
this.name = 'HTTPError';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Error thrown when the maximum number of retries is hit in an exponential backoff
|
||||
*/
|
||||
|
||||
@@ -1,2 +1,4 @@
|
||||
export * from './types.ts';
|
||||
export * from './sse-session.ts';
|
||||
export * from './sse-event-parser.ts';
|
||||
export * from './async-push-iterator.ts';
|
||||
|
||||
477
source/sse-session/sse-session.ts
Normal file
477
source/sse-session/sse-session.ts
Normal file
@@ -0,0 +1,477 @@
|
||||
import type { SSESessionOptions, SSESessionEventMap, SSEvent } from './types.ts';
|
||||
|
||||
import { HTTPError, ResponseBodyNullError } from '../errors.ts';
|
||||
import { tryAsync } from '../misc.ts';
|
||||
import { EventEmitter } from '../event-emitter.ts';
|
||||
import { ExponentialBackoff } from '../exponential-backoff.ts';
|
||||
|
||||
import { SSEEventParser } from './sse-event-parser.ts';
|
||||
import { AsyncPushIterator } from './async-push-iterator.ts';
|
||||
|
||||
/**
|
||||
* A fetch-based Server-Sent Events (SSE) client with reconnect and optional
|
||||
* browser tab visibility handling.
|
||||
*
|
||||
* Each session maintains one HTTP streaming connection at a time. Incoming
|
||||
* bytes are parsed into {@link SSEvent} objects and delivered through two
|
||||
* surfaces:
|
||||
*
|
||||
* - **Events** — `"connected"`, `"message"`, `"disconnected"`, `"error"`,
|
||||
* and `"closed"` on the session itself (extends {@link EventEmitter}).
|
||||
* - **Messages** — {@link messages}, an async iterable for `for await...of`
|
||||
* consumers.
|
||||
*
|
||||
* Typical usage:
|
||||
*
|
||||
* ```ts
|
||||
* const session = await SSESession.create("/events");
|
||||
*
|
||||
* session.on("message", (event) => console.log(event.data));
|
||||
*
|
||||
* for await (const event of session.messages) {
|
||||
* handle(event);
|
||||
* }
|
||||
* ```
|
||||
*
|
||||
* ## Lifecycle
|
||||
*
|
||||
* - {@link connect} opens (or reopens) the transport. It resolves once the
|
||||
* HTTP stream is established; reading continues in the background.
|
||||
* - {@link abort} stops the in-flight fetch without ending the session.
|
||||
* Used internally for tab visibility. The {@link messages} iterator stays
|
||||
* open so an existing consumer resumes when the tab becomes visible again.
|
||||
* - {@link disconnect} aborts the transport, closes {@link messages}, emits
|
||||
* `"closed"`, and disables attached visibility handlers until the next
|
||||
* manual {@link connect}.
|
||||
*
|
||||
* Automatic reconnect is controlled by {@link SSESessionOptions.persistent}
|
||||
* (server closed the stream) and
|
||||
* {@link SSESessionOptions.attemptReconnect} (transport error).
|
||||
*
|
||||
* ## Connection supersession
|
||||
*
|
||||
* Each {@link connect} or {@link abort} bumps an internal `connectionId`.
|
||||
* Background read loops capture their id at start and exit quietly when a
|
||||
* newer connection supersedes them, avoiding duplicate events or errors from
|
||||
* stale transports.
|
||||
*/
|
||||
export class SSESession extends EventEmitter<SSESessionEventMap> {
|
||||
/**
|
||||
* Creates a session and waits until the first connection is established.
|
||||
*
|
||||
* @param url - The SSE endpoint URL.
|
||||
* @param options - Configuration merged with instance defaults.
|
||||
* @returns A connected session.
|
||||
* @throws When the initial connection cannot be established.
|
||||
*/
|
||||
static async create(url: string, options: Partial<SSESessionOptions> = {}): Promise<SSESession> {
|
||||
const client = new SSESession(url, options);
|
||||
await client.connect();
|
||||
|
||||
return client;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enables SSE resume semantics by sending `Last-Event-ID` on reconnect.
|
||||
*
|
||||
* Listens for incoming `"message"` events and remembers the most recent
|
||||
* {@link SSEvent.id}. On every subsequent connect or reconnect, the session's
|
||||
* {@link onRequest} hook is wrapped so that header is attached when an id is
|
||||
* known, allowing the server to replay only events the client has not yet
|
||||
* received.
|
||||
*
|
||||
* The existing {@link onRequest} callback is preserved and runs after the
|
||||
* header is applied, so auth or other header mutations continue to work.
|
||||
*
|
||||
* Attach as early in the session lifetime as possible. When added after
|
||||
* {@link create}, the initial connection omits the header (no id yet);
|
||||
* all later reconnects include it. To instrument before the first connect,
|
||||
* call this on the session returned from {@link withBrowserVisibility}
|
||||
* before awaiting a separate {@link connect} when the tab starts hidden.
|
||||
*
|
||||
* ```ts
|
||||
* const session = await SSESession.create(url);
|
||||
* await SSESession.addLastEventIdReconnect(session);
|
||||
* // Reconnects send Last-Event-ID once an event with an id is received.
|
||||
* ```
|
||||
*
|
||||
* @param client - The session to instrument.
|
||||
* @returns The same session, for chaining.
|
||||
*/
|
||||
static async addLastEventIdReconnect(client: SSESession): Promise<SSESession> {
|
||||
let lastEventId: string | undefined;
|
||||
|
||||
client.on('message', (event) => {
|
||||
lastEventId = event.id;
|
||||
});
|
||||
|
||||
const originalOnRequest = client.options.onRequest;
|
||||
|
||||
client.options.onRequest = async (request: RequestInit): Promise<RequestInit> => {
|
||||
if (lastEventId) {
|
||||
request.headers = { ...request.headers, 'Last-Event-ID': lastEventId };
|
||||
}
|
||||
|
||||
return originalOnRequest(request);
|
||||
};
|
||||
|
||||
return client;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pauses and resumes a session based on browser tab visibility.
|
||||
*
|
||||
* Uses the Page Visibility API (`document.visibilitychange`):
|
||||
*
|
||||
* - **hidden** — {@link abort} stops the active fetch. {@link messages}
|
||||
* stays open; `"disconnected"` fires but `"closed"` does not.
|
||||
* - **visible** — {@link connect} re-establishes the stream if needed.
|
||||
*
|
||||
* The listener is removed when {@link disconnect} emits `"closed"`, and
|
||||
* re-attached automatically on the next `"connected"` event.
|
||||
*
|
||||
* No-op in non-browser environments where `document` is undefined.
|
||||
*
|
||||
* @param client - The session to manage.
|
||||
*/
|
||||
static addBrowserVisibilityHandler(client: SSESession): SSESession {
|
||||
if (typeof document === 'undefined') return client;
|
||||
|
||||
const handleVisibilityChange = (): void => {
|
||||
if (document.visibilityState === 'hidden') {
|
||||
void client.abort();
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
client.connect().catch(() => {
|
||||
// connect() reports failures via onError and the "error" event.
|
||||
});
|
||||
};
|
||||
|
||||
document.addEventListener('visibilitychange', handleVisibilityChange);
|
||||
|
||||
// Stop managing visibility after an explicit disconnect; re-register
|
||||
// when the same instance is manually connected again.
|
||||
client.once('closed', () => {
|
||||
document.removeEventListener('visibilitychange', handleVisibilityChange);
|
||||
|
||||
client.once('connected', () => {
|
||||
SSESession.addBrowserVisibilityHandler(client);
|
||||
});
|
||||
});
|
||||
|
||||
return client;
|
||||
}
|
||||
|
||||
/** SSE endpoint URL for this session. */
|
||||
private readonly url: string;
|
||||
|
||||
/**
|
||||
* Per-instance configuration.
|
||||
*
|
||||
* Defaults live on the instance field (not a shared static) so each session
|
||||
* gets its own {@link SSEEventParser} and {@link ExponentialBackoff}.
|
||||
*/
|
||||
public options: SSESessionOptions = {
|
||||
fetch: (...args) => fetch(...args),
|
||||
method: 'GET',
|
||||
headers: {
|
||||
Accept: 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
},
|
||||
body: new FormData(),
|
||||
|
||||
onRequest: (request) => Promise.resolve(request),
|
||||
onConnected: () => {},
|
||||
onDisconnected: () => {},
|
||||
onError: (error) => console.error('SSEClient error:', error),
|
||||
|
||||
// Retry the initial fetch until it succeeds (maxAttempts: 0 = unlimited).
|
||||
retry: new ExponentialBackoff({
|
||||
baseDelay: 1000,
|
||||
maxDelay: 10000,
|
||||
maxAttempts: 0,
|
||||
growthRate: 1.3,
|
||||
jitter: 0.3,
|
||||
}),
|
||||
|
||||
attemptReconnect: true,
|
||||
persistent: false,
|
||||
|
||||
eventParser: new SSEEventParser(),
|
||||
};
|
||||
|
||||
/** AbortController for the currently active fetch, if any. */
|
||||
private controller: AbortController | null = null;
|
||||
|
||||
/**
|
||||
* Asynchronous stream of parsed SSE events for the active connection.
|
||||
*
|
||||
* Stays open across {@link abort} and automatic reconnects so an existing
|
||||
* `for await` consumer keeps receiving events after visibility resumes.
|
||||
*
|
||||
* Closes when:
|
||||
* - the server ends the stream and {@link SSESessionOptions.persistent}
|
||||
* is false,
|
||||
* - {@link disconnect} is called, or
|
||||
* - a transport error occurs with
|
||||
* {@link SSESessionOptions.attemptReconnect} disabled.
|
||||
*
|
||||
* A later {@link connect} replaces this with a new iterator when the
|
||||
* previous one was closed. Consumers should read from `session.messages`
|
||||
* rather than caching a reference across terminal disconnects.
|
||||
*/
|
||||
public messages: AsyncPushIterator<SSEvent> = new AsyncPushIterator<SSEvent>();
|
||||
|
||||
public constructor(url: string, options: Partial<SSESessionOptions> = {}) {
|
||||
super();
|
||||
|
||||
this.url = url;
|
||||
this.options = {
|
||||
...this.options,
|
||||
...options,
|
||||
// Shallow merge would drop default headers when options.headers is set.
|
||||
headers: { ...this.options.headers, ...options.headers },
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Connects or reconnects to the SSE endpoint.
|
||||
*
|
||||
* Resolves once the HTTP stream is established and `"connected"` has been
|
||||
* emitted. Body reading continues asynchronously in the background via
|
||||
* {@link readStream}.
|
||||
*
|
||||
* @throws When the fetch retry policy exhausts attempts or the connection
|
||||
* is superseded before the reader is handed off (in the latter case the
|
||||
* promise resolves without throwing).
|
||||
*/
|
||||
public async connect(): Promise<void> {
|
||||
// If there is already a controller present, we are already connected.
|
||||
if (this.controller) return;
|
||||
|
||||
// Prepare for a fresh transport. Parser state from an abandoned connection
|
||||
// must not bleed into the next one; reopen messages if a prior terminal
|
||||
// close ended the consumer's iteration loop.
|
||||
this.resetEventParser();
|
||||
this.ensureMessageStreamOpen();
|
||||
|
||||
const controller = new AbortController();
|
||||
this.controller = controller;
|
||||
|
||||
const { method, headers, body } = this.options;
|
||||
|
||||
const fetchBody = method === 'POST' ? body : null;
|
||||
|
||||
const fetchOptions: RequestInit = {
|
||||
method,
|
||||
headers: headers || {},
|
||||
body: fetchBody,
|
||||
signal: controller.signal,
|
||||
cache: 'no-store',
|
||||
};
|
||||
|
||||
let reader: ReadableStreamDefaultReader<Uint8Array>;
|
||||
|
||||
try {
|
||||
reader = await this.options.retry.run(() => this.createReader(fetchOptions));
|
||||
} catch (error) {
|
||||
// A newer abort/connect superseded this attempt — leave state to the winner.
|
||||
if (this.controller !== controller) return;
|
||||
|
||||
this.controller = null;
|
||||
|
||||
await this.notifyDisconnected();
|
||||
await this.notifyError(error);
|
||||
this.closeMessageStream();
|
||||
|
||||
throw error;
|
||||
}
|
||||
|
||||
// Connection succeeded but was already replaced (for example abort during fetch).
|
||||
if (this.controller !== controller) {
|
||||
await reader.cancel();
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
await tryAsync(
|
||||
() => this.options.onConnected(),
|
||||
(error) => this.options.onError(error),
|
||||
);
|
||||
this.emit('connected', undefined);
|
||||
|
||||
// Fire-and-forget: connect() resolves while the stream is consumed.
|
||||
this.readStream(reader, controller).catch((error) => {
|
||||
this.options.onError(error);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Aborts only the currently active transport.
|
||||
*
|
||||
* The session remains reusable: {@link messages} stays open, visibility
|
||||
* handling stays attached, and {@link connect} can reopen the stream.
|
||||
* Partial parser state from the abandoned transport is discarded.
|
||||
*
|
||||
* Emits `"disconnected"` but not `"closed"`.
|
||||
*/
|
||||
public async abort(): Promise<void> {
|
||||
if (!this.controller) return;
|
||||
|
||||
// Grab the current controller to ensure we are aborting the correct one.
|
||||
const controller = this.controller;
|
||||
this.controller = null;
|
||||
|
||||
// Invalidate any in-flight read loop and fetch for this transport.
|
||||
controller.abort();
|
||||
this.resetEventParser();
|
||||
|
||||
await this.notifyDisconnected();
|
||||
}
|
||||
|
||||
/**
|
||||
* Terminates the session and disables attached visibility handling until
|
||||
* the same instance is manually {@link connect connected} again.
|
||||
*
|
||||
* Closes {@link messages} and emits `"closed"`.
|
||||
*/
|
||||
public async disconnect(): Promise<void> {
|
||||
this.closeMessageStream();
|
||||
this.emit('closed', undefined);
|
||||
|
||||
if (this.controller) {
|
||||
await this.abort();
|
||||
} else {
|
||||
this.resetEventParser();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs the HTTP request and returns a reader for the response body.
|
||||
*
|
||||
* {@link SSESessionOptions.onRequest} may mutate headers (for example auth
|
||||
* tokens or `Last-Event-ID`) before the fetch runs.
|
||||
*/
|
||||
private async createReader(fetchOptions: RequestInit): Promise<ReadableStreamDefaultReader<Uint8Array>> {
|
||||
const requestOptions = await this.options.onRequest(fetchOptions);
|
||||
const response = await this.options.fetch(this.url, requestOptions);
|
||||
|
||||
if (!response.ok) {
|
||||
const responseCode = response.status;
|
||||
const responseText = await response.text();
|
||||
|
||||
const error = new HTTPError(responseCode, responseText);
|
||||
void this.notifyError(error);
|
||||
throw error;
|
||||
}
|
||||
|
||||
if (!response.body) {
|
||||
const error = new ResponseBodyNullError();
|
||||
void this.notifyError(error);
|
||||
throw error;
|
||||
}
|
||||
|
||||
return response.body.getReader();
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads bytes from an established stream until it ends, errors, or is
|
||||
* superseded by a newer connection.
|
||||
*/
|
||||
private async readStream(reader: ReadableStreamDefaultReader<Uint8Array>, controller: AbortController): Promise<void> {
|
||||
try {
|
||||
while (this.controller === controller) {
|
||||
const { done, value } = await reader.read();
|
||||
|
||||
// abort() or a newer connect() may have landed while we were awaiting.
|
||||
if (this.controller !== controller) return;
|
||||
|
||||
if (done) {
|
||||
this.controller = null;
|
||||
|
||||
await this.notifyDisconnected();
|
||||
|
||||
if (this.options.persistent) {
|
||||
// Server closed gracefully — reopen unless the consumer opted out.
|
||||
await this.connect();
|
||||
} else {
|
||||
this.closeMessageStream();
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// Some environments yield `{ done: false, value: undefined }`.
|
||||
if (!value) continue;
|
||||
|
||||
for (const event of this.options.eventParser.parseEvents(value)) {
|
||||
this.emit('message', event);
|
||||
this.messages.push(event);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// If the controller is different, we already started a new connection and it would be confusing to handle this error.
|
||||
if (controller !== this.controller) return;
|
||||
|
||||
// Invalidate the current controller to allow for reconnection if needed
|
||||
this.controller = null;
|
||||
|
||||
await this.notifyDisconnected();
|
||||
|
||||
// Expected path for abort() — do not treat as an error or reconnect.
|
||||
if (controller.signal.aborted) return;
|
||||
|
||||
await this.notifyError(error);
|
||||
|
||||
if (this.options.attemptReconnect) {
|
||||
await this.connect();
|
||||
} else {
|
||||
this.closeMessageStream();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Clears partial SSE frames left over from an abandoned transport. */
|
||||
private resetEventParser(): void {
|
||||
this.options.eventParser.reset();
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new {@link messages} iterator when the previous one was closed
|
||||
* by a terminal disconnect or server stream end.
|
||||
*/
|
||||
private ensureMessageStreamOpen(): void {
|
||||
if (!this.messages.closed) return;
|
||||
|
||||
this.messages = new AsyncPushIterator<SSEvent>();
|
||||
}
|
||||
|
||||
/** Ends the message iteration loop for the current connection span. */
|
||||
private closeMessageStream(): void {
|
||||
if (this.messages.closed) return;
|
||||
|
||||
this.messages.close();
|
||||
}
|
||||
|
||||
/** Invokes {@link SSESessionOptions.onDisconnected} and emits `"disconnected"`. */
|
||||
private async notifyDisconnected(): Promise<void> {
|
||||
await tryAsync(
|
||||
() => this.options.onDisconnected(),
|
||||
(error) => this.options.onError(error),
|
||||
);
|
||||
this.emit('disconnected', undefined);
|
||||
}
|
||||
|
||||
/** Invokes {@link SSESessionOptions.onError} and emits `"error"`. */
|
||||
private async notifyError(error: unknown): Promise<void> {
|
||||
const errorInstance = error instanceof Error ? error : new Error(String(error));
|
||||
|
||||
await tryAsync(
|
||||
() => this.options.onError(errorInstance),
|
||||
(callbackError) => console.error('SSESession error:', callbackError),
|
||||
);
|
||||
this.emit('error', errorInstance);
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,131 @@
|
||||
export type SSERequestInit = {
|
||||
|
||||
/**
|
||||
* The HTTP method to use.
|
||||
*/
|
||||
method: 'GET' | 'POST';
|
||||
|
||||
/**
|
||||
* Request headers sent on every connect and reconnect.
|
||||
*/
|
||||
headers?: Record<string, string>;
|
||||
|
||||
/**
|
||||
* Request body for POST-based SSE endpoints.
|
||||
*/
|
||||
body?: string | FormData;
|
||||
};
|
||||
|
||||
/**
|
||||
* The fetch function to use.
|
||||
*
|
||||
* NOTE: This is compatible with Browser/Node's native "fetch" function.
|
||||
* We use this in place of "typeof fetch" so that we can accept non-standard URLs ("url" is a "string" here).
|
||||
* For example, a LibP2P adapter might not use a standardized URL format (and might only include "path").
|
||||
* This would cause a type error as native fetch expects type "URL".
|
||||
*/
|
||||
export type SSERequestFunction = {
|
||||
fetch: (url: string, options: RequestInit) => Promise<Response>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Lifecycle hooks invoked by {@link SSESession} during connect, read, and teardown.
|
||||
*/
|
||||
export type SSESessionCallbacks = {
|
||||
|
||||
/**
|
||||
* Called before each fetch so callers can attach auth headers, cookies, or
|
||||
* a `Last-Event-ID` for resume semantics.
|
||||
*/
|
||||
onRequest: (request: RequestInit) => Promise<RequestInit>;
|
||||
|
||||
/**
|
||||
* Called after the HTTP stream is established and before body reading begins.
|
||||
*/
|
||||
onConnected: () => void;
|
||||
|
||||
/**
|
||||
* Called when the active transport ends — including {@link SSESession.abort},
|
||||
* server stream completion, and errors. Not paired with {@link SSESessionCallbacks.onConnected}
|
||||
* when the initial connect never succeeds.
|
||||
*/
|
||||
onDisconnected: () => void;
|
||||
|
||||
/**
|
||||
* Called on fetch or read failures. Not invoked for intentional
|
||||
* {@link SSESession.abort} aborts.
|
||||
*/
|
||||
onError: (error: Error) => void;
|
||||
};
|
||||
|
||||
export type SSESessionRetryInterface = {
|
||||
|
||||
/**
|
||||
* Retry policy used while establishing the HTTP connection in
|
||||
* {@link SSESession.connect}. Defaults to {@link ExponentialBackoff} with
|
||||
* unlimited attempts.
|
||||
*/
|
||||
retry: {
|
||||
run<T>(fn: () => Promise<T>, onError?: (error: Error) => void): Promise<T>;
|
||||
};
|
||||
};
|
||||
|
||||
export interface SSEParser {
|
||||
|
||||
/**
|
||||
* Incremental SSE frame parser for the response body.
|
||||
*
|
||||
* {@link SSEEventParser.reset} is called by the session when abandoning a
|
||||
* transport so partial frames do not carry over to the next connection.
|
||||
*/
|
||||
eventParser: {
|
||||
parseEvents(buffer: Uint8Array): SSEvent[];
|
||||
reset(): void;
|
||||
};
|
||||
}
|
||||
|
||||
export type SSELifecycleOptions = {
|
||||
|
||||
/**
|
||||
* When true, {@link SSESession} calls {@link SSESession.connect} again after
|
||||
* a transport **error** (not an intentional abort).
|
||||
*/
|
||||
attemptReconnect: boolean;
|
||||
|
||||
/**
|
||||
* When true, {@link SSESession} calls {@link SSESession.connect} again after
|
||||
* the **server** closes the stream normally (`done`).
|
||||
*/
|
||||
persistent: boolean;
|
||||
};
|
||||
|
||||
/**
|
||||
* Events emitted by {@link SSESession}.
|
||||
*
|
||||
* - `"connected"` — HTTP stream established.
|
||||
* - `"message"` — A complete SSE event was parsed.
|
||||
* - `"disconnected"` — The active transport ended (including {@link SSESession.abort}).
|
||||
* - `"error"` — An unexpected fetch or read failure.
|
||||
* - `"closed"` — {@link SSESession.disconnect} was called; visibility handling is detached.
|
||||
*/
|
||||
export type SSESessionEventMap = {
|
||||
connected: void;
|
||||
disconnected: void;
|
||||
error: Error;
|
||||
message: SSEvent;
|
||||
closed: void;
|
||||
};
|
||||
|
||||
/**
|
||||
* Configuration for {@link SSESession}.
|
||||
*/
|
||||
export type SSESessionOptions = SSESessionCallbacks
|
||||
& SSERequestInit
|
||||
& SSERequestFunction
|
||||
& SSESessionRetryInterface
|
||||
& SSELifecycleOptions
|
||||
& SSEParser;
|
||||
|
||||
/**
|
||||
* Represents a Server-Sent Event.
|
||||
*/
|
||||
|
||||
65
test/sse-session/helpers/sse-stream.ts
Normal file
65
test/sse-session/helpers/sse-stream.ts
Normal file
@@ -0,0 +1,65 @@
|
||||
export type SseTestStreamOptions = {
|
||||
|
||||
/** Milliseconds to wait before enqueueing each chunk after the first. */
|
||||
chunkDelayMs?: number;
|
||||
|
||||
/**
|
||||
* When true, closes the body as soon as all initial chunks have been sent.
|
||||
* Use this to simulate a server that sends events and then ends the stream.
|
||||
*/
|
||||
closeWhenDone?: boolean;
|
||||
};
|
||||
|
||||
/**
|
||||
* Test double for an SSE HTTP response body.
|
||||
*
|
||||
* Enqueues fixture chunks in order and stays open until {@link close} is called,
|
||||
* matching real servers that keep the connection alive after each event's trailing
|
||||
* `\n\n` frame boundary.
|
||||
*/
|
||||
export class SseTestStream {
|
||||
readonly stream: ReadableStream<Uint8Array>;
|
||||
|
||||
private controller: ReadableStreamDefaultController<Uint8Array> | null = null;
|
||||
|
||||
private closed = false;
|
||||
|
||||
/**
|
||||
* @param chunks - Fixture `raw` strings, whole or split, to simulate chunk boundaries.
|
||||
* @param options - Delivery timing and optional auto-close after the initial chunks.
|
||||
*/
|
||||
constructor(chunks: string[], options: SseTestStreamOptions = {}) {
|
||||
const { chunkDelayMs = 0, closeWhenDone = false } = options;
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
this.stream = new ReadableStream({
|
||||
start: async (controller): Promise<void> => {
|
||||
this.controller = controller;
|
||||
|
||||
for (let i = 0; i < chunks.length; i++) {
|
||||
if (chunkDelayMs > 0 && i > 0) {
|
||||
await new Promise((resolve) => setTimeout(resolve, chunkDelayMs));
|
||||
}
|
||||
|
||||
if (this.closed) return;
|
||||
|
||||
controller.enqueue(encoder.encode(chunks[i]!));
|
||||
}
|
||||
|
||||
if (closeWhenDone) {
|
||||
this.close();
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Ends the HTTP body the way a server closing the SSE connection would.
|
||||
*/
|
||||
close(): void {
|
||||
if (this.closed) return;
|
||||
|
||||
this.closed = true;
|
||||
this.controller?.close();
|
||||
}
|
||||
}
|
||||
633
test/sse-session/sse-session.test.ts
Normal file
633
test/sse-session/sse-session.test.ts
Normal file
@@ -0,0 +1,633 @@
|
||||
import { expect, test, vi, type Mock } from 'vitest';
|
||||
|
||||
import { SSESession } from '../../source/sse-session/sse-session.ts';
|
||||
import { ExponentialBackoff } from '../../source/exponential-backoff.ts';
|
||||
import type { SSESessionOptions, SSEvent } from '../../source/sse-session/types.ts';
|
||||
|
||||
import { SseTestStream, type SseTestStreamOptions } from './helpers/sse-stream.ts';
|
||||
import { priceOracleEvents, storageEvents } from './fixtures/events.fixtures.ts';
|
||||
import { ExponentialBackoffMaxRetriesHitError, HTTPError, ResponseBodyNullError } from '../../source/errors.ts';
|
||||
|
||||
/** URL passed to every session under test. */
|
||||
const EVENTS_URL = '/events';
|
||||
|
||||
/** Headers required for a valid SSE response in these tests. */
|
||||
const SSE_HEADERS = { 'Content-Type': 'text/event-stream' };
|
||||
|
||||
type FetchFn = SSESessionOptions['fetch'];
|
||||
|
||||
/**
|
||||
* Builds a minimal ExponentialBackoff so reconnect and retry paths finish quickly in tests.
|
||||
* Real production delays would make vi.waitFor-based assertions time out.
|
||||
*
|
||||
* @param maxAttempts - Maximum retry attempts; defaults to 1.
|
||||
*/
|
||||
const testRetry = (maxAttempts = 1): ExponentialBackoff => {
|
||||
return new ExponentialBackoff({
|
||||
baseDelay: 1,
|
||||
maxDelay: 1,
|
||||
maxAttempts,
|
||||
growthRate: 1,
|
||||
jitter: 0,
|
||||
});
|
||||
};
|
||||
|
||||
/**
|
||||
* Wraps fixture chunks in a Response backed by {@link SseTestStream}.
|
||||
* SseTestStream simulates a real HTTP body: chunks arrive over time and the stream
|
||||
* can optionally close itself when all chunks are sent.
|
||||
*
|
||||
* @param chunks - Raw SSE payload strings to stream.
|
||||
* @param options - Optional stream timing and close behavior.
|
||||
*/
|
||||
const sseFetchResponse = (chunks: string[], options: SseTestStreamOptions = {}): Response => {
|
||||
return new Response(new SseTestStream(chunks, options).stream, {
|
||||
status: 200,
|
||||
headers: SSE_HEADERS,
|
||||
});
|
||||
};
|
||||
|
||||
/**
|
||||
* Creates a vitest mock fetch that delegates to the given responder.
|
||||
* SSESession requires fetch injection so tests never hit the network.
|
||||
*
|
||||
* @param responder - Function that returns the Response for each fetch call.
|
||||
*/
|
||||
const createFetchMock = (responder: (url: string, init: RequestInit) => Response | Promise<Response>): Mock<FetchFn> => {
|
||||
return vi.fn(async (url: string, init: RequestInit) => responder(url, init));
|
||||
};
|
||||
|
||||
/**
|
||||
* Session defaults that disable reconnect noise unless a test opts in.
|
||||
* attemptReconnect: false — transport errors should not auto-retry by default.
|
||||
* persistent: false — server closing the stream should close the message iterator.
|
||||
*/
|
||||
const defaultSessionOptions: Partial<SSESessionOptions> = {
|
||||
attemptReconnect: false,
|
||||
persistent: false,
|
||||
retry: testRetry(),
|
||||
};
|
||||
|
||||
/**
|
||||
* Creates an SSESession wired to the injected fetch mock.
|
||||
* SSESession.create immediately calls connect(), so fetch is invoked during creation.
|
||||
*
|
||||
* @param fetch - Mock fetch implementation.
|
||||
* @param options - Per-test session overrides merged on top of defaults.
|
||||
*/
|
||||
const createSession = async (fetch: FetchFn, options: Partial<SSESessionOptions> = {}): Promise<SSESession> => {
|
||||
return SSESession.create(EVENTS_URL, {
|
||||
...defaultSessionOptions,
|
||||
onError: vi.fn(),
|
||||
...options,
|
||||
fetch,
|
||||
});
|
||||
};
|
||||
|
||||
/**
|
||||
* Runs a callback against a session and always disconnects afterward.
|
||||
* Most tests use this so session lifecycle (connect on create, disconnect on exit)
|
||||
* is consistent and resources are not leaked between cases.
|
||||
*
|
||||
* @param fetch - Mock fetch for the session.
|
||||
* @param options - Session options.
|
||||
* @param run - Test body receiving the connected session.
|
||||
*/
|
||||
const withSession = async <T>(fetch: FetchFn, options: Partial<SSESessionOptions>, run: (session: SSESession) => Promise<T>): Promise<T> => {
|
||||
const session = await createSession(fetch, options);
|
||||
|
||||
try {
|
||||
return await run(session);
|
||||
} finally {
|
||||
await session.disconnect();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Reads up to `count` messages from the session's async iterator.
|
||||
* Breaking out of the for-await loop early leaves the underlying stream open,
|
||||
* which is intentional for tests that inspect post-read session state.
|
||||
*
|
||||
* @param session - Connected SSE session.
|
||||
* @param count - Number of events to collect before stopping.
|
||||
*/
|
||||
const readMessages = async (session: SSESession, count: number): Promise<SSEvent[]> => {
|
||||
const events: SSEvent[] = [];
|
||||
|
||||
for await (const event of session.messages) {
|
||||
events.push(event);
|
||||
|
||||
if (events.length >= count) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return events;
|
||||
};
|
||||
|
||||
/**
|
||||
* Builds a ReadableStream that emits one chunk then errors after a delay.
|
||||
* Simulates a mid-stream network failure: the client receives partial data,
|
||||
* then the connection drops before the server finishes sending.
|
||||
*
|
||||
* @param raw - SSE payload to enqueue before the error.
|
||||
* @param delayMs - Milliseconds to wait before erroring (gives the parser time to process the chunk).
|
||||
*/
|
||||
const failingStreamAfter = (raw: string, delayMs = 50): ReadableStream<Uint8Array> => {
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
return new ReadableStream({
|
||||
async start(controller): Promise<void> {
|
||||
controller.enqueue(encoder.encode(raw));
|
||||
await new Promise((resolve) => setTimeout(resolve, delayMs));
|
||||
controller.error(new Error('network failure'));
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: a client opens a standard GET SSE connection.
|
||||
* Verifies the session passes the correct fetch options for a spec-compliant SSE request.
|
||||
*/
|
||||
const testSseSessionConnectCallsFetchWithExpectedOptions = async (): Promise<void> => {
|
||||
// Return one valid storage fixture event so connect succeeds and the stream stays open briefly.
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
|
||||
|
||||
try {
|
||||
// withSession → createSession → SSESession.create → connect → fetchMock is called once.
|
||||
await withSession(fetchMock, {}, async () => {
|
||||
expect(fetchMock).toHaveBeenCalledWith(
|
||||
EVENTS_URL,
|
||||
expect.objectContaining({
|
||||
method: 'GET',
|
||||
cache: 'no-store',
|
||||
signal: expect.any(AbortSignal),
|
||||
headers: expect.objectContaining({
|
||||
Accept: 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
}),
|
||||
}),
|
||||
);
|
||||
|
||||
// While the stream is active the abort signal must not yet be triggered.
|
||||
expect(fetchMock.mock.calls[0]?.[1]?.signal?.aborted).toBe(false);
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: a client connects and the application wants a lifecycle callback when the stream is ready.
|
||||
* Verifies onConnected fires after the transport is established.
|
||||
*/
|
||||
const testSseSessionConnectInvokesOnConnected = async (): Promise<void> => {
|
||||
const onConnected = vi.fn();
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
|
||||
|
||||
try {
|
||||
await withSession(fetchMock, { onConnected }, async () => {
|
||||
expect(onConnected).toHaveBeenCalledOnce();
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: a client is already connected and something calls connect() again (e.g. a duplicate init).
|
||||
* Verifies the session does not open a second HTTP transport.
|
||||
*/
|
||||
const testSseSessionConnectDoesNotOpenSecondTransport = async (): Promise<void> => {
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
|
||||
|
||||
try {
|
||||
await withSession(fetchMock, {}, async (session) => {
|
||||
// Pull one event so the first transport is fully established and reading.
|
||||
await readMessages(session, 1);
|
||||
|
||||
// Idempotent connect — should be a no-op at the fetch layer.
|
||||
await session.connect();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledOnce();
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: the application attaches auth or other headers via onRequest before each fetch.
|
||||
* Verifies mutations from onRequest reach the actual fetch call.
|
||||
*/
|
||||
const testSseSessionConnectPassesOnRequestMutations = async (): Promise<void> => {
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
|
||||
|
||||
try {
|
||||
await withSession(
|
||||
fetchMock,
|
||||
{
|
||||
// onRequest runs during connect and can rewrite headers/body before fetch sees them.
|
||||
onRequest: async (request) => ({
|
||||
...request,
|
||||
headers: { ...request.headers, Authorization: 'Bearer test-token' },
|
||||
}),
|
||||
},
|
||||
async () => {
|
||||
expect(fetchMock).toHaveBeenCalledWith(
|
||||
EVENTS_URL,
|
||||
expect.objectContaining({
|
||||
headers: expect.objectContaining({
|
||||
Authorization: 'Bearer test-token',
|
||||
}),
|
||||
}),
|
||||
);
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: some SSE endpoints require POST with a form body instead of a plain GET.
|
||||
* Verifies method and body are forwarded to fetch.
|
||||
*/
|
||||
const testSseSessionConnectSendsPostBody = async (): Promise<void> => {
|
||||
const body = new FormData();
|
||||
body.set('topic', 'prices');
|
||||
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
|
||||
|
||||
try {
|
||||
await withSession(
|
||||
fetchMock,
|
||||
{
|
||||
method: 'POST',
|
||||
body,
|
||||
},
|
||||
async () => {
|
||||
expect(fetchMock).toHaveBeenCalledWith(
|
||||
EVENTS_URL,
|
||||
expect.objectContaining({
|
||||
method: 'POST',
|
||||
body,
|
||||
}),
|
||||
);
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: a single long-lived connection delivers many events from different domains (LLM, oracle, storage).
|
||||
* Verifies the parser and session deliver every fixture event in order through one stream.
|
||||
*/
|
||||
const testSseSessionDeliversMultipleFixtureEvents = async (): Promise<void> => {
|
||||
const fixtures = [ ...priceOracleEvents, ...storageEvents ];
|
||||
const expected = fixtures.flatMap(({ parsed }) => parsed ?? []);
|
||||
|
||||
// All raw payloads are concatenated into one SseTestStream response.
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse(fixtures.map(({ raw }) => raw)));
|
||||
|
||||
try {
|
||||
await withSession(fetchMock, {}, async (session) => {
|
||||
const received = await readMessages(session, expected.length);
|
||||
|
||||
expect(received).toEqual(expected);
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: application code listens via session.on('message') instead of the async iterator.
|
||||
* Verifies the EventEmitter path receives the same parsed events as the iterator.
|
||||
*/
|
||||
const testSseSessionEmitsMessageEvents = async (): Promise<void> => {
|
||||
const { raw, parsed } = storageEvents[0]!;
|
||||
const emitted: SSEvent[] = [];
|
||||
|
||||
// ':\n\n' is an SSE comment/heartbeat; chunkDelayMs forces it to arrive as a separate chunk
|
||||
// so incremental parsing is exercised before the real data frame.
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ ':\n\n', raw ], { chunkDelayMs: 50 }));
|
||||
|
||||
// Use createSession directly (not withSession) so we control disconnect timing in finally.
|
||||
const session = await createSession(fetchMock, {});
|
||||
|
||||
try {
|
||||
session.on('message', (event) => emitted.push(event));
|
||||
|
||||
await readMessages(session, 1);
|
||||
|
||||
expect(emitted).toEqual([ parsed![0] ]);
|
||||
} finally {
|
||||
await session.disconnect();
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: network chunks split an SSE frame at an arbitrary byte boundary.
|
||||
* Verifies the internal parser buffers partial data and still emits a complete event.
|
||||
*/
|
||||
const testSseSessionParsesEventSplitAcrossChunks = async (): Promise<void> => {
|
||||
const { raw, parsed } = storageEvents[0]!;
|
||||
const mid = Math.floor(raw.length / 2);
|
||||
|
||||
// SseTestStream sends each array element as a separate chunk — the event is cut in half.
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ raw.slice(0, mid), raw.slice(mid) ]));
|
||||
|
||||
try {
|
||||
await withSession(fetchMock, {}, async (session) => {
|
||||
const [ event ] = await readMessages(session, 1);
|
||||
|
||||
expect(event).toEqual(parsed![0]);
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: server closes the stream after one event and the client is not in persistent mode.
|
||||
* Verifies the message iterator closes and onDisconnected fires.
|
||||
*/
|
||||
const testSseSessionClosesOnServerCloseWhenNotPersistent = async (): Promise<void> => {
|
||||
// closeWhenDone: true makes SseTestStream end the body after sending the chunk.
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }));
|
||||
const disconnected = vi.fn();
|
||||
|
||||
try {
|
||||
// persistent defaults to false — terminal server close should shut down messages.
|
||||
await withSession(fetchMock, { onDisconnected: disconnected }, async (session) => {
|
||||
await readMessages(session, 1);
|
||||
|
||||
await vi.waitFor(() => expect(session.messages.closed).toBe(true));
|
||||
expect(disconnected).toHaveBeenCalled();
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: server closes the stream, the message iterator closes, then the app manually reconnects.
|
||||
* Verifies a second fetch opens and events flow again (non-persistent manual reconnect path).
|
||||
*/
|
||||
const testSseSessionReconnectsAfterTerminalClose = async (): Promise<void> => {
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }));
|
||||
|
||||
// Override the default responder: first fetch closes after one event, second stays open.
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }))
|
||||
.mockResolvedValueOnce(sseFetchResponse([ storageEvents[0]!.raw ]));
|
||||
|
||||
try {
|
||||
await withSession(fetchMock, {}, async (session) => {
|
||||
await readMessages(session, 1);
|
||||
|
||||
await vi.waitFor(() => expect(session.messages.closed).toBe(true));
|
||||
|
||||
// App-initiated reconnect after the first stream ended.
|
||||
await session.connect();
|
||||
|
||||
const [ event ] = await readMessages(session, 1);
|
||||
expect(event).toEqual(storageEvents[0]!.parsed![0]);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: persistent client — server closes the stream but the session should auto-reconnect.
|
||||
* Verifies two fetch calls and events from both streams arrive on the same message iterator.
|
||||
*/
|
||||
const testSseSessionPersistentReconnectsOnServerClose = async (): Promise<void> => {
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ priceOracleEvents[0]!.raw ], { closeWhenDone: true }));
|
||||
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[0]!.raw ], { closeWhenDone: true }))
|
||||
.mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ]));
|
||||
|
||||
try {
|
||||
await withSession(
|
||||
fetchMock,
|
||||
{
|
||||
persistent: true,
|
||||
retry: testRetry(2),
|
||||
},
|
||||
async (session) => {
|
||||
const [ first, second ] = await readMessages(session, 2);
|
||||
|
||||
expect(first).toEqual(priceOracleEvents[0]!.parsed![0]);
|
||||
expect(second).toEqual(priceOracleEvents[1]!.parsed![0]);
|
||||
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: mid-stream network error with attemptReconnect enabled.
|
||||
* Verifies the session fetches again and the iterator continues delivering events.
|
||||
*/
|
||||
const testSseSessionReconnectsOnTransportError = async (): Promise<void> => {
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ priceOracleEvents[0]!.raw ]));
|
||||
|
||||
fetchMock
|
||||
// First connection: delivers one event then the stream errors (failingStreamAfter).
|
||||
.mockResolvedValueOnce(new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }))
|
||||
// Second connection: clean stream with the next fixture event.
|
||||
.mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ]));
|
||||
|
||||
try {
|
||||
await withSession(
|
||||
fetchMock,
|
||||
{
|
||||
attemptReconnect: true,
|
||||
retry: testRetry(2),
|
||||
},
|
||||
async (session) => {
|
||||
const [ beforeError, recovered ] = await readMessages(session, 2);
|
||||
|
||||
expect(beforeError).toEqual(priceOracleEvents[0]!.parsed![0]);
|
||||
expect(recovered).toEqual(priceOracleEvents[1]!.parsed![0]);
|
||||
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: mid-stream network error with attemptReconnect disabled (default).
|
||||
* Verifies the session terminates: messages close, onError runs, and an error event is emitted.
|
||||
*/
|
||||
const testSseSessionClosesOnTransportErrorWhenNotReconnecting = async (): Promise<void> => {
|
||||
const onError = vi.fn();
|
||||
const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS });
|
||||
const fetchMock = createFetchMock(() => response);
|
||||
|
||||
try {
|
||||
await withSession(fetchMock, { onError }, async (session) => {
|
||||
const errorEvent = new Promise<Error>((resolve) => session.once('error', resolve));
|
||||
|
||||
await vi.waitFor(() => expect(session.messages.closed).toBe(true));
|
||||
expect(onError).toHaveBeenCalled();
|
||||
await expect(errorEvent).resolves.toBeInstanceOf(Error);
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: the user aborts an active connection (e.g. navigation away) but may reconnect later.
|
||||
* Verifies onDisconnected fires, fetch is aborted, but the message iterator stays open.
|
||||
*/
|
||||
const testSseSessionAbortEmitsDisconnectedAndKeepsMessagesOpen = async (): Promise<void> => {
|
||||
const onDisconnected = vi.fn();
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
|
||||
|
||||
try {
|
||||
await withSession(fetchMock, { onDisconnected }, async (session) => {
|
||||
await readMessages(session, 1);
|
||||
await session.abort();
|
||||
|
||||
expect(onDisconnected).toHaveBeenCalled();
|
||||
expect(session.messages.closed).toBe(false);
|
||||
});
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: the application fully tears down the session (logout, component unmount, etc.).
|
||||
* Verifies disconnect closes the message iterator and emits the closed event.
|
||||
*/
|
||||
const testSseSessionDisconnectClosesMessagesAndEmitsClosed = async (): Promise<void> => {
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));
|
||||
|
||||
const session = await createSession(fetchMock, {});
|
||||
const closed = new Promise<void>((resolve) => session.once('closed', () => resolve()));
|
||||
|
||||
try {
|
||||
await session.disconnect();
|
||||
|
||||
expect(session.messages.closed).toBe(true);
|
||||
await expect(closed).resolves.toBeUndefined();
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: the server returns a non-2xx HTTP status (500).
|
||||
* Verifies create rejects, onError is invoked, and the error is a real Error instance.
|
||||
*/
|
||||
const testSseSessionHttpErrorCallsOnErrorAndThrows = async (): Promise<void> => {
|
||||
const onError = vi.fn();
|
||||
const fetchMock = createFetchMock(() =>
|
||||
new Response('nope', {
|
||||
status: 500,
|
||||
statusText: 'Internal Server Error',
|
||||
}));
|
||||
|
||||
const expectedError = new ExponentialBackoffMaxRetriesHitError([ new HTTPError(500, 'nope') ]);
|
||||
|
||||
try {
|
||||
await expect(createSession(fetchMock, { onError })).rejects.toThrow(expectedError);
|
||||
|
||||
expect(onError).toHaveBeenCalled();
|
||||
expect(onError.mock.calls[0]?.[0]).toBeInstanceOf(Error);
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: fetch returns 200 but with a null body (misconfigured proxy or server bug).
|
||||
* Verifies create rejects because SSE requires a readable stream body.
|
||||
*/
|
||||
const testSseSessionRejectsWhenResponseBodyIsNull = async (): Promise<void> => {
|
||||
const onError = vi.fn();
|
||||
const fetchMock = createFetchMock(() => new Response(null, { status: 200, headers: SSE_HEADERS }));
|
||||
|
||||
const expectedError = new ExponentialBackoffMaxRetriesHitError([ new ResponseBodyNullError() ]);
|
||||
|
||||
try {
|
||||
await expect(createSession(fetchMock, { onError, retry: testRetry() })).rejects.toThrow(expectedError);
|
||||
|
||||
expect(onError).toHaveBeenCalled();
|
||||
} finally {
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Scenario: SSE spec resume — after receiving an event with an id, reconnect should send Last-Event-ID.
|
||||
* Verifies the second fetch includes the id from the first event (1234 in storageEvents[0]).
|
||||
*/
|
||||
const testSseSessionSendsLastEventIdOnReconnect = async (): Promise<void> => {
|
||||
const { raw } = storageEvents[0]!;
|
||||
let reconnectHeaders: Record<string, string> | undefined;
|
||||
const fetchMock = createFetchMock(() => sseFetchResponse([ raw ]));
|
||||
|
||||
fetchMock
|
||||
// First connection: heartbeat chunk, then the event (with id: 1234), then server closes.
|
||||
.mockResolvedValueOnce(sseFetchResponse([ ':\n\n', raw ], { chunkDelayMs: 10, closeWhenDone: true }))
|
||||
// Second connection: capture whatever headers the reconnect logic attached.
|
||||
.mockImplementationOnce(async (_url, init) => {
|
||||
reconnectHeaders = init.headers as Record<string, string>;
|
||||
|
||||
return sseFetchResponse([]);
|
||||
});
|
||||
|
||||
const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2) });
|
||||
|
||||
try {
|
||||
// Registers an onRequest hook that copies the last seen event id into reconnect headers.
|
||||
await SSESession.addLastEventIdReconnect(session);
|
||||
|
||||
await readMessages(session, 1);
|
||||
|
||||
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
|
||||
expect(reconnectHeaders?.['Last-Event-ID']).toBe('1234');
|
||||
} finally {
|
||||
await session.disconnect();
|
||||
vi.restoreAllMocks();
|
||||
}
|
||||
};
|
||||
|
||||
const runTests = async (): Promise<void> => {
|
||||
test('SSESession.connect: calls injected fetch with method, headers, and abort signal', testSseSessionConnectCallsFetchWithExpectedOptions);
|
||||
test('SSESession.connect: invokes onConnected when the stream is established', testSseSessionConnectInvokesOnConnected);
|
||||
test('SSESession.connect: does not open a second transport when connect is called again', testSseSessionConnectDoesNotOpenSecondTransport);
|
||||
test('SSESession.connect: passes request mutations from onRequest to fetch', testSseSessionConnectPassesOnRequestMutations);
|
||||
test('SSESession.connect: sends POST bodies for POST-based SSE endpoints', testSseSessionConnectSendsPostBody);
|
||||
test('SSESession: delivers multiple fixture events through a single session', testSseSessionDeliversMultipleFixtureEvents);
|
||||
test('SSESession: emits message events for incoming SSE frames', testSseSessionEmitsMessageEvents);
|
||||
test('SSESession: parses a fixture event split across chunk boundaries', testSseSessionParsesEventSplitAcrossChunks);
|
||||
test('SSESession: closes messages and emits disconnected when persistent is false', testSseSessionClosesOnServerCloseWhenNotPersistent);
|
||||
test('SSESession: opens a new message stream after reconnecting following a terminal close', testSseSessionReconnectsAfterTerminalClose);
|
||||
test('SSESession: reconnects when the server closes the stream and persistent is true', testSseSessionPersistentReconnectsOnServerClose);
|
||||
test('SSESession: reconnects when attemptReconnect is true', testSseSessionReconnectsOnTransportError);
|
||||
test('SSESession: closes messages and emits error when attemptReconnect is false', testSseSessionClosesOnTransportErrorWhenNotReconnecting);
|
||||
test('SSESession.abort: aborts fetch, emits disconnected, and keeps messages open', testSseSessionAbortEmitsDisconnectedAndKeepsMessagesOpen);
|
||||
test('SSESession.disconnect: closes messages and emits closed', testSseSessionDisconnectClosesMessagesAndEmitsClosed);
|
||||
test('SSESession: calls onError, emits error, and throws from create on HTTP error', testSseSessionHttpErrorCallsOnErrorAndThrows);
|
||||
test('SSESession: rejects when the response body is null', testSseSessionRejectsWhenResponseBodyIsNull);
|
||||
test('SSESession: sends Last-Event-ID on reconnect after receiving an event with an id', testSseSessionSendsLastEventIdOnReconnect);
|
||||
};
|
||||
|
||||
await runTests();
|
||||
Reference in New Issue
Block a user