Merge pull request #3 from lambdalisue/pipe_from

Add `pipeThroughFrom` function to transform writable stream like `pipeThrough` method of readable streams

README.md

@@ -10,6 +10,30 @@ This is a [Deno][deno] module that provides utilities for handling Streams API.
 
 ## Usage
 
+### pipeThroughFrom
+
+Pipes the readable side of a `TransformStream` to a `WritableStream`. Returns
+the writable side of the `TransformStream` for further piping.
+
+```ts
+import { channel } from "./channel.ts";
+import { collect } from "./collect.ts";
+import { pipeThroughFrom } from "./pipe_through_from.ts";
+
+const encoder = new TextEncoder();
+const output = channel<string>();
+const stream = pipeThroughFrom(output.writer, new TextDecoderStream());
+const writer = stream.getWriter();
+
+await writer.write(encoder.encode("Hello"));
+await writer.write(encoder.encode("World"));
+await writer.close();
+writer.releaseLock();
+
+const result = await collect(output.reader);
+console.log(result); // ["Hello", "World"]
+```
+
 ### channel
 
 `channel` creates a new channel, which is a pair of a readable and writable
@@ -20,7 +44,7 @@ import { channel } from "./channel.ts";
 import { push } from "./push.ts";
 import { pop } from "./pop.ts";
 
-const [reader, writer] = channel<number>();
+const { reader, writer } = channel<number>();
 
 await push(writer, 1);
 await push(writer, 2);

deno.jsonc

@@ -1,6 +1,6 @@
 {
   "tasks": {
-    "test": "deno test --no-lock --unstable -A --parallel",
+    "test": "deno test --no-lock --unstable -A --parallel --doc",
     "check": "deno check --no-lock --unstable $(find . -name '*.ts')",
     "upgrade": "deno run --no-lock -A https://deno.land/x/udd/main.ts $(find . -name '*.ts')"
   }

mod.ts

@@ -1,5 +1,6 @@
 export * from "./channel.ts";
 export * from "./collect.ts";
+export * from "./pipe_through_from.ts";
 export * from "./pop.ts";
 export * from "./provide.ts";
 export * from "./push.ts";

pipe_through_from.ts

@@ -0,0 +1,36 @@
+/**
+ * Pipes the readable side of a TransformStream to a WritableStream.
+ * Returns the writable side of the TransformStream for further piping.
+ *
+ * ```ts
+ * import { channel } from "./channel.ts";
+ * import { collect } from "./collect.ts";
+ * import { pipeThroughFrom } from "./pipe_through_from.ts";
+ *
+ * const encoder = new TextEncoder();
+ * const output = channel<string>();
+ * const stream = pipeThroughFrom(output.writer, new TextDecoderStream());
+ * const writer = stream.getWriter();
+ *
+ * await writer.write(encoder.encode("Hello"));
+ * await writer.write(encoder.encode("World"));
+ * await writer.close();
+ * writer.releaseLock();
+ *
+ * const result = await collect(output.reader);
+ * console.log(result); // ["Hello", "World"]
+ * ```
+ *
+ * @template I The type of data that the readable side of the TransformStream accepts.
+ * @template O The type of data that the TransformStream transforms the input data into.
+ * @param {WritableStream<O>} stream The destination WritableStream to pipe the data into.
+ * @param {TransformStream<I, O>} transform The TransformStream that transforms the input data.
+ * @returns {WritableStream<I>} The writable side of the TransformStream for further piping.
+ */
+export function pipeThroughFrom<I, O>(
+  stream: WritableStream<O>,
+  transform: TransformStream<I, O>,
+): WritableStream<I> {
+  transform.readable.pipeTo(stream);
+  return transform.writable;
+}

pipe_through_from_test.ts

@@ -0,0 +1,22 @@
+import { assertEquals } from "https://deno.land/[email protected]/testing/asserts.ts";
+import { collect } from "./collect.ts";
+import { channel } from "./channel.ts";
+import { pipeThroughFrom } from "./pipe_through_from.ts";
+
+Deno.test("pipeThroughFrom", async (t) => {
+  await t.step(
+    "returns a writable stream that pipe through to the specified writable stream",
+    async () => {
+      const encoder = new TextEncoder();
+      const output = channel<string>();
+      const stream = pipeThroughFrom(output.writer, new TextDecoderStream());
+      const writer = stream.getWriter();
+      await writer.write(encoder.encode("Hello"));
+      await writer.write(encoder.encode("World"));
+      await writer.close();
+      writer.releaseLock();
+      const result = await collect(output.reader);
+      assertEquals(result, ["Hello", "World"]);
+    },
+  );
+});