Merge pull request #43 from msgpack/whatwg-streams

support whatwg-stream.ReadableStream as decodeAsync() input source

Author: gfx

src/decodeAsync.ts

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

src/utils/stream.ts

@@ -0,0 +1,31 @@
+// utility for whatwg streams
+
+export type ReadableStreamLike<T> = AsyncIterable<T> | ReadableStream<T>;
+
+export function isReadableStream<T>(object: unknown): object is ReadableStream<T> {
+  return typeof ReadableStream !== "undefined" && 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();
+  }
+}
+
+export function ensureAsyncIterabe<T>(streamLike: ReadableStreamLike<T>): AsyncIterable<T> {
+  if (isReadableStream(streamLike)) {
+    return asyncIterableFromStream(streamLike);
+  } else {
+    return streamLike;
+  }
+}

test/readme.test.ts

@@ -1,8 +1,8 @@
 import { deepStrictEqual } from "assert";
-import { encode, decode } from "../src";
+import { encode, decode } from "@msgpack/msgpack";
 
 describe("README", () => {
-  context("#synopsis", () => {
+  context("## Synopsis", () => {
     it("runs", () => {
       const object = {
         nil: null,

test/whatwg-streams.test.ts

@@ -0,0 +1,44 @@
+import { deepStrictEqual } from "assert";
+import { decodeAsync, encode, decodeArrayStream } from "@msgpack/msgpack";
+
+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);
+  });
+});