add ensureAsyncIterable()

Author: gfx

src/decodeAsync.ts

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

src/utils/stream.ts

@@ -1,5 +1,7 @@
 // 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;
 }
@@ -19,3 +21,11 @@ export async function* asyncIterableFromStream<T>(stream: ReadableStream<T>): As
     reader.releaseLock();
   }
 }
+
+export function ensureAsyncIterabe<T>(streamLike: ReadableStreamLike<T>): AsyncIterable<T> {
+  if (isReadableStream(streamLike)) {
+    return asyncIterableFromStream(streamLike);
+  } else {
+    return streamLike;
+  }
+}