diff --git a/src/core/line-buffer.test.ts b/src/core/line-buffer.test.ts index ddd38b5d..4e4d7175 100644 --- a/src/core/line-buffer.test.ts +++ b/src/core/line-buffer.test.ts @@ -1,5 +1,6 @@ import { describe, it, expect } from "vitest"; import { LineBuffer } from "./line-buffer.js"; +import { MessageTooLargeError } from "./stream-limits.js"; const encoder = new TextEncoder(); const decoder = new TextDecoder(); @@ -55,4 +56,43 @@ describe("LineBuffer", () => { expect(pushAll(buffer, ["", "a", "", "\n"])).toEqual(["a"]); }); + + it("keeps a bare CR inside a line by default", () => { + const buffer = new LineBuffer(); + + expect(pushAll(buffer, ["a\rb\r\n", "c\r", "\n"])).toEqual(["a\rb", "c"]); + }); + + describe("with splitOnCarriageReturn", () => { + it("splits at CR, LF, and CRLF", () => { + const buffer = new LineBuffer(undefined, { splitOnCarriageReturn: true }); + + expect(pushAll(buffer, ["a\rb\nc\r\nd\n\re\r\r\nf"])).toEqual([ + "a", + "b", + "c", + "d", + "", + "e", + "", + ]); + expect(decoder.decode(buffer.flush())).toBe("f"); + }); + + it("joins a CRLF split across chunks", () => { + const buffer = new LineBuffer(undefined, { splitOnCarriageReturn: true }); + + expect( + pushAll(buffer, ["a\r", "\nb\r", "", "\nc\r", "\r\nd\r", "e", "\n"]), + ).toEqual(["a", "b", "c", "", "d", "e"]); + expect(buffer.flush()).toBeUndefined(); + }); + + it("applies the line limit without counting the line ending", () => { + const buffer = new LineBuffer(3, { splitOnCarriageReturn: true }); + + expect(pushAll(buffer, ["abc\r", "\nde", "f\r"])).toEqual(["abc", "def"]); + expect(() => pushAll(buffer, ["abcd\r"])).toThrow(MessageTooLargeError); + }); + }); }); diff --git a/src/core/line-buffer.ts b/src/core/line-buffer.ts index de421880..1e28d4ce 100644 --- a/src/core/line-buffer.ts +++ b/src/core/line-buffer.ts @@ -5,11 +5,18 @@ import { } from "./stream-limits.js"; const newline = 0x0a; +const carriageReturn = 0x0d; + +export interface LineBufferOptions { + /** Also end lines at a bare CR, as SSE allows. Defaults to false. */ + splitOnCarriageReturn?: boolean; +} /** - * Incrementally splits a byte stream into newline-terminated lines. + * Incrementally splits a byte stream into LF- or CRLF-terminated lines, and + * optionally CR-terminated lines. * - * Only the newly pushed chunk is scanned for newlines, so splitting costs + * Only the newly pushed chunk is scanned for line endings, so splitting costs * O(total bytes) no matter how many chunks a line spans. * * Chunks passed to {@link push} must not be mutated while its iterator is @@ -19,25 +26,57 @@ export class LineBuffer { /** Bytes of the current (incomplete) line, carried across chunks. */ readonly #pending: MessageBuffer; readonly #maxMessageBytes: number; + readonly #splitOnCarriageReturn: boolean; + /** Whether the last line ended at a CR whose LF may start the next chunk. */ + #afterCarriageReturn = false; - constructor(maxMessageBytes?: number) { + constructor( + maxMessageBytes?: number, + { splitOnCarriageReturn = false }: LineBufferOptions = {}, + ) { this.#maxMessageBytes = resolveMaxMessageBytes(maxMessageBytes); this.#pending = new MessageBuffer( Math.min(Number.MAX_SAFE_INTEGER, this.#maxMessageBytes + 1), ); + this.#splitOnCarriageReturn = splitOnCarriageReturn; } /** - * Consumes a chunk, yielding each complete line without its trailing - * LF or CRLF. + * Consumes a chunk, yielding each complete line without its line ending. */ *push(chunk: Uint8Array): Generator { let start = 0; + // A CRLF may be split across chunks, even with empty chunks in between. + if (this.#afterCarriageReturn && chunk.byteLength > 0) { + this.#afterCarriageReturn = false; + if (chunk[0] === newline) { + start = 1; + } + } + let newlineIndex = chunk.indexOf(newline, start); - while (newlineIndex !== -1) { - yield this.#takeLine(chunk.subarray(start, newlineIndex)); - start = newlineIndex + 1; - newlineIndex = chunk.indexOf(newline, start); + let carriageReturnIndex = this.#splitOnCarriageReturn + ? chunk.indexOf(carriageReturn, start) + : -1; + while (newlineIndex !== -1 || carriageReturnIndex !== -1) { + if ( + carriageReturnIndex === -1 || + (newlineIndex !== -1 && newlineIndex < carriageReturnIndex) + ) { + yield this.#takeLine(chunk.subarray(start, newlineIndex)); + start = newlineIndex + 1; + newlineIndex = chunk.indexOf(newline, start); + } else { + yield this.#takeLine(chunk.subarray(start, carriageReturnIndex)); + start = carriageReturnIndex + 1; + if (start === chunk.byteLength) { + this.#afterCarriageReturn = true; + } else if (chunk[start] === newline) { + start++; + newlineIndex = chunk.indexOf(newline, start); + } + carriageReturnIndex = chunk.indexOf(carriageReturn, start); + } } if (start < chunk.byteLength) { const tail = chunk.subarray(start); @@ -51,6 +90,7 @@ export class LineBuffer { * undefined if no bytes are buffered. */ flush(): Uint8Array | undefined { + this.#afterCarriageReturn = false; if (this.#pending.byteLength === 0) { return undefined; } @@ -58,6 +98,7 @@ export class LineBuffer { } clear(): void { + this.#afterCarriageReturn = false; this.#pending.clear(); } @@ -74,7 +115,9 @@ export class LineBuffer { const lastByte = tail.byteLength > 0 ? tail[tail.byteLength - 1] : this.#pending.lastByte; const byteLength = - this.#pending.byteLength + tail.byteLength - (lastByte === 0x0d ? 1 : 0); + this.#pending.byteLength + + tail.byteLength - + (lastByte === carriageReturn ? 1 : 0); if (byteLength > this.#maxMessageBytes) { this.clear(); throw new MessageTooLargeError(this.#maxMessageBytes); @@ -83,5 +126,7 @@ export class LineBuffer { } function stripCarriageReturn(line: Uint8Array): Uint8Array { - return line[line.byteLength - 1] === 0x0d ? line.subarray(0, -1) : line; + return line[line.byteLength - 1] === carriageReturn + ? line.subarray(0, -1) + : line; } diff --git a/src/transports/sse.test.ts b/src/transports/sse.test.ts index d85afc27..ba0fdc89 100644 --- a/src/transports/sse.test.ts +++ b/src/transports/sse.test.ts @@ -154,6 +154,58 @@ describe("SSE transport helpers", () => { ]); }); + it("parses events separated by bare CR blank lines", async () => { + const first: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } }; + const second: AnyMessage = { + jsonrpc: "2.0", + method: "session/update", + params: { sessionId: "s1" }, + }; + const body = + `data: ${JSON.stringify(first)}\r\r` + + `data: ${JSON.stringify(second)}\r\r`; + + await expect(collectMessages(streamFromChunks([body]))).resolves.toEqual([ + first, + second, + ]); + }); + + it("parses events with mixed line endings split byte by byte", async () => { + const first: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } }; + const second: AnyMessage = { + jsonrpc: "2.0", + method: "session/update", + params: { sessionId: "s1" }, + }; + const body = + 'data: {"jsonrpc":"2.0",\rdata: "id":1,\r\n' + + 'data: "result":{"ok":true}}\n\r' + + `data: ${JSON.stringify(second)}\r\n\n`; + + await expect( + collectMessages(streamFromChunks(chunkBytes(body, 1))), + ).resolves.toEqual([first, second]); + }); + + it("yields a CR-terminated event before the stream ends", async () => { + const message: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } }; + const body = new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode(`data: ${JSON.stringify(message)}\r\r`), + ); + }, + }); + const iterator = parseSseStream(body)[Symbol.asyncIterator](); + + await expect(iterator.next()).resolves.toEqual({ + value: message, + done: false, + }); + await iterator.return?.(); + }); + it("parses a final event without a trailing blank line", async () => { const message: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } }; diff --git a/src/transports/sse.ts b/src/transports/sse.ts index 9c92c948..d9bb47d5 100644 --- a/src/transports/sse.ts +++ b/src/transports/sse.ts @@ -9,7 +9,7 @@ import { const dataPrefix = new TextEncoder().encode("data:"); const dataSeparator = new Uint8Array([0x0a]); -const maxLineOverhead = 9; // UTF-8 BOM + "data: "; LineBuffer strips LF/CRLF. +const maxLineOverhead = 9; // UTF-8 BOM + "data: "; LineBuffer strips line endings. export function serializeSseEvent(msg: AnyMessage): string { return serializeSseJson(JSON.stringify(msg)); @@ -30,8 +30,10 @@ export async function* parseSseStream( ): AsyncIterable { const decoder = new TextDecoder("utf-8", { ignoreBOM: true }); const eventData = new MessageBuffer(maxMessageBytes); + // SSE lines may end in CRLF, LF, or a bare CR. const lines = new LineBuffer( Math.min(Number.MAX_SAFE_INTEGER, maxMessageBytes + maxLineOverhead), + { splitOnCarriageReturn: true }, ); const reader = body.getReader(); let hasData = false;