feat(streams): TextDelimiterStream

streams/delimiter.ts

@@ -210,6 +210,70 @@ export class DelimiterStream extends TransformStream<Uint8Array, Uint8Array> {
   }
 }
 
+/** 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<string, string> {
+  #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<string>,
+  ) {
+    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);

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);
+});