SSE Session and examples
This commit is contained in:
Kuldeep
@@ -0,0 +1,635 @@
|
||||
import type { SSESessionOptions, SSESessionEventMap, SSEvent, SSESessionEventCallbacks, SSESessionOperationContext } from './types.ts';
|
||||
import { SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS, SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT, SSE_SESSION_PERSISTENT_DEFAULT } from './constants.ts';
|
||||
|
||||
import { EventEmitter, type OffCallback } from '../event-emitter.ts';
|
||||
import { ExponentialBackoff } from '../exponential-backoff/index.ts';
|
||||
import { normalizeError } from '../misc.ts';
|
||||
|
||||
import { SSEEventParser } from './sse-event-parser.ts';
|
||||
import { AsyncPushIterator } from './async-push-iterator.ts';
|
||||
import { HTTPError, ResponseBodyNullError, SSESessionUnsubscribePluginNotAFunctionError } from './errors.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 disconnect} 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 close} aborts the transport, closes {@link messages}, and emits
|
||||
* `"closed"`.
|
||||
*
|
||||
* Automatic reconnect is controlled by {@link SSESessionOptions.persistent}
|
||||
* (server closed the stream) and
|
||||
* {@link SSESessionOptions.attemptReconnect} (transport error).
|
||||
*
|
||||
* ## Connection supersession
|
||||
*
|
||||
* Each {@link connect} will create a new controller if one does not exist. Otherwise,
|
||||
* it will return immediately.
|
||||
*/
|
||||
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 & SSESessionEventCallbacks> = {}): Promise<SSESession> {
|
||||
const session = new SSESession(url, options);
|
||||
await session.connect();
|
||||
|
||||
return session;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*
|
||||
* Since this relies on listening to the `"message"` event, this should ideally
|
||||
* be called before "{@link connect} is called so no messages (and no ids) are missed.
|
||||
*
|
||||
* ```ts
|
||||
* const session = new SSESession(url);
|
||||
* SSESession.addLastEventIdReconnect(session);
|
||||
* await session.connect();
|
||||
* // Reconnects send Last-Event-ID once an event with an id is received.
|
||||
* ```
|
||||
*
|
||||
* @param session - The session to instrument.
|
||||
* @returns The same session, for chaining.
|
||||
*/
|
||||
static addLastEventIdReconnect(session: SSESession): { session: SSESession; removeListener?: OffCallback } {
|
||||
// If the plugin is already installed, we return the session and the remove listener function.
|
||||
if (session.plugins.has(SSESession.addLastEventIdReconnect)) {
|
||||
// Get the remove listener function from the session.
|
||||
const removeListener = session.plugins.get(SSESession.addLastEventIdReconnect) as OffCallback;
|
||||
|
||||
// If the plugin was added with a non-function listener, we throw an error.
|
||||
if (typeof removeListener !== 'function') {
|
||||
throw new SSESessionUnsubscribePluginNotAFunctionError('addLastEventIdReconnect');
|
||||
}
|
||||
|
||||
// If the plugin is already installed, we return the session and the remove listener function.
|
||||
return { session, removeListener };
|
||||
}
|
||||
|
||||
// Store the last event id so we can send it on reconnect.
|
||||
let lastEventId: string | undefined;
|
||||
|
||||
// Define the message listener that will store the last event id and begin listening for 'message' events.
|
||||
const messageListener = (event: SSEvent): void => {
|
||||
// If the event has no id, we should not update the lastEventId and instead keep the old one.
|
||||
lastEventId = event.id ?? lastEventId;
|
||||
};
|
||||
|
||||
session.on('message', messageListener);
|
||||
|
||||
// Store the original onRequest callback so we can extend its behavior to write the reconnect header.
|
||||
const originalOnRequest = session.options.onRequest;
|
||||
|
||||
// Whether we actively send the Last-Event-ID header on reconnect.
|
||||
// Because the `onRequest` callback may be extended further by other code, we can't just roll-back to the `originalOnRequest`
|
||||
// Instead, we just use this flag to determine if we should write the Last-Event-ID header.
|
||||
let enabled = true;
|
||||
|
||||
// Extend the onRequest callback to write the Last-Event-ID header if it is enabled.
|
||||
session.options.onRequest = async (request: RequestInit): Promise<RequestInit> => {
|
||||
if (lastEventId && enabled) {
|
||||
request.headers = { ...request.headers, 'Last-Event-ID': lastEventId };
|
||||
}
|
||||
|
||||
return originalOnRequest(request);
|
||||
};
|
||||
|
||||
// Define the remove listener function that will disable the Last-Event-ID header on reconnect.
|
||||
const removeListener = (): void => {
|
||||
enabled = false;
|
||||
session.off('message', messageListener);
|
||||
|
||||
// If the remove listener function is the same as the existing remove listener function, we remove the plugin from the session.
|
||||
// This protects against the `removeListener` function being called onto after a new instance of this plugin is added
|
||||
const existingRemoveListener = session.plugins.get(SSESession.addLastEventIdReconnect) as OffCallback;
|
||||
if (existingRemoveListener === removeListener) {
|
||||
session.plugins.delete(SSESession.addLastEventIdReconnect);
|
||||
}
|
||||
};
|
||||
|
||||
// Store the plugin in the session.
|
||||
session.plugins.set(SSESession.addLastEventIdReconnect, removeListener);
|
||||
|
||||
return { session, removeListener };
|
||||
}
|
||||
|
||||
/**
|
||||
* Pauses and resumes a session based on browser tab visibility.
|
||||
*
|
||||
* Uses the Page Visibility API (`document.visibilitychange`):
|
||||
*
|
||||
* - **hidden** — {@link disconnect} stops the active fetch. {@link messages}
|
||||
* stays open; `"disconnected"` fires but `"closed"` does not.
|
||||
* - **visible** — {@link connect} re-establishes the stream if needed.
|
||||
*
|
||||
* This controller will not re-connect if the session was disconnected by something else.
|
||||
*
|
||||
* No-op in non-browser environments where `document` is undefined.
|
||||
*
|
||||
* @param session - The session to manage.
|
||||
*/
|
||||
static addBrowserVisibilityHandler(session: SSESession): { session: SSESession; removeListener?: OffCallback } {
|
||||
if (typeof document === 'undefined') return { session };
|
||||
|
||||
// If the plugin is already installed, we return the session and the remove listener function.
|
||||
if (session.plugins.has(SSESession.addBrowserVisibilityHandler)) {
|
||||
// Get the remove listener function from the session.
|
||||
const removeListener = session.plugins.get(SSESession.addBrowserVisibilityHandler) as OffCallback;
|
||||
|
||||
// If the plugin was added with a non-function listener, we throw an error.
|
||||
if (typeof removeListener !== 'function') {
|
||||
throw new SSESessionUnsubscribePluginNotAFunctionError('addBrowserVisibilityHandler');
|
||||
}
|
||||
|
||||
// If the plugin is already installed, we return the session and the remove listener function.
|
||||
return { session, removeListener };
|
||||
}
|
||||
|
||||
// Create a listener that will reconnect the session
|
||||
// This listener is added when the hidden->disconnect listener disconnects the session, but will not be added
|
||||
// if the session was disconnected by something else, like a manual disconnect, server-disconnect, or another plugin.
|
||||
const reconnectOnVisible = async (): Promise<void> => {
|
||||
if (document.visibilityState === 'visible') {
|
||||
// Remove this listener
|
||||
document.removeEventListener('visibilitychange', reconnectOnVisible);
|
||||
|
||||
// Reconnect the session
|
||||
await session.connect();
|
||||
}
|
||||
};
|
||||
|
||||
// Create a listener that will disable the reconnect listener if the session is disconnected by something else.
|
||||
const disableOnExternalDisconnect = ({ source }: SSESessionOperationContext): void => {
|
||||
// If the source is not the browser visibility handler, we remove the listener.
|
||||
if (source !== SSESession.addBrowserVisibilityHandler) {
|
||||
document.removeEventListener('visibilitychange', reconnectOnVisible);
|
||||
}
|
||||
};
|
||||
|
||||
// Create a listener that will live as-long as this session is not closed
|
||||
// This listener will create a new 'on visible' listener that will reconnect the session when it disconnects the session.
|
||||
// This prevents it from trying to reconnect the session if the session was disconnected by something else
|
||||
const handleVisibilityChange = async (): Promise<void> => {
|
||||
// If the session is not connected, we do nothing.
|
||||
if (!session.active) {
|
||||
return;
|
||||
}
|
||||
|
||||
// If the session is connected and the visibility is hidden, we disconnect it.
|
||||
if (document.visibilityState === 'hidden') {
|
||||
// Add the listener for connected event to reconnect the session.
|
||||
document.addEventListener('visibilitychange', reconnectOnVisible);
|
||||
|
||||
// Disconnect the session.
|
||||
session.disconnect({ source: SSESession.addBrowserVisibilityHandler });
|
||||
}
|
||||
};
|
||||
|
||||
// Add the listener for visibility change.
|
||||
document.addEventListener('visibilitychange', handleVisibilityChange);
|
||||
|
||||
// Add a listener for `beforeDisconnect` event to disable the reconnect listener if the session is disconnected by something else.
|
||||
session.on('beforeDisconnect', disableOnExternalDisconnect);
|
||||
|
||||
// Remove the listener for visibility change.
|
||||
const removeListener = (): void => {
|
||||
document.removeEventListener('visibilitychange', handleVisibilityChange);
|
||||
document.removeEventListener('visibilitychange', reconnectOnVisible);
|
||||
|
||||
session.off('beforeDisconnect', disableOnExternalDisconnect);
|
||||
|
||||
// If the remove listener function is the same as the existing remove listener function, we remove the plugin from the session.
|
||||
// This protects against the `removeListener` function being called onto after a new instance of this plugin is added
|
||||
const existingRemoveListener = session.plugins.get(SSESession.addBrowserVisibilityHandler) as OffCallback;
|
||||
if (existingRemoveListener === removeListener) {
|
||||
session.plugins.delete(SSESession.addBrowserVisibilityHandler);
|
||||
}
|
||||
};
|
||||
|
||||
// Store the plugin in the session.
|
||||
session.plugins.set(SSESession.addBrowserVisibilityHandler, removeListener);
|
||||
|
||||
// Return the session and the remove listener function.
|
||||
return { session, removeListener };
|
||||
}
|
||||
|
||||
/** SSE endpoint URL for this session. */
|
||||
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 options
|
||||
fetch: (...args) => fetch(...args),
|
||||
method: 'GET',
|
||||
headers: {
|
||||
Accept: 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
},
|
||||
body: new FormData(),
|
||||
|
||||
// callback to mutate the fetch options before the fetch is made.
|
||||
onRequest: (request) => Promise.resolve(request),
|
||||
|
||||
// Retry the initial fetch.
|
||||
// Use the spread operator so it's impossible for us to override the default options. (default is exponential backoff)
|
||||
retry: new ExponentialBackoff({
|
||||
...SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS,
|
||||
}),
|
||||
|
||||
// Reconnection options
|
||||
attemptReconnect: SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT,
|
||||
persistent: SSE_SESSION_PERSISTENT_DEFAULT,
|
||||
|
||||
// Event parser options
|
||||
eventParser: new SSEEventParser(),
|
||||
};
|
||||
|
||||
/**
|
||||
* Registered plugins for the session.
|
||||
* This can be used to prevent duplicate plugins from being added to the session.
|
||||
* The value for a plugin is arbitrary, for example a function that removes the plugin.
|
||||
*/
|
||||
public readonly plugins: Map<unknown, unknown> = new Map();
|
||||
|
||||
/** AbortController for the currently active fetch, if any. */
|
||||
#connectionController: AbortController | null = null;
|
||||
|
||||
/** The server's requested retry interval in milliseconds (per SSE spec) if any */
|
||||
#retryInterval: number | null = null;
|
||||
|
||||
/**
|
||||
* Asynchronous stream of parsed SSE events for the active connection.
|
||||
*
|
||||
* Stays open across {@link disconnect} 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 close} 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 & SSESessionEventCallbacks> = {}) {
|
||||
super();
|
||||
|
||||
const { onConnected, onDisconnected, onError, ...restOptions } = options;
|
||||
|
||||
this.#url = url;
|
||||
this.options = {
|
||||
...this.options,
|
||||
...restOptions,
|
||||
// Shallow merge would drop default headers when options.headers is set.
|
||||
headers: { ...this.options.headers, ...options.headers },
|
||||
};
|
||||
|
||||
// Add the onConnected, onDisconnected, and onError as event listeners to the session.
|
||||
// We accept these as options to the constructor so that we can both instantiate the session and add the event listeners at the same time,
|
||||
// avoiding any potential race-conditions or async/await issues.
|
||||
if (onConnected) {
|
||||
this.on('connected', onConnected);
|
||||
}
|
||||
|
||||
if (onDisconnected) {
|
||||
this.on('disconnected', onDisconnected);
|
||||
}
|
||||
|
||||
if (onError) {
|
||||
this.on('error', onError);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns true if the session is active by checking if the controller is not null.
|
||||
*/
|
||||
public get active(): boolean {
|
||||
return this.#connectionController !== null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 the
|
||||
* internal `#readStream` method.
|
||||
*
|
||||
* @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.#connectionController) 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 connectionController = new AbortController();
|
||||
this.#connectionController = connectionController;
|
||||
|
||||
const { method, headers, body } = this.options;
|
||||
|
||||
// Create the fetch options.
|
||||
const fetchOptions: RequestInit = {
|
||||
method,
|
||||
headers: headers || {},
|
||||
signal: connectionController.signal,
|
||||
cache: 'no-store',
|
||||
};
|
||||
|
||||
// If the method is POST, we will set the body of the fetch options.
|
||||
// NOTE: RequestInit doesn't allow body to be `undefined`, so if its false-y we will set it to null.
|
||||
if (method === 'POST') {
|
||||
fetchOptions.body = body || null;
|
||||
}
|
||||
|
||||
// Create a reader for the response body.
|
||||
let reader: ReadableStreamDefaultReader<Uint8Array>;
|
||||
|
||||
try {
|
||||
// Retry the fetch using the provided retry policy. (default is exponential backoff)
|
||||
reader = await this.options.retry.run(() => this.#createReader(fetchOptions), {
|
||||
signal: connectionController.signal,
|
||||
});
|
||||
} catch (error) {
|
||||
// A newer `connect()` superseded this attempt — leave state to the winner.
|
||||
if (this.#connectionController !== connectionController) return;
|
||||
|
||||
// Reset the controller to null to allow for reconnection if needed.
|
||||
this.#connectionController = null;
|
||||
|
||||
// Normalize the error so it can be passed to the event listeners.
|
||||
const normalizedError = normalizeError(error);
|
||||
|
||||
// Emit the disconnected and error events.
|
||||
this.emit('disconnected', { reason: 'error', error: normalizedError });
|
||||
this.emit('error', normalizedError);
|
||||
|
||||
// Close the message stream so `for await...of` consumers resolve.
|
||||
this.#closeMessageStream();
|
||||
|
||||
// Throw the error so the caller can handle it.
|
||||
throw error;
|
||||
}
|
||||
|
||||
// Connection succeeded but was already replaced (for example disconnect during fetch).
|
||||
if (this.#connectionController !== connectionController) {
|
||||
await reader.cancel();
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
this.emit('connected', undefined);
|
||||
|
||||
// Fire-and-forget: connect() resolves while the stream is consumed.
|
||||
// Data is emitted as a `message` event and from the `messages` iterator.
|
||||
this.#readStream(reader, connectionController).catch((error) => {
|
||||
this.emit('error', error);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Disconnects only the currently active transport.
|
||||
*
|
||||
* `beforeDisconnect` is emitted on every invocation, even when no
|
||||
* transport is currently active. This allows observers to react to
|
||||
* an explicit disconnect operation.
|
||||
*
|
||||
* @param context - Context describing the operation initiator.
|
||||
* @emits `beforeDisconnect` with the supplied operation context.
|
||||
* @emits `disconnected` with reason `"disconnect"` when an active
|
||||
* transport is actually disconnected.
|
||||
*/
|
||||
public disconnect(context: SSESessionOperationContext = {}): void {
|
||||
this.emit('beforeDisconnect', context);
|
||||
|
||||
if (this.#connectionController) {
|
||||
// Grab the current controller to ensure we are aborting the correct one.
|
||||
const connectionController = this.#connectionController;
|
||||
this.#connectionController = null;
|
||||
|
||||
// Invalidate any in-flight read loop and fetch for this transport.
|
||||
connectionController.abort();
|
||||
|
||||
// Emit the disconnected event.
|
||||
this.emit('disconnected', {
|
||||
reason: 'disconnect',
|
||||
source: context.source,
|
||||
});
|
||||
}
|
||||
|
||||
// Reset the event parser buffer so data is not carried through resulting in malformed events.
|
||||
this.#resetEventParser();
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 close(): void {
|
||||
// Disconnect the session.
|
||||
this.disconnect();
|
||||
|
||||
// Close the message stream so `for await...of` consumers resolve.
|
||||
this.#closeMessageStream();
|
||||
|
||||
// Emit the closed event.
|
||||
this.emit('closed', undefined);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
async #createReader(fetchOptions: RequestInit): Promise<ReadableStreamDefaultReader<Uint8Array>> {
|
||||
const requestOptions = await this.options.onRequest(fetchOptions);
|
||||
const response = await this.options.fetch(this.#url, requestOptions);
|
||||
|
||||
// Handle a bad response from the server
|
||||
if (!response.ok) {
|
||||
// Get the response code and text.
|
||||
const responseCode = response.status;
|
||||
const responseText = await response.text();
|
||||
|
||||
// Create a new error and emit it.
|
||||
const error = new HTTPError(responseCode, responseText);
|
||||
this.emit('error', error);
|
||||
|
||||
// Throw the error so the caller can handle it.
|
||||
throw error;
|
||||
}
|
||||
|
||||
// Handle a response with no body (SSE still uses the response body)
|
||||
if (!response.body) {
|
||||
const error = new ResponseBodyNullError();
|
||||
this.emit('error', error);
|
||||
throw error;
|
||||
}
|
||||
|
||||
// Return the reader for the response body.
|
||||
return response.body.getReader();
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads bytes from an established stream until it ends, errors, or is
|
||||
* superseded by a newer connection.
|
||||
*/
|
||||
async #readStream(reader: ReadableStreamDefaultReader<Uint8Array>, connectionController: AbortController): Promise<void> {
|
||||
try {
|
||||
while (this.#connectionController === connectionController) {
|
||||
const { done, value } = await reader.read();
|
||||
|
||||
// `disconnect()` or a newer connect() may have landed while we were awaiting.
|
||||
if (this.#connectionController !== connectionController) return;
|
||||
|
||||
// If the stream is done, we've been disconnected.
|
||||
// We need to clean up and either complete or reconnect the session.
|
||||
if (done) {
|
||||
// Reset the controller to null to allow for reconnection if needed.
|
||||
this.#connectionController = null;
|
||||
|
||||
// Emit the disconnected event.
|
||||
this.emit('disconnected', { reason: 'remote' });
|
||||
|
||||
// If the session is persistent, we will attempt to reconnect.
|
||||
// NOTE: in very rare cases, a malfunctioning server can make this code turn into an infinite loop.
|
||||
if (this.options.persistent) {
|
||||
// Server closed gracefully — reopen unless the consumer opted out.
|
||||
await this.connect();
|
||||
} else {
|
||||
// Close the message stream so `for await...of` consumers resolve.
|
||||
this.#closeMessageStream();
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// Some environments yield `{ done: false, value: undefined }`.
|
||||
if (!value) continue;
|
||||
|
||||
// Emit the parsed events to the event listeners and the messages iterator.
|
||||
for (const event of this.options.eventParser.parseEvents(value)) {
|
||||
// If the event has a retry interval, we will store it.
|
||||
if (event.retry) {
|
||||
this.#retryInterval = event.retry;
|
||||
}
|
||||
|
||||
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.
|
||||
// The controller can also be null here if the session was disconnected.
|
||||
if (connectionController !== this.#connectionController) return;
|
||||
|
||||
// Invalidate the current controller to allow for reconnection if needed
|
||||
this.#connectionController = null;
|
||||
|
||||
const normalizedError = normalizeError(error);
|
||||
this.emit('disconnected', { reason: 'error', error: normalizedError });
|
||||
|
||||
// Expected path for `disconnect()` — do not treat as an error or reconnect.
|
||||
if (connectionController.signal.aborted) return;
|
||||
|
||||
this.emit('error', normalizedError);
|
||||
|
||||
// If the session is configured to attempt reconnect, we will attempt to reconnect.
|
||||
if (this.options.attemptReconnect) {
|
||||
// If the server has requested a retry interval, we will wait for it before attempting to reconnect.
|
||||
if (this.#retryInterval) {
|
||||
await new Promise((resolve) => setTimeout(resolve, this.#retryInterval!));
|
||||
}
|
||||
|
||||
// Attempt to reconnect.
|
||||
await this.connect();
|
||||
} else {
|
||||
// Close the message stream so `for await...of` consumers resolve.
|
||||
this.#closeMessageStream();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Clears partial SSE frames left over from an abandoned transport. */
|
||||
#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.
|
||||
*/
|
||||
#ensureMessageStreamOpen(): void {
|
||||
if (!this.messages.closed) return;
|
||||
|
||||
this.messages = new AsyncPushIterator<SSEvent>();
|
||||
}
|
||||
|
||||
/** Ends the message iteration loop for the current connection span. */
|
||||
#closeMessageStream(): void {
|
||||
if (this.messages.closed) return;
|
||||
|
||||
this.messages.close();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user