feat(streams/unstable): toLines() (#5121)

* feat(streams): add toLines

* chore(streams): remove testing code

Co-authored-by: Asher Gomez <ashersaupingomez@gmail.com>

* chore(streams): fmt

* adjust(streams): change toLines return type to `ReadableStream<string>`

- Change the return type from `AsyncGenerator<string>` to `ReadableStream<string>`
- Change `readable` to only be of type `ReadableStream<Uint8Array>`
- Add `options` argument of type `PipeOptions.`

* work

* work

---------

Co-authored-by: Asher Gomez <ashersaupingomez@gmail.com>

streams/deno.json

@@ -17,6 +17,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

@@ -32,6 +32,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,41 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+import { TextLineStream } from "./text_line_stream.ts";
+
+/**
+ * Converts a {@linkcode ReadableStream} of {@linkcode Uint8Array}s into one of
+ * lines delimited by `\n` or `\r\n`. Trims the last line if empty.
+ *
+ * @param readable A stream of {@linkcode Uint8Array}s.
+ * @param options Stream options.
+ * @returns A stream of lines delimited by `\n` or `\r\n`.
+ *
+ * @example Usage
+ * ```ts
+ * import { toLines } from "@std/streams/to-lines";
+ * import { assertEquals } from "@std/assert/equals";
+ *
+ * const readable = ReadableStream.from([
+ *   "qwertzu",
+ *   "iopasd\r\nmnbvc",
+ *   "xylk\rjhgfds\napoiuzt\r",
+ *   "qwr\r09ei\rqwrjiowqr\r",
+ * ]).pipeThrough(new TextEncoderStream());
+ *
+ * assertEquals(await Array.fromAsync(toLines(readable)), [
+ *   "qwertzuiopasd",
+ *   "mnbvcxylk\rjhgfds",
+ *   "apoiuzt\rqwr\r09ei\rqwrjiowqr\r",
+ * ]);
+ * ```
+ *
+ * @experimental
+ */
+export function toLines(
+  readable: ReadableStream<Uint8Array>,
+  options?: StreamPipeOptions,
+): ReadableStream<string> {
+  return readable
+    .pipeThrough(new TextDecoderStream())
+    .pipeThrough(new TextLineStream(), options);
+}

streams/to_lines_test.ts

@@ -0,0 +1,38 @@
+// 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",
+  ]).pipeThrough(new TextEncoderStream());
+
+  assertEquals(await Array.fromAsync(toLines(stream)), [
+    "qwertzuiopasd",
+    "mnbvcxylk\rjhgfds",
+    "apoiuzt\rqwr\r09ei\rqwrjiowqr",
+    "rewq0987",
+    "",
+    "654321",
+    "rewq0987",
+    "",
+    "654321\r",
+  ]);
+});
+
+Deno.test("toLines() parses large chunks", async () => {
+  const totalLines = 20_000;
+  const stream = ReadableStream.from(["\n".repeat(totalLines)]).pipeThrough(
+    new TextEncoderStream(),
+  );
+  const lines = await Array.fromAsync(toLines(stream));
+
+  assertEquals(lines.length, totalLines);
+  assertEquals(lines, Array.from({ length: totalLines }).fill(""));
+});