allow stream ponyfill as ReadableStream

Author: gfx

package-lock.json

@@ -87,6 +87,7 @@
     "ts-node": "^8.2.0",
     "tsconfig-paths": "^3.8.0",
     "typescript": "^3.5.1",
+    "web-streams-polyfill": "^2.0.3",
     "webpack": "^4.30.0",
     "webpack-cli": "^3.3.1"
   },

package.json

@@ -1,9 +1,13 @@
 // utility for whatwg streams
 
+// The living standard of whatwg streams says
+// ReadableStream is also AsyncIterable, but
+// as of June 2019, no browser implements it.
+// See https://streams.spec.whatwg.org/ for details
 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 function isAsyncIterable<T>(object: object): object is AsyncIterable<T> {
+  return (object as any)[Symbol.asyncIterator] != null;
 }
 
 export async function* asyncIterableFromStream<T>(stream: ReadableStream<T>): AsyncIterable<T> {
@@ -23,9 +27,9 @@ export async function* asyncIterableFromStream<T>(stream: ReadableStream<T>): As
 }
 
 export function ensureAsyncIterabe<T>(streamLike: ReadableStreamLike<T>): AsyncIterable<T> {
-  if (isReadableStream(streamLike)) {
-    return asyncIterableFromStream(streamLike);
-  } else {
+  if (isAsyncIterable(streamLike)) {
     return streamLike;
+  } else {
+    return asyncIterableFromStream(streamLike);
   }
 }

src/utils/stream.ts

@@ -1,6 +1,6 @@
 import { deepStrictEqual } from "assert";
 import { decodeAsync, encode, decodeArrayStream } from "@msgpack/msgpack";
-
+import { ReadableStream as PonyReadableStream } from "web-streams-polyfill/ponyfill";
 const isReadableStreamConstructorAvailable: boolean = (() => {
   try {
     // Edge <= 18 has ReadableStream but its constructor is not available
@@ -13,24 +13,28 @@ const isReadableStreamConstructorAvailable: boolean = (() => {
   }
 })();
 
-describe("whatwg streams", () => {
-  before(function() {
-    if (!isReadableStreamConstructorAvailable) {
-      this.skip();
-    }
-  });
+const MyReadableStream: typeof ReadableStream = isReadableStreamConstructorAvailable
+  ? ReadableStream
+  : PonyReadableStream;
 
-  it("decodeAsync", async () => {
+// Downgrade stream not to implement AsycIterable<T>
+function downgradeReadableStream(stream: ReadableStream) {
+  (stream as any)[Symbol.asyncIterator] = undefined;
+}
+
+describe("whatwg streams", () => {
+  it("decodeArrayStream", async () => {
     const data = [1, 2, 3];
     const encoded = encode(data);
-    const stream = new ReadableStream({
+    const stream = new MyReadableStream({
       start(controller) {
         for (const byte of encoded) {
           controller.enqueue([byte]);
         }
         controller.close();
       },
     });
+    downgradeReadableStream(stream);
 
     const items: Array<unknown> = [];
     for await (const item of decodeArrayStream(stream)) {
@@ -39,17 +43,18 @@ describe("whatwg streams", () => {
     deepStrictEqual(items, data);
   });
 
-  it("reads from stream", async () => {
+  it("decodeAsync", async () => {
     const data = [1, 2, 3];
     const encoded = encode(data);
-    const stream = new ReadableStream({
+    const stream = new MyReadableStream({
       start(controller) {
         for (const byte of encoded) {
           controller.enqueue([byte]);
         }
         controller.close();
       },
     });
+    downgradeReadableStream(stream);
 
     deepStrictEqual(await decodeAsync(stream), data);
   });