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

Merge branch 'sse-event-parser' into 'development'

Browse Source 
Sse event parser

See merge request GeneralProtocols/xo/utils!11
This commit is contained in:
Kuldeep
2026-07-19 09:16:50 +00:00
parent 299ea23c5a 76f787211a
commit dd6915ceae
7 changed files with 561 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
source/index.ts
View File

@@ -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';

34
source/sse-session/constants.ts Normal file
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';

1
source/sse-session/index.ts Normal file
View File

@@ -0,0 +1 @@
export * from './sse-event-parser.ts';

194
source/sse-session/sse-event-parser.ts Normal file
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);
}
}

29
source/sse-session/types.ts Normal file
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;
}

155
test/sse-session/fixtures/events.fixtures.ts Normal file
View File

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

147
test/sse-session/sse-event-parser.test.ts Normal file
View File

@@ -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<void> => {
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();