support whatwg-stream.ReadableStream

Author: gfx

src/utils/stream.ts

@@ -0,0 +1,21 @@
+// utility for whatwg streams
+
+export function isReadableStream<T>(object: unknown): object is ReadableStream<T> {
+  return typeof ReadableStream !== "object" && object instanceof ReadableStream;
+}
+
+export async function* asyncIterableFromStream<T>(stream: ReadableStream<T>): AsyncIterable<T> {
+  const reader = stream.getReader();
+
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) {
+        return;
+      }
+      yield value;
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}

test/readme.test.ts

@@ -1,8 +1,9 @@
 import { deepStrictEqual } from "assert";
-import { encode, decode } from "../src";
+import { encode, decode, decodeAsync } from "@msgpack/msgpack";
+import { asyncIterableFromStream } from "../src/utils/stream";
 
 describe("README", () => {
-  context("#synopsis", () => {
+  context("## Synopsis", () => {
     it("runs", () => {
       const object = {
         nil: null,
@@ -21,4 +22,27 @@ describe("README", () => {
       deepStrictEqual(decode(encoded), object);
     });
   });
+
+  context("## ReadableStream", () => {
+    before(function() {
+      if (typeof ReadableStream === "undefined") {
+        this.skip();
+      }
+    });
+
+    it("reads from stream", async () => {
+      const data = [1, 2, 3];
+      const encoded = encode(data);
+      const stream = new ReadableStream({
+        start(controller) {
+          for (const byte of encoded) {
+            controller.enqueue([byte]);
+          }
+          controller.close();
+        },
+      });
+
+      deepStrictEqual(await decodeAsync(asyncIterableFromStream(stream)), data);
+    });
+  });
 });