diff --git a/libs/sdk-js/src/tests/sse.test.ts b/libs/sdk-js/src/tests/sse.test.ts index 51ebbbd0c..b4f7ca7d8 100644 --- a/libs/sdk-js/src/tests/sse.test.ts +++ b/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 (stream: ReadableStream): Promise => { + 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; }; - const gather = async ( - stream: ReadableStream, - ): Promise => { - 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; - }; - - const collectResults = async ( - stream: ReadableStream, - ): Promise => { - 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; }; 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", diff --git a/libs/sdk-js/src/utils/sse.ts b/libs/sdk-js/src/utils/sse.ts index 0c6ccdc92..2df40a1df 100644 --- a/libs/sdk-js/src/utils/sse.ts +++ b/libs/sdk-js/src/utils/sse.ts @@ -44,25 +44,26 @@ export class BytesLineDecoder extends TransformStream { 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) {