feat(streams/unstable): toLines()

streams/deno.json

@@ -16,6 +16,7 @@
     "./to-array-buffer": "./to_array_buffer.ts",
     "./to-blob": "./to_blob.ts",
     "./to-json": "./to_json.ts",
+    "./to-lines": "./to_lines.ts",
     "./to-text": "./to_text.ts",
     "./to-transform-stream": "./to_transform_stream.ts",
     "./zip-readable-streams": "./zip_readable_streams.ts"

streams/mod.ts

@@ -31,6 +31,7 @@ export * from "./text_line_stream.ts";
 export * from "./to_array_buffer.ts";
 export * from "./to_blob.ts";
 export * from "./to_json.ts";
+export * from "./to_lines.ts";
 export * from "./to_text.ts";
 export * from "./to_transform_stream.ts";
 export * from "./zip_readable_streams.ts";

streams/to_lines.ts

@@ -0,0 +1,76 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+/**
+ * Converts a ReadableStream of `Uint8Array` or `string` into an
+ * `AsyncGenerator` of `string`, where each value is divided by a newline at
+ * `\n` or `\r\n`. Trimming the last line if it is empty.
+ *
+ * @param readable A `ReadableStream` of `Uint8Array` or `string`.
+ * @returns An `AsyncGenerator<string>`
+ *
+ * @example JSON Lines
+ * ```ts
+ * import { toLines } from "@std/streams/to-lines";
+ * import { assertEquals } from "@std/assert/assert-equals";
+ *
+ * const readable = ReadableStream.from([
+ *   '{"name": "Alice", "age": ',
+ *   '30}\r\n{"name": "Bob", "age"',
+ *   ": 25}\n",
+ * ]);
+ *
+ * type Person = { name: string, age: number };
+ *
+ * const people: Person[] = []
+ * for await (const line of toLines(readable)) {
+ *   people.push(JSON.parse(line) as Person)
+ * }
+ *
+ * assertEquals(
+ *   people,
+ *   [{ "name": "Alice", "age": 30 }, { "name": "Bob", "age": 25 }],
+ * );
+ * ```
+ */
+export async function* toLines(
+  readable: ReadableStream<Uint8Array> | ReadableStream<string>,
+): AsyncGenerator<string> {
+  const reader = readable.getReader();
+  const decoder = new TextDecoder();
+  let buffer = "";
+  let index = 0;
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) {
+        break;
+      }
+      buffer += typeof value === "string" ? value : decoder.decode(value);
+      while (index < buffer.length) {
+        if (buffer[index] === "\n") {
+          yield buffer.slice(0, index - (buffer[index - 1] === "\r" ? 1 : 0));
+          buffer = buffer.slice(index + 1);
+          index = 0;
+        } else {
+          ++index;
+        }
+      }
+    }
+    if (buffer.length) {
+      yield buffer;
+    }
+  } catch (reason) {
+    await reader.cancel(reason);
+  }
+}
+
+console.log(
+  await Array.fromAsync(toLines(ReadableStream.from([
+    "qwertzu",
+    "iopasd\r\nmnbvc",
+    "xylk\rjhgfds\napoiuzt\r",
+    "qwr\r09ei\rqwrjiowqr\r",
+    "\nrewq0987\n\n654321",
+    "\nrewq0987\r\n\r\n654321\r",
+  ]))),
+);

streams/to_lines_test.ts

@@ -0,0 +1,53 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+import { toLines } from "./to_lines.ts";
+import { assertEquals } from "@std/assert";
+
+Deno.test("toLines parses simple input", async () => {
+  const stream = ReadableStream.from([
+    "qwertzu",
+    "iopasd\r\nmnbvc",
+    "xylk\rjhgfds\napoiuzt\r",
+    "qwr\r09ei\rqwrjiowqr\r",
+    "\nrewq0987\n\n654321",
+    "\nrewq0987\r\n\r\n654321\r",
+  ]);
+
+  assertEquals(await Array.fromAsync(toLines(stream)), [
+    "qwertzuiopasd",
+    "mnbvcxylk\rjhgfds",
+    "apoiuzt\rqwr\r09ei\rqwrjiowqr",
+    "rewq0987",
+    "",
+    "654321",
+    "rewq0987",
+    "",
+    "654321\r",
+  ]);
+
+  const stream2 = ReadableStream.from("rewq0987\r\n\r\n654321\n");
+
+  assertEquals(await Array.fromAsync(toLines(stream2)), [
+    "rewq0987",
+    "",
+    "654321",
+  ]);
+});
+
+Deno.test("toLines parses large chunks", async () => {
+  const totalLines = 20_000;
+  const stream = ReadableStream.from("\n".repeat(totalLines));
+  const lines = await Array.fromAsync(toLines(stream));
+
+  assertEquals(lines.length, totalLines);
+  assertEquals(lines, Array.from({ length: totalLines }).fill(""));
+});
+
+Deno.test("toLines converts Uint8Array chunks to strings", async () => {
+  const stream = ReadableStream.from([
+    Uint8Array.from([72, 101, 108, 108, 111]),
+    Uint8Array.from([10, 87, 111, 114, 108, 100, 33, 10]),
+  ]);
+
+  assertEquals(await Array.fromAsync(toLines(stream)), ["Hello", "World!"]);
+});