From 76f787211a829b4cd61403b9b3421dfc26908e20 Mon Sep 17 00:00:00 2001 From: Harvey Zuccon Date: Sun, 19 Jul 2026 09:16:49 +0000 Subject: [PATCH] Sse event parser --- source/index.ts | 1 + source/sse-session/constants.ts | 34 ++++ source/sse-session/index.ts | 1 + source/sse-session/sse-event-parser.ts | 194 +++++++++++++++++++ source/sse-session/types.ts | 29 +++ test/sse-session/fixtures/events.fixtures.ts | 155 +++++++++++++++ test/sse-session/sse-event-parser.test.ts | 147 ++++++++++++++ 7 files changed, 561 insertions(+) create mode 100644 source/sse-session/constants.ts create mode 100644 source/sse-session/index.ts create mode 100644 source/sse-session/sse-event-parser.ts create mode 100644 source/sse-session/types.ts create mode 100644 test/sse-session/fixtures/events.fixtures.ts create mode 100644 test/sse-session/sse-event-parser.test.ts diff --git a/source/index.ts b/source/index.ts index 5c97c08..a29e6ae 100644 --- a/source/index.ts +++ b/source/index.ts @@ -1,5 +1,6 @@ export * from './extended-json.ts'; export * from './script.ts'; +export * from './sse-session/index.ts'; export * from './template/errors.ts'; export * from './template/identifier.ts'; export * from './template/parser.ts'; diff --git a/source/sse-session/constants.ts b/source/sse-session/constants.ts new file mode 100644 index 0000000..6ed2005 --- /dev/null +++ b/source/sse-session/constants.ts @@ -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'; diff --git a/source/sse-session/index.ts b/source/sse-session/index.ts new file mode 100644 index 0000000..563fb3e --- /dev/null +++ b/source/sse-session/index.ts @@ -0,0 +1 @@ +export * from './sse-event-parser.ts'; diff --git a/source/sse-session/sse-event-parser.ts b/source/sse-session/sse-event-parser.ts new file mode 100644 index 0000000..c3bd1e8 --- /dev/null +++ b/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 = {}) { + 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); + } +} diff --git a/source/sse-session/types.ts b/source/sse-session/types.ts new file mode 100644 index 0000000..224e048 --- /dev/null +++ b/source/sse-session/types.ts @@ -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; +} diff --git a/test/sse-session/fixtures/events.fixtures.ts b/test/sse-session/fixtures/events.fixtures.ts new file mode 100644 index 0000000..ba8afee --- /dev/null +++ b/test/sse-session/fixtures/events.fixtures.ts @@ -0,0 +1,155 @@ +import type { SSEvent } from '../../../source/sse-session/types.ts'; + +type EventFixture = { + raw: string; + parsed?: SSEvent[]; +}; + +/** Combines all the raw strings into a single chunk and flattens the parsed arrays into a single array to simulate multi-event chunks. */ +const withCombinedChunk = (fixtures: EventFixture[]): EventFixture => { + return { + raw: fixtures.map(({ raw }) => raw).join(''), + parsed: fixtures.flatMap(({ parsed }) => parsed ?? []), + }; +}; + +export const priceOracleEvents: EventFixture[] = [ + { + raw: 'retry: 1000\nevent: 02664276fb7513f838f505c221680a9d963479ffb45452b0c744ddb6bd19ecacb3\ndata: {"message":"411d396aa53516008f35160049a46100","signature":"936ee3de4c179a1c23d5227c6cadd7ccee11aa1a5a48624ff59ff6954dc607752be8d069610a342ed7610444741bc5fa5dd9bd438fa6c1f1307a7b50a663e17c"}\n\n', + parsed: [ + { + retry: 1000, + event: '02664276fb7513f838f505c221680a9d963479ffb45452b0c744ddb6bd19ecacb3', + data: '{"message":"411d396aa53516008f35160049a46100","signature":"936ee3de4c179a1c23d5227c6cadd7ccee11aa1a5a48624ff59ff6954dc607752be8d069610a342ed7610444741bc5fa5dd9bd438fa6c1f1307a7b50a663e17c"}', + }, + ], + }, + { + raw: 'event: 0336f13d65e3bd6a521bf582f22b74f50edab7c278d38b80e319673b859f95d830\ndata: {"message":"471d396ac4571200b057120047e80000","signature":"4fb94f8a46475ebac87d49860372c67dae9e89412ff704b14e5248c4911228ad4c6429ae3771bf33122ebbfac3084c6a13efdbff9cb987af55032364ad6e1816"}\n\n', + parsed: [ + { + event: '0336f13d65e3bd6a521bf582f22b74f50edab7c278d38b80e319673b859f95d830', + data: '{"message":"471d396ac4571200b057120047e80000","signature":"4fb94f8a46475ebac87d49860372c67dae9e89412ff704b14e5248c4911228ad4c6429ae3771bf33122ebbfac3084c6a13efdbff9cb987af55032364ad6e1816"}', + }, + ], + }, + { + raw: 'event: 021f8338ccd45a7790025de198a266f252ac43c95bf81d2469feff110beeac89dd\ndata: {"message":"491d396aad341a0092341a0064120000","signature":"dde76381753edd39beaeef0a413da54e906c9a2e2bb2ae482c14e72702cdff43e1ad26354663c15236ad9ec04e054ef39b4047c24c859d4f5aef1f58502b4686"}\n\n', + parsed: [ + { + event: '021f8338ccd45a7790025de198a266f252ac43c95bf81d2469feff110beeac89dd', + data: '{"message":"491d396aad341a0092341a0064120000","signature":"dde76381753edd39beaeef0a413da54e906c9a2e2bb2ae482c14e72702cdff43e1ad26354663c15236ad9ec04e054ef39b4047c24c859d4f5aef1f58502b4686"}', + }, + ], + }, + { + raw: 'event: 02e82ad82eb88fcdfd02fd5e2e0a67bc6ef4139bbcb63ce0b107a7604deb9f7ce1\ndata: {"message":"4b1d396ab5341a009a341a00a0490000","signature":"3bbd83943e3cad352c3346fe6fa68913f66529551dab5aef574b73b5c276f9d860011dfb2d7ecae7f12b92b53b1dac4b94d15a3d78adcc70133bfef4c45e13c9"}\n\n', + parsed: [ + { + event: '02e82ad82eb88fcdfd02fd5e2e0a67bc6ef4139bbcb63ce0b107a7604deb9f7ce1', + data: '{"message":"4b1d396ab5341a009a341a00a0490000","signature":"3bbd83943e3cad352c3346fe6fa68913f66529551dab5aef574b73b5c276f9d860011dfb2d7ecae7f12b92b53b1dac4b94d15a3d78adcc70133bfef4c45e13c9"}', + }, + ], + }, + { + raw: 'event: 038ab22e37cf020f6bbef40111ddc51083a936f0821de56ac01f799cf15b87904d\ndata: {"message":"4f1d396abd341a00a2341a00722c0000","signature":"3b771dc4490010066ccca9e02aa467b950d6fb5cd0000e3f59ec4fadffc5aedb6369b7a62b9c08695d6fa114cfbd62f3e98286ac98e276f66e481a89764a3716"}\n\n', + parsed: [ + { + event: '038ab22e37cf020f6bbef40111ddc51083a936f0821de56ac01f799cf15b87904d', + data: '{"message":"4f1d396abd341a00a2341a00722c0000","signature":"3b771dc4490010066ccca9e02aa467b950d6fb5cd0000e3f59ec4fadffc5aedb6369b7a62b9c08695d6fa114cfbd62f3e98286ac98e276f66e481a89764a3716"}', + }, + ], + }, + { + raw: 'event: 030654b9598186fe4bc9e1b0490c6b85b13991cdb9a7afa34af1bbeee22a35487a\ndata: {"message":"531d396abc341a00a1341a001f0f0200","signature":"380e5954336b855736e22e602b543c4cb2918f0c0bb67d8723573f9cabb4dcf14714ebbfffb7687f97cad21115659d28d493b4e83014a7b62e36dd277d1c44c3"}\n\n', + parsed: [ + { + event: '030654b9598186fe4bc9e1b0490c6b85b13991cdb9a7afa34af1bbeee22a35487a', + data: '{"message":"531d396abc341a00a1341a001f0f0200","signature":"380e5954336b855736e22e602b543c4cb2918f0c0bb67d8723573f9cabb4dcf14714ebbfffb7687f97cad21115659d28d493b4e83014a7b62e36dd277d1c44c3"}', + }, + ], + }, + { + raw: 'event: 03e980928f14fc98e1f9d75d15f0b67dc58cdd3f5c641b8f825b146bcc04bd232c\ndata: {"message":"531d396aa6952100e791210064120000","signature":"fc9acd0b9a0f09e1a5f48d6b61b27a445529cf36b88b39d5df5ed83edbc6b4469ede6933f001f9202f2b590dd8e0f016b9175157fbc5d6189a5fc78cc0443e4d"}\n\n', + parsed: [ + { + event: '03e980928f14fc98e1f9d75d15f0b67dc58cdd3f5c641b8f825b146bcc04bd232c', + data: '{"message":"531d396aa6952100e791210064120000","signature":"fc9acd0b9a0f09e1a5f48d6b61b27a445529cf36b88b39d5df5ed83edbc6b4469ede6933f001f9202f2b590dd8e0f016b9175157fbc5d6189a5fc78cc0443e4d"}', + }, + ], + }, + { + raw: 'event: 02bb9b3324df889a66a57bc890b3452b84a2a74ba753f8842b06bba03e0fa0dfc5\ndata: {"message":"541d396adc191800c419180060440000","signature":"ed1e37324b58815cf448d16b653654c9759817461460fb929893b8afb763ea2aa8ad566149cc2e429590308f4c17f4c1cc74ba0384dff01fc039f941238d8590"}\n\n', + parsed: [ + { + event: '02bb9b3324df889a66a57bc890b3452b84a2a74ba753f8842b06bba03e0fa0dfc5', + data: '{"message":"541d396adc191800c419180060440000","signature":"ed1e37324b58815cf448d16b653654c9759817461460fb929893b8afb763ea2aa8ad566149cc2e429590308f4c17f4c1cc74ba0384dff01fc039f941238d8590"}', + }, + ], + }, + { + raw: 'event: 02d3c1de9d4bc77d6c3608cbe44d10138c7488e592dc2b1e10a6cf0e92c2ecb047\ndata: {"message":"551d396a17952100d2912100474e0000","signature":"74d737547c4ee207d1bcf43eba0ae3c73264f1f477a2947ecf8685cdf0c79a408b22e8df49fe487c34ba0efe8cc520745d45b6338d69c493db36bdbf511b72fb"}\n\n', + parsed: [ + { + event: '02d3c1de9d4bc77d6c3608cbe44d10138c7488e592dc2b1e10a6cf0e92c2ecb047', + data: '{"message":"551d396a17952100d2912100474e0000","signature":"74d737547c4ee207d1bcf43eba0ae3c73264f1f477a2947ecf8685cdf0c79a408b22e8df49fe487c34ba0efe8cc520745d45b6338d69c493db36bdbf511b72fb"}', + }, + ], + }, +]; + +export const storageEvents: EventFixture[] = [ + { + raw: 'id: 1234\ndata: { "hello": "world" }\n\n', + parsed: [ + { + id: '1234', + data: '{ "hello": "world" }', + }, + ], + }, +]; + +export const edgeCases: EventFixture[] = [ + // Multiple data lines + { + raw: 'data: { "hello": "world" }\ndata: { "hello": "world" }\n\n', + parsed: [ + { + data: '{ "hello": "world" }\n{ "hello": "world" }', + }, + ], + }, + // Message without any colons + { + raw: 'message without any colons\n\n', + parsed: [], + }, + // Retry without a number + { + raw: 'retry: not a number\n\n', + parsed: [], + }, + // Data that contains a string with a new line in it + { + raw: 'data: { "hello": "world\\n" }\n\n', + parsed: [ + { + data: '{ "hello": "world\\n" }', + }, + ], + }, + // Emoji character support (mostly to test partial chunks) + { + raw: 'data: Hello 😀 world\n\n', + parsed: [ + { + data: 'Hello 😀 world', + }, + ], + }, +]; + +export const priceOracleEventsCombined = withCombinedChunk(priceOracleEvents); +export const storageEventsCombined = withCombinedChunk(storageEvents); +export const edgeCasesCombined = withCombinedChunk(edgeCases); diff --git a/test/sse-session/sse-event-parser.test.ts b/test/sse-session/sse-event-parser.test.ts new file mode 100644 index 0000000..b5816f0 --- /dev/null +++ b/test/sse-session/sse-event-parser.test.ts @@ -0,0 +1,147 @@ +import { expect, test } from 'vitest'; +import { SSEEventParser } from '../../source/sse-session/sse-event-parser.ts'; +import type { SSEvent } from '../../source/sse-session/types.ts'; + +import { + edgeCases, + priceOracleEvents, + storageEvents, + priceOracleEventsCombined, + storageEventsCombined, + edgeCasesCombined, +} from './fixtures/events.fixtures.ts'; + +/** Shared encoder for turning fixture strings into stream bytes. */ +const textEncoder = new TextEncoder(); + +/** + * Tests that SSEEventParser parses a simple data event. + */ +const testSseEventParserParsesSimpleEvent = (): void => { + const parser = new SSEEventParser(); + + const events = parser.parseEvents(textEncoder.encode('data: test\n\n')); + + expect(events).toEqual([{ data: 'test' }]); +}; + +/** + * Tests that SSEEventParser parses all fixture events correctly. + */ +const testSseEventParserParsesAllFixtures = (): void => { + const parser = new SSEEventParser(); + + // Combine all individual event fixtures from each domain. + const combinedEvents = [ ...priceOracleEvents, ...storageEvents, ...edgeCases ]; + + // Iterate over each combined fixture and test that the parser parses all events correctly. + for (const { raw, parsed } of combinedEvents) { + const events = parser.parseEvents(textEncoder.encode(raw)); + + expect(events).toEqual(parsed); + } +}; + +/** + * Tests that SSEEventParser handles multiple events in the same chunk. + */ +const testSseEventParserHandlesMultipleEventsInOneChunk = (): void => { + const parser = new SSEEventParser(); + + // Each combined fixture packs several events into one raw payload. + const allEvents = [ priceOracleEventsCombined, storageEventsCombined, edgeCasesCombined ]; + + // Iterate over each combined fixture and test that the parser handles the multiple events in one chunk correctly. + for (const { raw, parsed } of allEvents) { + const bytes = textEncoder.encode(raw); + const events = parser.parseEvents(bytes); + + expect(events).toEqual(parsed); + } +}; + +/** + * Tests that SSEEventParser handles partial chunks delivered one character at a time. + */ +const testSseEventParserHandlesPartialChunks = (): void => { + const fixtures = [ ...priceOracleEvents, ...storageEvents, ...edgeCases ]; + + // Iterate over each fixture and test that the parser handles the partial chunks correctly. + for (const { raw, parsed } of fixtures) { + const parser = new SSEEventParser(); + const finalEvents: SSEvent[] = []; + + // Iterate over each character in the raw string and try to parse the events in its buffer + for (const character of raw) { + const bytes = textEncoder.encode(character); + const events = parser.parseEvents(bytes); + + finalEvents.push(...events); + } + + // Verify the events match the expected events. + expect(finalEvents).toEqual(parsed ?? []); + } +}; + +/** + * Tests that SSEEventParser handles partial byte chunks delivered one byte at a time. + * This tests that unicode characters (like emojis) are still parsed correctly despite being split between two "chunks". + */ +const testSseEventParserHandlesPartialByteChunks = (): void => { + // Combine all individual event fixtures from each domain. + const fixtures = [ ...priceOracleEvents, ...storageEvents, ...edgeCases ]; + + // Iterate over each fixture and test that the parser handles the partial byte chunks correctly. + for (const { raw, parsed } of fixtures) { + const parser = new SSEEventParser(); + const finalEvents: SSEvent[] = []; + const bytes = textEncoder.encode(raw); + + // Iterate over each byte in the raw string and try to parse the events in its buffer + for (const byte of bytes) { + const uint8Bytes = Uint8Array.of(byte); + const events = parser.parseEvents(uint8Bytes); + + finalEvents.push(...events); + } + + // Verify the events match the expected events. + expect(finalEvents).toEqual(parsed ?? []); + } +}; + +/** + * Tests that SSEEventParser clears its buffer when reset is called. + */ +const testSseEventParserClearsBufferOnReset = (): void => { + // Create a new parser. + const parser = new SSEEventParser(); + + // Parse the events in the buffer. + parser.parseEvents(textEncoder.encode('data: stale')); + + // Reset the parser. + parser.reset(); + + // Parse the events in the buffer. + const events = parser.parseEvents(textEncoder.encode('data: fresh\n\n')); + + // Verify the events match the expected events. + expect(events).toEqual([ + { + data: 'fresh', + }, + ]); +}; + +const runTests = async (): Promise => { + test('SSEEventParser: parses a simple data event', testSseEventParserParsesSimpleEvent); + test('SSEEventParser: parses all fixture events', testSseEventParserParsesAllFixtures); + test('SSEEventParser: handles multiple events in one chunk', testSseEventParserHandlesMultipleEventsInOneChunk); + test('SSEEventParser: handles partial chunks', testSseEventParserHandlesPartialChunks); + test('SSEEventParser: handles partial byte chunks', testSseEventParserHandlesPartialByteChunks); + test('SSEEventParser: clears the buffer on reset', testSseEventParserClearsBufferOnReset); +}; + +await runTests();