diff --git a/source/sse-session/sse-session.ts b/source/sse-session/sse-session.ts index 7ff1719..fd44f1c 100644 --- a/source/sse-session/sse-session.ts +++ b/source/sse-session/sse-session.ts @@ -165,7 +165,7 @@ export class SSESession extends EventEmitter { } /** SSE endpoint URL for this session. */ - private readonly url: string; + readonly #url: string; /** * Per-instance configuration. @@ -203,7 +203,7 @@ export class SSESession extends EventEmitter { }; /** AbortController for the currently active fetch, if any. */ - private controller: AbortController | null = null; + #controller: AbortController | null = null; /** * Asynchronous stream of parsed SSE events for the active connection. @@ -227,7 +227,7 @@ export class SSESession extends EventEmitter { public constructor(url: string, options: Partial = {}) { super(); - this.url = url; + this.#url = url; this.options = { ...this.options, ...options, @@ -241,7 +241,7 @@ export class SSESession extends EventEmitter { * * Resolves once the HTTP stream is established and `"connected"` has been * emitted. Body reading continues asynchronously in the background via - * {@link readStream}. + * {@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 @@ -249,16 +249,16 @@ export class SSESession extends EventEmitter { */ public async connect(): Promise { // If there is already a controller present, we are already connected. - if (this.controller) return; + 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(); + this.#resetEventParser(); + this.#ensureMessageStreamOpen(); const controller = new AbortController(); - this.controller = controller; + this.#controller = controller; const { method, headers, body } = this.options; @@ -275,22 +275,22 @@ export class SSESession extends EventEmitter { let reader: ReadableStreamDefaultReader; try { - reader = await this.options.retry.run(() => this.createReader(fetchOptions)); + 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; + if (this.#controller !== controller) return; - this.controller = null; + this.#controller = null; - await this.notifyDisconnected(); - await this.notifyError(error); - this.closeMessageStream(); + 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) { + if (this.#controller !== controller) { await reader.cancel(); return; @@ -303,7 +303,7 @@ export class SSESession extends EventEmitter { this.emit('connected', undefined); // Fire-and-forget: connect() resolves while the stream is consumed. - this.readStream(reader, controller).catch((error) => { + this.#readStream(reader, controller).catch((error) => { this.options.onError(error); }); } @@ -318,17 +318,17 @@ export class SSESession extends EventEmitter { * Emits `"disconnected"` but not `"closed"`. */ public async abort(): Promise { - if (!this.controller) return; + if (!this.#controller) return; // Grab the current controller to ensure we are aborting the correct one. - const controller = this.controller; - this.controller = null; + const controller = this.#controller; + this.#controller = null; // Invalidate any in-flight read loop and fetch for this transport. controller.abort(); - this.resetEventParser(); + this.#resetEventParser(); - await this.notifyDisconnected(); + await this.#notifyDisconnected(); } /** @@ -338,13 +338,13 @@ export class SSESession extends EventEmitter { * Closes {@link messages} and emits `"closed"`. */ public async disconnect(): Promise { - this.closeMessageStream(); + this.#closeMessageStream(); this.emit('closed', undefined); - if (this.controller) { + if (this.#controller) { await this.abort(); } else { - this.resetEventParser(); + this.#resetEventParser(); } } @@ -354,22 +354,22 @@ export class SSESession extends EventEmitter { * {@link SSESessionOptions.onRequest} may mutate headers (for example auth * tokens or `Last-Event-ID`) before the fetch runs. */ - private async createReader(fetchOptions: RequestInit): Promise> { + async #createReader(fetchOptions: RequestInit): Promise> { const requestOptions = await this.options.onRequest(fetchOptions); - const response = await this.options.fetch(this.url, requestOptions); + 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); + void this.#notifyError(error); throw error; } if (!response.body) { const error = new ResponseBodyNullError(); - void this.notifyError(error); + void this.#notifyError(error); throw error; } @@ -380,24 +380,24 @@ export class SSESession extends EventEmitter { * Reads bytes from an established stream until it ends, errors, or is * superseded by a newer connection. */ - private async readStream(reader: ReadableStreamDefaultReader, controller: AbortController): Promise { + async #readStream(reader: ReadableStreamDefaultReader, controller: AbortController): Promise { try { - while (this.controller === controller) { + 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 (this.#controller !== controller) return; if (done) { - this.controller = null; + this.#controller = null; - await this.notifyDisconnected(); + await this.#notifyDisconnected(); if (this.options.persistent) { // Server closed gracefully — reopen unless the consumer opted out. await this.connect(); } else { - this.closeMessageStream(); + this.#closeMessageStream(); } return; @@ -413,28 +413,28 @@ export class SSESession extends EventEmitter { } } 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; + if (controller !== this.#controller) return; // Invalidate the current controller to allow for reconnection if needed - this.controller = null; + this.#controller = null; - await this.notifyDisconnected(); + await this.#notifyDisconnected(); // Expected path for abort() — do not treat as an error or reconnect. if (controller.signal.aborted) return; - await this.notifyError(error); + await this.#notifyError(error); if (this.options.attemptReconnect) { await this.connect(); } else { - this.closeMessageStream(); + this.#closeMessageStream(); } } } /** Clears partial SSE frames left over from an abandoned transport. */ - private resetEventParser(): void { + #resetEventParser(): void { this.options.eventParser.reset(); } @@ -442,21 +442,21 @@ export class SSESession extends EventEmitter { * Creates a new {@link messages} iterator when the previous one was closed * by a terminal disconnect or server stream end. */ - private ensureMessageStreamOpen(): void { + #ensureMessageStreamOpen(): void { if (!this.messages.closed) return; this.messages = new AsyncPushIterator(); } /** Ends the message iteration loop for the current connection span. */ - private closeMessageStream(): void { + #closeMessageStream(): void { if (this.messages.closed) return; this.messages.close(); } /** Invokes {@link SSESessionOptions.onDisconnected} and emits `"disconnected"`. */ - private async notifyDisconnected(): Promise { + async #notifyDisconnected(): Promise { await tryAsync( () => this.options.onDisconnected(), (error) => this.options.onError(error), @@ -465,7 +465,7 @@ export class SSESession extends EventEmitter { } /** Invokes {@link SSESessionOptions.onError} and emits `"error"`. */ - private async notifyError(error: unknown): Promise { + async #notifyError(error: unknown): Promise { const errorInstance = error instanceof Error ? error : new Error(String(error)); await tryAsync(