support whatwg streams in decodeAsync() and decodeArrayStream()

Author: gfx

src/decodeAsync.ts

@@ -1,13 +1,18 @@
 import { Decoder } from "./Decoder";
 import { defaultDecodeOptions, DecodeOptions } from "./decode";
+import { isReadableStream, asyncIterableFromStream } from "./utils/stream";
 
 export type DecodeAsyncOptions = DecodeOptions;
 export const defaultDecodeAsyncOptions = defaultDecodeOptions;
 
+type StreamLike<T> = AsyncIterable<T> | ReadableStream<T>;
+
 export async function decodeAsync(
-  stream: AsyncIterable<Uint8Array | ArrayLike<number>>,
+  streamLike: StreamLike<Uint8Array | ArrayLike<number>>,
   options: DecodeAsyncOptions = defaultDecodeOptions,
 ): Promise<unknown> {
+  const stream = isReadableStream(streamLike) ? asyncIterableFromStream(streamLike) : streamLike;
+
   const decoder = new Decoder(
     options.extensionCodec,
     options.maxStrLength,
@@ -20,9 +25,11 @@ export async function decodeAsync(
 }
 
 export async function* decodeArrayStream(
-  stream: AsyncIterable<Uint8Array | ArrayLike<number>>,
+  streamLike: StreamLike<Uint8Array | ArrayLike<number>>,
   options: DecodeAsyncOptions = defaultDecodeOptions,
 ) {
+  const stream = isReadableStream(streamLike) ? asyncIterableFromStream(streamLike) : streamLike;
+
   const decoder = new Decoder(
     options.extensionCodec,
     options.maxStrLength,

src/utils/stream.ts

@@ -1,7 +1,7 @@
 // utility for whatwg streams
 
 export function isReadableStream<T>(object: unknown): object is ReadableStream<T> {
-  return typeof ReadableStream !== "object" && object instanceof ReadableStream;
+  return typeof ReadableStream !== "undefined" && object instanceof ReadableStream;
 }
 
 export async function* asyncIterableFromStream<T>(stream: ReadableStream<T>): AsyncIterable<T> {

test/readme.test.ts

@@ -1,6 +1,5 @@
 import { deepStrictEqual } from "assert";
-import { encode, decode, decodeAsync } from "@msgpack/msgpack";
-import { asyncIterableFromStream } from "../src/utils/stream";
+import { encode, decode } from "@msgpack/msgpack";
 
 describe("README", () => {
   context("## Synopsis", () => {
@@ -22,27 +21,4 @@ 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);
-    });
-  });
 });

test/whatwg-streams.test.ts

@@ -0,0 +1,45 @@
+import { deepStrictEqual } from "assert";
+import { decodeAsync, encode, decodeArrayStream } from "@msgpack/msgpack";
+import { constants } from "http2";
+
+describe("whatwg streams", () => {
+  before(function() {
+    if (typeof ReadableStream === "undefined") {
+      this.skip();
+    }
+  });
+
+  context("decodeAsync", 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();
+      },
+    });
+
+    const items = [];
+    for await (const item of decodeArrayStream(stream)) {
+      items.push(item);
+    }
+    deepStrictEqual(items, data);
+  });
+
+  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(stream), data);
+  });
+});