Cleanup

libs/sdk-js/src/tests/sse.test.ts

@@ -2,26 +2,21 @@ import { Readable } from "node:stream";
 import { IterableReadableStream } from "../utils/stream.js";
 import { BytesLineDecoder, SSEDecoder } from "../utils/sse.js";
 
+const gather = async <T>(stream: ReadableStream<T>): Promise<T[]> => {
+  const results: T[] = [];
+  const iterator = IterableReadableStream.fromReadableStream(stream);
+  for await (const chunk of iterator) results.push(chunk);
+  return results;
+};
+
+const textEncoder = new TextEncoder();
+const textDecoder = new TextDecoder();
+
 describe("BytesLineDecoder", () => {
   const createStream = (chunks: Uint8Array[]) => {
     return Readable.toWeb(Readable.from(chunks)) as ReadableStream<Uint8Array>;
   };
 
-  const gather = async (
-    stream: ReadableStream<Uint8Array>,
-  ): Promise<Uint8Array[]> => {
-    const results: Uint8Array[] = [];
-    for await (const chunk of IterableReadableStream.fromReadableStream(
-      stream,
-    )) {
-      results.push(chunk);
-    }
-    return results;
-  };
-
-  const textEncoder = new TextEncoder();
-  const textDecoder = new TextDecoder();
-
   test("handles single line with newline", async () => {
     const input = createStream([textEncoder.encode("hello\n")]);
     const decoded = input.pipeThrough(new BytesLineDecoder());
@@ -78,25 +73,22 @@ describe("BytesLineDecoder", () => {
     expect(textDecoder.decode(results[0])).toBe("line1");
     expect(textDecoder.decode(results[1])).toBe("line2");
   });
+
+  test("handles stale line", async () => {
+    const input = createStream([textEncoder.encode("hello")]);
+    const decoded = input.pipeThrough(new BytesLineDecoder());
+    const results = await gather(decoded);
+
+    expect(results.length).toBe(1);
+    expect(textDecoder.decode(results[0])).toBe("hello");
+  });
 });
 
 describe("SSEDecoder", () => {
   const createStream = (lines: string[]) => {
-    const encoder = new TextEncoder();
-    const chunks = lines.map((line) => encoder.encode(line));
-    return Readable.toWeb(Readable.from(chunks)) as ReadableStream<Uint8Array>;
-  };
-
-  const collectResults = async (
-    stream: ReadableStream<any>,
-  ): Promise<any[]> => {
-    const results: any[] = [];
-    for await (const chunk of IterableReadableStream.fromReadableStream(
-      stream,
-    )) {
-      results.push(chunk);
-    }
-    return results;
+    return Readable.toWeb(
+      Readable.from(lines.map((line) => textEncoder.encode(line))),
+    ) as ReadableStream<Uint8Array>;
   };
 
   test("decodes simple event", async () => {
@@ -109,7 +101,7 @@ describe("SSEDecoder", () => {
       .pipeThrough(new BytesLineDecoder())
       .pipeThrough(new SSEDecoder());
 
-    const results = await collectResults(decoded);
+    const results = await gather(decoded);
     expect(results.length).toBe(1);
     expect(results[0]).toEqual({
       event: "test",
@@ -127,7 +119,7 @@ describe("SSEDecoder", () => {
       .pipeThrough(new BytesLineDecoder())
       .pipeThrough(new SSEDecoder());
 
-    const results = await collectResults(decoded);
+    const results = await gather(decoded);
     expect(results.length).toBe(1);
     expect(results[0]).toEqual({
       event: "test",
@@ -148,7 +140,7 @@ describe("SSEDecoder", () => {
       .pipeThrough(new BytesLineDecoder())
       .pipeThrough(new SSEDecoder());
 
-    const results = await collectResults(decoded);
+    const results = await gather(decoded);
     expect(results.length).toBe(2);
     expect(results[0]).toEqual({
       event: "test1",
@@ -166,7 +158,7 @@ describe("SSEDecoder", () => {
       .pipeThrough(new BytesLineDecoder())
       .pipeThrough(new SSEDecoder());
 
-    const results = await collectResults(decoded);
+    const results = await gather(decoded);
     expect(results.length).toBe(1);
     expect(results[0]).toEqual({
       event: "test",
@@ -180,7 +172,7 @@ describe("SSEDecoder", () => {
       .pipeThrough(new BytesLineDecoder())
       .pipeThrough(new SSEDecoder());
 
-    const results = await collectResults(decoded);
+    const results = await gather(decoded);
     expect(results.length).toBe(1);
     expect(results[0]).toEqual({
       event: "end",

libs/sdk-js/src/utils/sse.ts

@@ -44,25 +44,26 @@ export class BytesLineDecoder extends TransformStream<Uint8Array, Uint8Array> {
         const trailingNewline = TRAILING_NEWLINE.includes(text.at(-1)!);
 
         const lastIdx = text.length - 1;
-        const { lines } = text.reduce(
+        const { lines } = text.reduce<{ lines: Uint8Array[]; from: number }>(
           (acc, cur, idx) => {
             if (acc.from > idx) return acc;
 
-            if (cur === CR && text[idx + 1] === LF) {
+            if (cur === CR || cur === LF) {
               acc.lines.push(text.subarray(acc.from, idx));
-              acc.from = idx + 2;
-            } else if (cur === CR || cur === LF) {
-              acc.lines.push(text.subarray(acc.from, idx));
-              acc.from = idx + 1;
+              if (cur === CR && text[idx + 1] === LF) {
+                acc.from = idx + 2;
+              } else {
+                acc.from = idx + 1;
+              }
             }
 
-            if (idx === lastIdx && acc.from < lastIdx) {
+            if (idx === lastIdx && acc.from <= lastIdx) {
               acc.lines.push(text.subarray(acc.from));
             }
 
             return acc;
           },
-          { lines: [], from: 0 } as { lines: Uint8Array[]; from: number },
+          { lines: [], from: 0 },
         );
 
         if (lines.length === 1 && !trailingNewline) {