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

1 files changed, 32 insertions, 4 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);
+      }
+    },
+  });
+}