Harvmaster/xo-cash-utils
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'sse-and-backoff' into HEAD

Browse Source
This commit is contained in:
Harvmaster
2026-08-12 03:24:35 +00:00
parent bc6f9a8c7f c31b51a1b6
commit 31ac7dcc0b
22 changed files with 3833 additions and 1026 deletions
Download Patch File Download Diff File Expand all files Collapse all files
source/sse-session/sse-session.ts
+45 -45
View File
@@ -166,7 +166,7 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
}
/** SSE endpoint URL for this session. */
private readonly url: string;
readonly #url: string;
/**
* Per-instance configuration.
@@ -204,7 +204,7 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
};
/** 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.
@@ -228,7 +228,7 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
public constructor(url: string, options: Partial<SSESessionOptions> = {}) {
super();
this.url = url;
this.#url = url;
this.options = {
...this.options,
...options,
@@ -242,7 +242,7 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
*
* 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
@@ -250,16 +250,16 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
*/
public async connect(): Promise<void> {
// 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;
@@ -268,7 +268,7 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
const fetchOptions: RequestInit = {
method,
headers: headers || {},
body: fetchBody,
body: fetchBody ?? null,
signal: controller.signal,
cache: 'no-store',
};
@@ -276,22 +276,22 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
let reader: ReadableStreamDefaultReader<Uint8Array>;
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;
@@ -304,7 +304,7 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
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);
});
}
@@ -319,17 +319,17 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
* Emits `"disconnected"` but not `"closed"`.
*/
public async abort(): Promise<void> {
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();
}
/**
@@ -339,13 +339,13 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
* Closes {@link messages} and emits `"closed"`.
*/
public async disconnect(): Promise<void> {
this.closeMessageStream();
this.#closeMessageStream();
this.emit('closed', undefined);
if (this.controller) {
if (this.#controller) {
await this.abort();
} else {
this.resetEventParser();
this.#resetEventParser();
}
}
@@ -355,22 +355,22 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
* {@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>> {
async #createReader(fetchOptions: RequestInit): Promise<ReadableStreamDefaultReader<Uint8Array>> {
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;
}
@@ -381,24 +381,24 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
* 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> {
async #readStream(reader: ReadableStreamDefaultReader<Uint8Array>, controller: AbortController): Promise<void> {
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;
@@ -414,28 +414,28 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
}
} 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();
}
@@ -443,21 +443,21 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
* 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<SSEvent>();
}
/** 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<void> {
async #notifyDisconnected(): Promise<void> {
await tryAsync(
() => this.options.onDisconnected(),
(error) => this.options.onError(error),
@@ -466,7 +466,7 @@ export class SSESession extends EventEmitter<SSESessionEventMap> {
}
/** Invokes {@link SSESessionOptions.onError} and emits `"error"`. */
private async notifyError(error: unknown): Promise<void> {
async #notifyError(error: unknown): Promise<void> {
const errorInstance = error instanceof Error ? error : new Error(String(error));
await tryAsync(