From ce0c417ccf0f86cebf94fc4c3f72e23dd512ed0b Mon Sep 17 00:00:00 2001 From: Harvey Zuccon Date: Mon, 26 May 2025 02:36:22 +0000 Subject: [PATCH] Feat(SSE): SSE Session auto-reconnect --- .../stack/src/storage/store-remote-http.ts | 4 +- packages/stack/src/utils/index.ts | 4 +- ...nt-msgpackr.ts => sse-session-msgpackr.ts} | 6 +- .../utils/{sse-client.ts => sse-session.ts} | 196 ++++++++++++++---- 4 files changed, 163 insertions(+), 47 deletions(-) rename packages/stack/src/utils/{sse-client-msgpackr.ts => sse-session-msgpackr.ts} (88%) rename packages/stack/src/utils/{sse-client.ts => sse-session.ts} (54%) diff --git a/packages/stack/src/storage/store-remote-http.ts b/packages/stack/src/storage/store-remote-http.ts index 6803438..c6381c2 100644 --- a/packages/stack/src/storage/store-remote-http.ts +++ b/packages/stack/src/storage/store-remote-http.ts @@ -1,5 +1,5 @@ import { BaseStore } from './base-store.js'; -import { SSEClient } from '../utils/index.js'; +import { SSESession } from '../utils/index.js'; import { Bytes, Encrypter, @@ -96,7 +96,7 @@ export class StoreRemoteHTTP extends BaseStore { // NOTE: SSE only supports text-encoding, so we use Extended JSON instead of msgpackr (which is binary). // TODO: It would be really nice to somehow be able to support msgpackr for consistency though. // But, from a bandwidth point of view, this is not crucial: These payloads are "live" payloads (thus, there's not much data). - await SSEClient.from(`${this.baseURL}events`, { + await SSESession.from(`${this.baseURL}events`, { method: 'POST', headers, onMessage: (e) => { diff --git a/packages/stack/src/utils/index.ts b/packages/stack/src/utils/index.ts index 217fab5..4deb577 100644 --- a/packages/stack/src/utils/index.ts +++ b/packages/stack/src/utils/index.ts @@ -3,6 +3,6 @@ export * from './ext-error.js'; export * from './inverse-bloom-filter.js'; export * from './memoizer.js'; export * from './misc.js'; -export * from './sse-client.js'; -export * from './sse-client-msgpackr.js'; +export * from './sse-session.js'; +export * from './sse-session-msgpackr.js'; export * from './uuid.js'; diff --git a/packages/stack/src/utils/sse-client-msgpackr.ts b/packages/stack/src/utils/sse-session-msgpackr.ts similarity index 88% rename from packages/stack/src/utils/sse-client-msgpackr.ts rename to packages/stack/src/utils/sse-session-msgpackr.ts index 6f7583f..d76d725 100644 --- a/packages/stack/src/utils/sse-client-msgpackr.ts +++ b/packages/stack/src/utils/sse-session-msgpackr.ts @@ -1,18 +1,18 @@ -import { SSEClient, SSEClientOptions, SSEvent } from './sse-client.js'; +import { SSESession, SSESessionOptions, SSEvent } from './sse-session.js'; import { unpackMultiple } from 'msgpackr'; /** * SSEClientMsgpackr is a subclass of SSEClient that uses msgpackr to parse the events. */ -export class SSEClientMsgpackr extends SSEClient { +export class SSEClientMsgpackr extends SSESession { /** * Creates a new SSEClientMsgpackr instance. * @param {string} url The URL to connect to * @param {SSEClientOptions} options The options for the SSEClient * @returns {SSEClientMsgpackr} A new SSEClientMsgpackr instance */ - public static async from(url: string, options: SSEClientOptions) { + public static async from(url: string, options: SSESessionOptions) { const client = new SSEClientMsgpackr(url, options); await client.connect(); return client; diff --git a/packages/stack/src/utils/sse-client.ts b/packages/stack/src/utils/sse-session.ts similarity index 54% rename from packages/stack/src/utils/sse-client.ts rename to packages/stack/src/utils/sse-session.ts index 4ab5273..f4e20ad 100644 --- a/packages/stack/src/utils/sse-client.ts +++ b/packages/stack/src/utils/sse-session.ts @@ -1,25 +1,19 @@ -// TODO: 1. Handle reconnection logic. -// 2. Allow specifying custom -// 3. Try to make it so that connect DOES NOT resolve until connection is actually establish. -// There's a very slim chance of a race-condition otherwise where we might miss an event. -// Note that this IS NOT trivial. Claude will probably NOT be able to do this for you! - /** * A Server-Sent Events client implementation using fetch API. * Supports custom headers, POST requests, and is non-blocking. */ -export class SSEClient { +export class SSESession { /** - * Creates and connects a new SSEClient instance. + * Creates and connects a new SSESession instance. * @param url The URL to connect to * @param options Configuration options - * @returns A new connected SSEClient instance + * @returns A new connected SSESession instance */ public static async from( url: string, - options: Partial = {}, - ): Promise { - const client = new SSEClient(url, options); + options: Partial = {}, + ): Promise { + const client = new SSESession(url, options); await client.connect(); return client; } @@ -28,35 +22,68 @@ export class SSEClient { private url: string; private controller: AbortController; private connected: boolean = false; - protected options: SSEClientOptions; + protected options: SSESessionOptions; protected messageBuffer: Uint8Array = new Uint8Array(); + // Listener for when the tab is hidden or shown. + private visibilityChangeHandler: ((event: Event) => void) | null = null; + + // Keep track of the current reconnection attempt to use for exponential backoff. + protected currentAttempt: number = 0; + // Text decoders and encoders for parsing the message buffer. private textDecoder: TextDecoder = new TextDecoder(); private textEncoder: TextEncoder = new TextEncoder(); /** - * Creates a new SSEClient instance. + * Creates a new SSESession instance. * @param url The URL to connect to * @param options Configuration options */ - constructor(url: string, options: Partial = {}) { + constructor(url: string, options: Partial = {}) { this.url = url; this.options = { // Use default fetch function. - fetch: fetch, + fetch: (...args) => fetch(...args), method: 'GET', headers: { Accept: 'text/event-stream', 'Cache-Control': 'no-cache', }, - onOpen: () => {}, + onConnected: () => {}, onMessage: () => {}, - onError: (error) => console.error('SSEClient error:', error), - onClose: () => {}, + onError: (error) => console.error('SSESession error:', error), + onDisconnected: () => {}, + + // Reconnection options + attemptReconnect: true, + retryDelay: 1000, + persistent: false, ...options, }; this.controller = new AbortController(); + + // Set up visibility change handling if in mobile browser environment + if (typeof document !== 'undefined') { + this.visibilityChangeHandler = this.handleVisibilityChange.bind(this); + document.addEventListener('visibilitychange', this.visibilityChangeHandler); + } + } + + /** + * Handles visibility change events in the browser. + */ + private async handleVisibilityChange(): Promise { + // When going to background, close the current connection cleanly + // This allows us to reconnect mobile devices when they come back after leaving the tab or browser app. + if (document.visibilityState === 'hidden') { + this.controller.abort(); + } + + // When coming back to foreground, attempt to reconnect if not connected + if (document.visibilityState === 'visible' && !this.connected) { + await this.attemptReconnect(); + } } /** @@ -66,6 +93,9 @@ export class SSEClient { if (this.connected) return; this.connected = true; + const abortController = new AbortController(); + this.controller = abortController; + const { method, headers, body } = this.options; const fetchOptions: RequestInit = { @@ -92,10 +122,11 @@ export class SSEClient { throw new Error('Response body is null'); } - // Notify that the connection has been established - if (this.options.onOpen) { - this.options.onOpen(); - } + // Reset the current attempt. + this.currentAttempt = 0; + + // Call the onConnected callback. + this.handleCallback(this.options.onConnected, undefined); const reader = response.body.getReader(); @@ -112,9 +143,15 @@ export class SSEClient { if (done) { this.connected = false; - if (this.options.onClose) { - this.options.onClose(); + + // Call the onDisconnected callback. + this.handleCallback(this.options.onDisconnected, undefined); + + // If the connection was closed by the server, we want to attempt a reconnect if the connection should be persistent. + if (this.options.persistent) { + await this.attemptReconnect(); } + break; } @@ -124,7 +161,7 @@ export class SSEClient { // Handle each complete event for (const event of events) { if (this.options.onMessage) { - this.options.onMessage(event); + this.handleCallback(this.options.onMessage, event); } } @@ -134,15 +171,28 @@ export class SSEClient { } catch (error) { this.connected = false; - if (this.options.onError) { + // Call the onDisconnected callback. + this.handleCallback(this.options.onDisconnected, error); + + // If the connection was aborted using the controller, we don't need to call onError. + if (abortController.signal.aborted) { + return; + } + + // Call the onError callback. + // NOTE: we dont use the handleCallback here because it would result in 2 error callbacks. + try { this.options.onError(error); + } catch (error) { + console.log(`SSE Session: onError callback error:`, error); } - if (this.options.onClose) { - this.options.onClose(); + // Attempt to reconnect if enabled + if (this.options.attemptReconnect) { + return await this.attemptReconnect(); } - // Now we'll reject the promise instead of resolving + // Reject the promise reject(error); } }; @@ -214,13 +264,45 @@ export class SSEClient { } /** - * Closes the SSE connection. + * Attempts to reconnect to the SSE endpoint. + * @returns void + */ + protected attemptReconnect(): void { + // Don't attempt to reconnect if it's not enabled or if we are already connected + if (!this.options.attemptReconnect || this.connected) return; + + // Exponential backoff with a maximum of 10 seconds + // TODO: We should update this to use an Exponential Backoff utility when we have one. + this.currentAttempt++; + const delay = Math.min(this.options.retryDelay * Math.pow(1.3, this.currentAttempt - 1), 10000); + console.log(`Attempting to reconnect in ${delay}ms`); + + setTimeout(async () => { + await this.connect(); + }, delay); + } + + /** + * Override the onMessage callback. + * + * @param onMessage The callback to set. + */ + public setOnMessage(onMessage: (event: SSEvent) => void): void { + this.options.onMessage = onMessage; + } + + /** + * Closes the SSE connection and cleans up event listeners. */ public close(): void { - if (!this.connected) return; - + // Clean up everything including the visibility handler this.controller.abort(); - this.connected = false; + + // Remove the visibility handler (This is only required on browsers) + if (this.visibilityChangeHandler && typeof document !== 'undefined') { + document.removeEventListener('visibilitychange', this.visibilityChangeHandler); + this.visibilityChangeHandler = null; + } } /** @@ -230,12 +312,31 @@ export class SSEClient { public isConnected(): boolean { return this.connected; } + + /** + * Will handle thrown errors from the callback and call the onError callback. + * This is to avoid the sse-session from disconnecting from errors that are not a result of the sse-session itself. + * + * @param callback The callback to handle. + * @param args The arguments to pass to the callback. + */ + private handleCallback(callback: (args: T) => void, args: T): void { + try { + callback(args); + } catch (error) { + try { + this.options.onError(error); + } catch (error) { + console.log(`SSE Session: onError callback error:`, error); + } + } + } } /** - * Configuration options for the SSEClient. + * Configuration options for the SSESession. */ -export interface SSEClientOptions { +export interface SSESessionOptions { /** * The fetch function to use. * @@ -264,22 +365,37 @@ export interface SSEClientOptions { /** * Called when the connection is established. */ - onOpen?: () => void; + onConnected: () => void; /** * Called when a message is received. */ - onMessage?: (event: SSEvent) => void; + onMessage: (event: SSEvent) => void; /** * Called when an error occurs. */ - onError?: (error: unknown) => void; + onError: (error: unknown) => void; /** * Called when the connection is closed. */ - onClose?: () => void; + onDisconnected: () => void; + + /** + * Whether to attempt to reconnect. + */ + attemptReconnect: boolean; + + /** + * The delay in milliseconds between reconnection attempts. + */ + retryDelay: number; + + /** + * Whether to reconnect when the session is terminated by the server. + */ + persistent: boolean; } /**