diff --git a/streams/delimiter.ts b/streams/delimiter.ts index 6242c6252941..56bd10b49955 100644 --- a/streams/delimiter.ts +++ b/streams/delimiter.ts @@ -210,6 +210,70 @@ export class DelimiterStream extends TransformStream { } } +/** Transform a stream into a stream where each chunk is divided by a given delimiter. + * + * ```ts + * import { TextDelimiterStream } from "./delimiter.ts"; + * const res = await fetch("https://example.com"); + * const parts = res.body! + * .pipeThrough(new TextDecoderStream()) + * .pipeThrough(new TextDelimiterStream("foo")); + * ``` + */ +export class TextDelimiterStream extends TransformStream { + #buf = ""; + #delimiter: string; + #inspectIndex = 0; + #matchIndex = 0; + #delimLPS: Uint8Array; + + constructor(delimiter: string) { + super({ + transform: (chunk, controller) => { + this.#handle(chunk, controller); + }, + flush: (controller) => { + controller.enqueue(this.#buf); + }, + }); + + this.#delimiter = delimiter; + this.#delimLPS = createLPS(new TextEncoder().encode(delimiter)); + } + + #handle( + chunk: string, + controller: TransformStreamDefaultController, + ) { + this.#buf += chunk; + let localIndex = 0; + while (this.#inspectIndex < this.#buf.length) { + if (chunk[localIndex] === this.#delimiter[this.#matchIndex]) { + this.#inspectIndex++; + localIndex++; + this.#matchIndex++; + if (this.#matchIndex === this.#delimiter.length) { + // Full match + const matchEnd = this.#inspectIndex - this.#delimiter.length; + const readyString = this.#buf.slice(0, matchEnd); + controller.enqueue(readyString); + // Reset match, different from KMP. + this.#buf = this.#buf.slice(this.#inspectIndex); + this.#inspectIndex = 0; + this.#matchIndex = 0; + } + } else { + if (this.#matchIndex === 0) { + this.#inspectIndex++; + localIndex++; + } else { + this.#matchIndex = this.#delimLPS[this.#matchIndex - 1]; + } + } + } + } +} + /** Generate longest proper prefix which is also suffix array. */ function createLPS(pat: Uint8Array): Uint8Array { const lps = new Uint8Array(pat.length); diff --git a/streams/delimiter_test.ts b/streams/delimiter_test.ts index 12fdcea9e797..754e8a893ed4 100644 --- a/streams/delimiter_test.ts +++ b/streams/delimiter_test.ts @@ -1,6 +1,11 @@ // Copyright 2018-2022 the Deno authors. All rights reserved. MIT license. -import { DelimiterStream, LineStream, TextLineStream } from "./delimiter.ts"; +import { + DelimiterStream, + LineStream, + TextDelimiterStream, + TextLineStream, +} from "./delimiter.ts"; import { assert, assertEquals } from "../testing/asserts.ts"; Deno.test("[streams] LineStream", async () => { @@ -106,3 +111,33 @@ Deno.test("[streams] DelimiterStream", async () => { const f = await reader.read(); assert(f.done); }); + +Deno.test("[streams] TextDelimiterStream", async () => { + const textStream = new ReadableStream({ + start(controller) { + controller.enqueue("qwertzu"); + controller.enqueue("iopasdfoomnbvc"); + controller.enqueue("xylkjhfoogfdsapfoooiuzt"); + controller.enqueue("rewq098765432fo"); + controller.enqueue("o349012i491290"); + controller.close(); + }, + }); + + const lines = textStream + .pipeThrough(new TextDelimiterStream("foo")); + const reader = lines.getReader(); + + const a = await reader.read(); + assertEquals(a.value, "qwertzuiopasd"); + const b = await reader.read(); + assertEquals(b.value, "mnbvcxylkjh"); + const c = await reader.read(); + assertEquals(c.value, "gfdsap"); + const d = await reader.read(); + assertEquals(d.value, "oiuztrewq098765432"); + const e = await reader.read(); + assertEquals(e.value, "349012i491290"); + const f = await reader.read(); + assert(f.done); +});