diff --git a/streams/text_line_stream.ts b/streams/text_line_stream.ts index 3f40f0b1b13f..368aafef4d0d 100644 --- a/streams/text_line_stream.ts +++ b/streams/text_line_stream.ts @@ -65,7 +65,8 @@ export interface TextLineStreamOptions { * ``` */ export class TextLineStream extends TransformStream { - #currentLine = ""; + #pending: string[] = []; + #pendingTrailsCR = false; /** * Constructs a new instance. @@ -73,37 +74,72 @@ export class TextLineStream extends TransformStream { * @param options Options for the stream. */ constructor(options: TextLineStreamOptions = { allowCR: false }) { + const allowCR = options.allowCR ?? false; super({ transform: (chars, controller) => { - chars = this.#currentLine + chars; + if (chars.length === 0) return; - while (true) { - const lfIndex = chars.indexOf("\n"); - const crIndex = options.allowCR ? chars.indexOf("\r") : -1; + // Fast path: if no line can complete within this chunk, buffer it + // without rescanning the accumulated text. This keeps the total cost + // linear when a single line spans many chunks. (With `allowCR`, a + // buffered chunk-final "\r" resolves as soon as the next chunk + // arrives, so it forces processing.) + if ( + !chars.includes("\n") && + (!allowCR || (!this.#pendingTrailsCR && !chars.includes("\r"))) + ) { + this.#pending.push(chars); + return; + } + + this.#pending.push(chars); + const text = this.#pending.length === 1 + ? this.#pending[0]! + : this.#pending.join(""); + this.#pending.length = 0; + + let start = 0; + let lfIndex = text.indexOf("\n"); + let crIndex = allowCR ? text.indexOf("\r") : -1; + while (true) { if ( - crIndex !== -1 && crIndex !== (chars.length - 1) && + crIndex !== -1 && crIndex !== (text.length - 1) && (lfIndex === -1 || (lfIndex - 1) > crIndex) ) { - controller.enqueue(chars.slice(0, crIndex)); - chars = chars.slice(crIndex + 1); + controller.enqueue(text.slice(start, crIndex)); + start = crIndex + 1; + crIndex = text.indexOf("\r", start); continue; } if (lfIndex === -1) break; - const endIndex = chars[lfIndex - 1] === "\r" ? lfIndex - 1 : lfIndex; - controller.enqueue(chars.slice(0, endIndex)); - chars = chars.slice(lfIndex + 1); + const endIndex = text[lfIndex - 1] === "\r" ? lfIndex - 1 : lfIndex; + controller.enqueue(text.slice(start, endIndex)); + start = lfIndex + 1; + lfIndex = text.indexOf("\n", start); + if (crIndex !== -1 && crIndex < start) { + crIndex = text.indexOf("\r", start); + } } - this.#currentLine = chars; + if (start < text.length) { + const leftover = start === 0 ? text : text.slice(start); + this.#pending.push(leftover); + this.#pendingTrailsCR = allowCR && leftover.endsWith("\r"); + } else { + this.#pendingTrailsCR = false; + } }, flush: (controller) => { - if (this.#currentLine === "") return; - const currentLine = options.allowCR && this.#currentLine.endsWith("\r") - ? this.#currentLine.slice(0, -1) - : this.#currentLine; + if (this.#pending.length === 0) return; + let currentLine = this.#pending.length === 1 + ? this.#pending[0]! + : this.#pending.join(""); + if (allowCR && currentLine.endsWith("\r")) { + currentLine = currentLine.slice(0, -1); + } controller.enqueue(currentLine); }, }); diff --git a/streams/text_line_stream_test.ts b/streams/text_line_stream_test.ts index 3df593122953..3a07a57156a8 100644 --- a/streams/text_line_stream_test.ts +++ b/streams/text_line_stream_test.ts @@ -104,6 +104,35 @@ Deno.test("TextLineStream parses no final empty chunk with terminal newline", as ]); }); +Deno.test("TextLineStream parses a line spanning many chunks", async () => { + const longLine = "a".repeat(4096); + const chunks: string[] = []; + for (let i = 0; i < longLine.length; i += 7) { + chunks.push(longLine.slice(i, i + 7)); + } + chunks.push("\nrest"); + + for (const allowCR of [false, true]) { + const stream = ReadableStream.from(chunks.slice()) + .pipeThrough(new TextLineStream({ allowCR })); + assertEquals(await Array.fromAsync(stream), [longLine, "rest"]); + } +}); + +Deno.test("TextLineStream parses input with empty chunks", async () => { + const stream = ReadableStream.from(["", "abc\nd", "", "ef", ""]) + .pipeThrough(new TextLineStream()); + + assertEquals(await Array.fromAsync(stream), ["abc", "def"]); +}); + +Deno.test("TextLineStream with `allowCR` parses a chunk-final `\\r` once the next chunk arrives", async () => { + const stream = ReadableStream.from(["abc\r", "def"]) + .pipeThrough(new TextLineStream({ allowCR: true })); + + assertEquals(await Array.fromAsync(stream), ["abc", "def"]); +}); + Deno.test("TextLineStream parses no final empty chunk without terminal newline", async () => { const stream = ReadableStream.from([ "abc\n",