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 = {}) { 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 = {}; 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): 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): 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 { 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); } }