diff --git a/packages/fetch/src/stream.test.ts b/packages/fetch/src/stream.test.ts index 084002ac54a..d84bbdf7c96 100644 --- a/packages/fetch/src/stream.test.ts +++ b/packages/fetch/src/stream.test.ts @@ -21,6 +21,23 @@ function createMockResponse(sseLines: string[]): Response { } as unknown as Response; } +function createMockResponseFromRawChunks(chunks: string[]): Response { + const stream = new Readable({ + read() { + for (const chunk of chunks) { + this.push(chunk); + } + this.push(null); + }, + }) as any; + + return { + status: 200, + body: stream, + text: async () => "", + } as unknown as Response; +} + describe("streamSse", () => { it("yields parsed SSE data objects that ends with `data:[DONE]`", async () => { const sseLines = [ @@ -54,6 +71,19 @@ describe("streamSse", () => { expect(results).toEqual([{ foo: "bar" }, { baz: 42 }]); }); + it("ignores comment keepalive lines that share a chunk with data lines", async () => { + const response = createMockResponseFromRawChunks([ + 'data: {"foo": "bar"}\n\n: ping\ndata: {"baz": 42}\n\ndata: [DONE]\n\n', + ]); + + const results = []; + for await (const data of streamSse(response)) { + results.push(data); + } + + expect(results).toEqual([{ foo: "bar" }, { baz: 42 }]); + }); + it("throws on malformed JSON", async () => { const sseLines = ['data: {"foo": "bar"', "data:[DONE]"]; const response = createMockResponse(sseLines); diff --git a/packages/fetch/src/stream.ts b/packages/fetch/src/stream.ts index f73d61dbafc..d2ee4fab212 100644 --- a/packages/fetch/src/stream.ts +++ b/packages/fetch/src/stream.ts @@ -115,9 +115,8 @@ function parseSseLine(line: string): { done: boolean; data: any } { if (line.startsWith("data:")) { return { done: false, data: parseDataLine(line) }; } - if (line.startsWith(": ping")) { - return { done: true, data: undefined }; - } + // Lines starting with ":" are SSE comments (e.g. `: ping` keepalives) and + // must be skipped without interrupting the rest of the stream return { done: false, data: undefined }; }