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

Merge development

Browse Source
This commit is contained in:
Harvmaster
2026-08-10 02:48:12 +00:00
parent 941719e4e6
commit f793655cd9
22 changed files with 3240 additions and 733 deletions
Download Patch File Download Diff File Expand all files Collapse all files
source/sse-session/async-push-iterator.ts
+105
View File
@@ -0,0 +1,105 @@
/**
* An async iterable queue that bridges push-based producers and pull-based consumers.
*
* Composes an internal {@link ReadableStream} instead of extending it, so producers
* call {@link push} while consumers use standard async iteration (`for await...of`).
*
* ```ts
* const messages = new AsyncPushIterator<SSEvent>();
*
* // Producer (elsewhere)
* messages.push(event);
*
* // Consumer
* for await (const event of messages) {
* handle(event);
* }
* ```
*
* {@link Symbol.asyncIterator} returns `stream.values({ preventCancel: true })` so
* breaking out of `for await...of` does not cancel the underlying stream. That
* matters for long-lived sessions where the producer keeps pushing after a consumer
* stops reading early (for example, test helpers that only collect a fixed count).
*/
export class AsyncPushIterator<T> {
/** ReadableStream backing the async iterator returned from {@link Symbol.asyncIterator}. */
#stream: ReadableStream<T>;
/** Controller used to enqueue values and close the stream from {@link push} and {@link close}. */
#controller: ReadableStreamDefaultController<T> | undefined;
/** When true, no more values are accepted and iteration eventually completes. */
#closed = false;
public constructor() {
// `start`'s `this` is the underlying source object when using a plain method.
// An arrow function captures the class instance so the controller is stored here.
this.#stream = new ReadableStream({
start: (controller: ReadableStreamDefaultController<T>): void => {
this.#controller = controller;
},
});
}
/**
* Flag indicating if the iterator is closed.
*/
public get closed(): boolean {
return this.#closed;
}
/**
* Enqueues a value for the consumer.
*
* After {@link close}, pushes are silently dropped.
*
* @param value - The next value to yield from the iterator.
*/
push(value: T): void {
if (this.#closed) return;
this.#controller?.enqueue(value);
}
/**
* Causes any future interactions with the associated stream to error with {@link error}.
* Calling this will also clear the pending values immediately, so iterators that were listening will not receive them.
*
* @param error - The error to throw from the stream.
*/
error(error: Error): void {
if (this.#closed) return;
this.#closed = true;
this.#controller?.error(error);
}
/**
* Ends the stream.
*
* Marks the iterator closed so future {@link push} calls are ignored.
* Buffered values are still yielded before iteration completes.
*/
close(): void {
this.#closed = true;
try {
this.#controller?.close();
} catch {
// The reader may already have released or cancelled the stream.
}
}
/**
* Returns an async iterator over the composed stream.
*
* Uses `preventCancel: true` so early `break` from `for await...of` does not
* close the stream and block later pushes.
*
* Because values are discarded after being read, only a single consumer is supported.
* Additional consumers will receive a stream lock error. Unread values will be preserved until {@link close} is called.
*/
[Symbol.asyncIterator](): AsyncIterableIterator<T> {
return this.#stream.values({ preventCancel: true });
}
}
source/sse-session/constants.ts
+34
View File
@@ -0,0 +1,34 @@
/**
* Regex that splits decoded SSE text into lines.
*
* The SSE wire format is line-oriented (`field: value` per line). Servers may
* send `\r\n` (HTTP default), `\n` (Unix), or `\r` (legacy Mac). Matching all
* three keeps parsing correct regardless of platform or server implementation.
*/
export const SSE_LINE_ENDINGS = /\r\n|\r|\n/;
/**
* Regex that matches the single optional leading space in an SSE field value.
*
* Per the SSE spec, `field: value` may include one space immediately after the
* colon; that space is not part of the value. Used with `.replace()` to strip
* it when parsing lines such as `data: hello` → `hello`.
*/
export const SSE_FIELD_VALUE_REGEX = /^ /;
/**
* Regex that matches a trailing newline at the end of a string.
*
* Multiple `data:` lines in one event are joined with `\n`. When the event is
* completed, this removes any stray trailing newline so callers receive the
* payload without an extra line break at the end.
*/
export const SSE_TRAILING_NEWLINE_REGEX = /\n$/;
/**
* The newline character used when normalizing SSE text internally.
*
* Used to join consecutive `data:` lines into one payload and to reassemble
* buffered partial lines between streamed chunks before the next parse call.
*/
export const NEW_LINE = '\n';
source/sse-session/index.ts
+2
View File
@@ -0,0 +1,2 @@
export * from './async-push-iterator.ts';
export * from './sse-event-parser.ts';
source/sse-session/sse-event-parser.ts
+194
View File
@@ -0,0 +1,194 @@
import type { SSEvent } from './types.ts';
import { NEW_LINE, SSE_FIELD_VALUE_REGEX, SSE_LINE_ENDINGS, SSE_TRAILING_NEWLINE_REGEX } from './constants.ts';
/**
* Optional encoders used when decoding incoming SSE bytes and re-encoding
* any buffered remainder between chunks.
*/
export interface SSEEventParserOptions {
/** Decodes raw stream bytes into text. Defaults to a new `TextDecoder`. */
textDecoder: TextDecoder;
}
/**
* Incrementally parses Server-Sent Events (SSE) from streamed byte chunks.
*
* SSE payloads are line-oriented: each event is a sequence of `field: value`
* lines terminated by a blank line. This parser accepts arbitrary chunk
* boundaries from a live HTTP response body and emits only complete events.
*
* Typical usage is one parser instance per connection, calling {@link parseEvents}
* for each chunk received from the stream:
*
* ```ts
* const parser = new SSEEventParser();
*
* for await (const chunk of response.body) {
* for (const event of parser.parseEvents(chunk)) {
* // handle event.data, event.event, event.id, event.retry
* }
* }
* ```
*
* Supported fields follow the SSE spec: `data`, `event`, `id`, and `retry`.
* Multiple `data:` lines in one event are joined with `\n`. An event is only
* emitted once a blank line is seen and at least one `data` field was collected.
*/
export class SSEEventParser {
readonly #textDecoder: TextDecoder;
/** Bytes from a partial line or incomplete event, carried over to the next chunk. */
#messageBuffer: string = '';
/**
* Creates a parser for one SSE stream.
*
* Inject custom encoders in tests or when a non-default character encoding
* is required; production callers can rely on the defaults.
*
* @param options - Optional text encoders for decode/encode of stream bytes.
*/
constructor(options: Partial<SSEEventParserOptions> = {}) {
this.#textDecoder = options.textDecoder ?? new TextDecoder();
}
/**
* Clears any buffered bytes from a partial line or incomplete event.
*
* Call when abandoning a transport so the next connection does not prepend
* stale bytes to incoming chunks.
*/
public reset(): void {
// Clear the message buffer
this.#messageBuffer = '';
// Reset the decoder to clear any buffered bytes
this.#textDecoder.decode();
}
/**
* Parses all complete SSE events contained in a newly received chunk.
*
* The chunk is appended to any bytes buffered from earlier calls. Complete
* events (blank-line delimited blocks with at least one `data` field) are
* returned immediately; any trailing partial line or in-progress event stays
* in the internal buffer until a later chunk completes it.
*
* @param chunk - Newly received SSE stream bytes.
* @returns Zero or more complete parsed SSE events from this chunk.
*/
public parseEvents(chunk: Uint8Array): SSEvent[] {
const lines = this.getBufferedLines(chunk);
const eventLines = lines.slice(0, -1);
const events: SSEvent[] = [];
let event: Partial<SSEvent> = {};
let processedLineCount = 0;
for (const [ index, line ] of eventLines.entries()) {
// A blank line indicates the end of an event. If we have received data, we can complete the event
if (line === '') {
if (event.data !== undefined) {
events.push(this.completeEvent(event));
event = {};
processedLineCount = index + 1;
}
continue;
}
this.parseLine(line, event);
}
this.storeRemainingLines(lines, processedLineCount);
return events;
}
/**
* Appends a new chunk to the buffered bytes and splits the combined payload
* into lines.
*
* Accepts `\r\n`, `\r`, and `\n` line endings so events parse correctly
* regardless of server or platform conventions.
*/
private getBufferedLines(chunk: Uint8Array): string[] {
this.#messageBuffer += this.#textDecoder.decode(chunk, { stream: true });
return this.#messageBuffer.split(SSE_LINE_ENDINGS);
}
/**
* Parses one SSE field line into an in-progress event.
*
* Lines without a colon are ignored. A single optional space after the colon
* is stripped from the field value, per the SSE spec.
*/
private parseLine(line: string, event: Partial<SSEvent>): void {
const colonIndex = line.indexOf(':');
if (colonIndex === -1) return;
const field = line.slice(0, colonIndex);
const value = line.slice(colonIndex + 1).replace(SSE_FIELD_VALUE_REGEX, '');
switch (field) {
case 'data':
event.data = event.data ? `${event.data}${NEW_LINE}${value}` : value;
return;
case 'event':
event.event = value;
return;
case 'id':
event.id = value;
return;
case 'retry':
this.parseRetry(value, event);
return;
}
}
/**
* Applies a numeric `retry:` field to an in-progress event.
*
* Non-numeric values are ignored rather than failing the parse.
*/
private parseRetry(value: string, event: Partial<SSEvent>): void {
const retry = parseInt(value, 10);
if (!isNaN(retry)) {
event.retry = retry;
}
}
/**
* Constructs a completed SSE event from accumulated fields.
*
* Trims a trailing newline from multi-line `data` values so callers receive
* the payload without an extra line break at the end.
*/
private completeEvent(event: Partial<SSEvent>): SSEvent {
return {
...event,
data: event.data?.replace(SSE_TRAILING_NEWLINE_REGEX, ''),
} as SSEvent;
}
/**
* Preserves incomplete trailing lines for the next received chunk.
*
* Only lines that were fully processed (through a completed event boundary)
* are discarded; the remainder is re-encoded into {@link messageBuffer}.
*/
private storeRemainingLines(lines: string[], processedLineCount: number): void {
this.#messageBuffer = lines.slice(processedLineCount).join(NEW_LINE);
}
}
source/sse-session/types.ts
+29
View File
@@ -0,0 +1,29 @@
/**
* Represents a Server-Sent Event.
*/
export interface SSEvent {
/**
* Event data.
*/
data: string;
/**
* Event type.
* This value is optionally sent by the server. Traditional EventSource allows listeners for specific event types.
* The SSE Session collapses all event types into "message" event.
*/
event?: string;
/**
* Event ID.
* This value is optionally sent by the server as a "checkpoint" the client can use to resume from using the Last-Event-ID header.
*/
id?: string;
/**
* Reconnection time in milliseconds.
* This value is optionally sent by the server to indicate the server's preferred time before reconnecting
*/
retry?: number;
}