Sse event parser

source/sse-session/sse-event-parser.ts

@@ -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);
+    }
+}