diff --git a/packages/stack/src/utils/index.ts b/packages/stack/src/utils/index.ts index f5e273d..b460fad 100644 --- a/packages/stack/src/utils/index.ts +++ b/packages/stack/src/utils/index.ts @@ -6,4 +6,5 @@ export * from './inverse-bloom-filter.js'; export * from './memoizer.js'; export * from './misc.js'; export * from './sse-client.js'; +export * from './sse-client-msgpackr.js'; export * from './uuid.js'; diff --git a/packages/stack/src/utils/sse-client-msgpackr.ts b/packages/stack/src/utils/sse-client-msgpackr.ts new file mode 100644 index 0000000..44bb06d --- /dev/null +++ b/packages/stack/src/utils/sse-client-msgpackr.ts @@ -0,0 +1,54 @@ +import { SSEClient, SSEClientOptions, SSEvent } from './sse-client'; + +import { unpackMultiple } from 'msgpackr'; + +/** + * SSEClientMsgpackr is a subclass of SSEClient that uses msgpackr to parse the events. + */ +export class SSEClientMsgpackr extends SSEClient { + + /** + * Creates a new SSEClientMsgpackr instance. + * @param {string} url The URL to connect to + * @param {SSEClientOptions} options The options for the SSEClient + * @returns {SSEClientMsgpackr} A new SSEClientMsgpackr instance + */ + public static async from(url: string, options: SSEClientOptions) { + const client = new SSEClientMsgpackr(url, options); + await client.connect(); + return client; + } + + /** + * Parses the events from the msgpackr message buffer. + * @param {Uint8Array} chunk The chunk to parse + * @returns {SSEvent[]} The parsed events + */ + protected parseEvents(chunk: Uint8Array): SSEvent[] { + // Append new chunk to existing buffer + this.messageBuffer = new Uint8Array([...this.messageBuffer, ...chunk]); + + // Create an array to store the parsed events + const events: SSEvent[] = []; + + try { + // Try to decode the entire buffer as msgPackr messages + unpackMultiple(this.messageBuffer, (value, start, end) => { + // Remove the event from the buffer + this.messageBuffer = this.messageBuffer.slice(end); + + // Add the event to the events array + events.push(value); + }); + } catch (error) { + // If we can't decode the buffer, it might be incomplete + // Keep the entire buffer for next time + } finally { + // Update buffer with any remaining unprocessed content + this.messageBuffer = new Uint8Array(this.messageBuffer); + + // Return the events + return events; + } + } +} diff --git a/packages/stack/src/utils/sse-client.ts b/packages/stack/src/utils/sse-client.ts index 0c299f9..e46a657 100644 --- a/packages/stack/src/utils/sse-client.ts +++ b/packages/stack/src/utils/sse-client.ts @@ -26,9 +26,14 @@ export class SSEClient { // State. private url: string; - private options: SSEClientOptions; private controller: AbortController; private connected: boolean = false; + protected options: SSEClientOptions; + protected messageBuffer: Uint8Array = new Uint8Array(); + + // Text decoders and encoders for parsing the message buffer. + private textDecoder: TextDecoder = new TextDecoder(); + private textEncoder: TextEncoder = new TextEncoder(); /** * Creates a new SSEClient instance. @@ -93,8 +98,6 @@ export class SSEClient { } const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; // Start the read here so we can resolve and then start handling the stream // We will await this at the start of the loop, and redeclare it at the end of each loop iteration @@ -115,16 +118,11 @@ export class SSEClient { break; } - // Decode the received chunk and add it to the buffer - const chunk = decoder.decode(value, { stream: true }); - buffer += chunk; - - // Process any complete events in the buffer - const events = this.parseEvents(buffer); - buffer = events.remainder; + // Decode and parse the chunk + const events = this.parseEvents(value); // Handle each complete event - for (const event of events.parsed) { + for (const event of events) { if (this.options.onMessage) { this.options.onMessage(event); } @@ -154,21 +152,17 @@ export class SSEClient { }); } - /** - * Parses SSE events from a text buffer. - * @param buffer The buffer to parse - * @returns Parsed events and the remaining buffer - */ - private parseEvents(buffer: string): { - parsed: SSEvent[]; - remainder: string; - } { + protected parseEvents(chunk: Uint8Array): SSEvent[] { + // Append new chunk to existing buffer + this.messageBuffer = new Uint8Array([...this.messageBuffer, ...chunk]); + const events: SSEvent[] = []; - const lines = buffer.split(/\r\n|\r|\n/); + const lines = this.textDecoder.decode(this.messageBuffer).split(/\r\n|\r|\n/); let currentEvent: Partial = {}; let completeEventCount = 0; + // Iterate over the lines to find complete events for (let i = 0; i < lines.length; i++) { const line = lines[i]; @@ -210,8 +204,11 @@ export class SSEClient { } } + // Store the remainder of the buffer for the next chunk const remainder = lines.slice(completeEventCount).join('\n'); - return { parsed: events, remainder }; + this.messageBuffer = this.textEncoder.encode(remainder); + + return events; } /**