feat(std/io): ReadableStream from AsyncIterator & WritableStream from Writer (#8378)

2 files changed, 90 insertions, 9 deletions

diff --git a/std/io/streams.ts b/std/io/streams.ts
index 1d4e3e12c..2c4fbb6b9 100644
--- a/std/io/streams.ts
+++ b/std/io/streams.ts
@@ -1,7 +1,7 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 
-/** Create a `Writer` from a `WritablseStreamDefaultReader`. */
-export function fromStreamWriter(
+/** Create a `Writer` from a `WritableStreamDefaultReader`. */
+export function writerFromStreamWriter(
   streamWriter: WritableStreamDefaultWriter<Uint8Array>,
 ): Deno.Writer {
   return {
@@ -13,8 +13,8 @@ export function fromStreamWriter(
   };
 }
 
-/** Create a `Reader` from a `ReadableSteramDefaultReader`. */
-export function fromStreamReader(
+/** Create a `Reader` from a `ReadableStreamDefaultReader`. */
+export function readerFromStreamReader(
   streamReader: ReadableStreamDefaultReader<Uint8Array>,
 ): Deno.Reader {
   const buffer = new Deno.Buffer();
@@ -34,3 +34,31 @@ export function fromStreamReader(
     },
   };
 }
+
+/** Create a `WritableStream` from a `Writer`. */
+export function writableStreamFromWriter(
+  writer: Deno.Writer,
+): WritableStream<Uint8Array> {
+  return new WritableStream({
+    async write(chunk) {
+      await Deno.writeAll(writer, chunk);
+    },
+  });
+}
+
+/** Create a `ReadableStream` from an `AsyncIterator`. */
+export function readableStreamFromAsyncIterator<T>(
+  iterator: AsyncIterableIterator<T>,
+): ReadableStream<T> {
+  return new ReadableStream({
+    async pull(controller) {
+      const { value, done } = await iterator.next();
+
+      if (done) {
+        controller.close();
+      } else {
+        controller.enqueue(value);
+      }
+    },
+  });
+}
diff --git a/std/io/streams_test.ts b/std/io/streams_test.ts
index 81881ceb9..b36bf8938 100644
--- a/std/io/streams_test.ts
+++ b/std/io/streams_test.ts
@@ -1,7 +1,12 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 
 import { assert, assertEquals } from "../testing/asserts.ts";
-import { fromStreamReader, fromStreamWriter } from "./streams.ts";
+import {
+  readableStreamFromAsyncIterator,
+  readerFromStreamReader,
+  writableStreamFromWriter,
+  writerFromStreamWriter,
+} from "./streams.ts";
 
 function repeat(c: string, bytes: number): Uint8Array {
   assertEquals(c.length, 1);
@@ -21,7 +26,7 @@ Deno.test("toWriterCheck", async function (): Promise<void> {
   });
 
   const encoder = new TextEncoder();
-  const writer = fromStreamWriter(writableStream.getWriter());
+  const writer = writerFromStreamWriter(writableStream.getWriter());
 
   for (const chunk of chunks) {
     const n = await writer.write(encoder.encode(chunk));
@@ -46,7 +51,7 @@ Deno.test("toReaderCheck", async function (): Promise<void> {
   });
 
   const decoder = new TextDecoder();
-  const reader = fromStreamReader(readableStream.getReader());
+  const reader = readerFromStreamReader(readableStream.getReader());
 
   let i = 0;
 
@@ -91,7 +96,7 @@ Deno.test("toReaderBigChunksCheck", async function (): Promise<void> {
     },
   });
 
-  const reader = fromStreamReader(readableStream.getReader());
+  const reader = readerFromStreamReader(readableStream.getReader());
   const n = await Deno.copy(reader, writer, { bufSize });
 
   const expectedWritten = chunkSize * expected.length;
@@ -126,9 +131,57 @@ Deno.test("toReaderBigIrregularChunksCheck", async function (): Promise<void> {
     },
   });
 
-  const reader = fromStreamReader(readableStream.getReader());
+  const reader = readerFromStreamReader(readableStream.getReader());
 
   const n = await Deno.copy(reader, writer, { bufSize });
   assertEquals(n, expected.length);
   assertEquals(expected, writer.bytes());
 });
+
+Deno.test("toWritableCheck", async function (): Promise<void> {
+  const written: string[] = [];
+  const chunks: string[] = ["hello", "deno", "land"];
+  const decoder = new TextDecoder();
+
+  async function write(p: Uint8Array): Promise<number> {
+    written.push(decoder.decode(p));
+    return p.length;
+  }
+
+  const writableStream = writableStreamFromWriter({ write });
+
+  const encoder = new TextEncoder();
+  const streamWriter = writableStream.getWriter();
+  for (const chunk of chunks) {
+    await streamWriter.write(encoder.encode(chunk));
+  }
+
+  assertEquals(written, chunks);
+});
+
+Deno.test("toReadableCheck", async function (): Promise<void> {
+  const chunks: string[] = ["hello", "deno", "land"];
+  const expected = chunks.slice();
+  const readChunks: string[] = [];
+  const encoder = new TextEncoder();
+
+  async function read(p: Uint8Array): Promise<number | null> {
+    const chunk = chunks.shift();
+    if (chunk === undefined) {
+      return null;
+    } else {
+      const encoded = encoder.encode(chunk);
+      p.set(encoded);
+      return encoded.length;
+    }
+  }
+  const iter = Deno.iter({ read });
+  const writableStream = readableStreamFromAsyncIterator(iter);
+
+  const decoder = new TextDecoder();
+  for await (const chunk of writableStream.getIterator()) {
+    readChunks.push(decoder.decode(chunk));
+  }
+
+  assertEquals(expected, readChunks);
+});