Merge branch '33-feat-sse-add-msgpackr-parser-for-sse-client' into 'development'

Resolve "Feat(SSE): Add msgpackr parser for SSE Client"

Closes #33

See merge request GeneralProtocols/xo/stack!54

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

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

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
+            // Decode and parse the chunk
-            const chunk = decoder.decode(value, { stream: true });
+            const events = this.parseEvents(value);
-            buffer += chunk;
-
-            // Process any complete events in the buffer
-            const events = this.parseEvents(buffer);
-            buffer = events.remainder;
 
             // 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 {
     });
   }
 
-  /**
+  protected parseEvents(chunk: Uint8Array): SSEvent[] {
-   * Parses SSE events from a text buffer.
+    // Append new chunk to existing buffer
-   * @param buffer The buffer to parse
+    this.messageBuffer = new Uint8Array([...this.messageBuffer, ...chunk]);
-   * @returns Parsed events and the remaining buffer
+
-   */
-  private parseEvents(buffer: string): {
-    parsed: SSEvent[];
-    remainder: string;
-  } {
     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<SSEvent> = {};
     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;
   }
 
   /**