feat: add full NodeJS.ReadableStream support

see also: microsoft/TypeScript#420 (comment)
thanks: @nthypes @ericmasiello

Author: ctjlewis

.github/workflows/ci.yml

@@ -16,8 +16,8 @@ jobs:
     strategy:
       matrix:
         nodejs: [
-          # 14, 
-          # 16,
+          14, 
+          16,
           18,
           # "lts/*"
         ]

README.md

@@ -1,106 +1,18 @@
 # `yield-stream`
 
-- **Docs: https://yield-stream.vercel.app**
+[**Github**](https://github.com/gptlabs/yield-stream) |
+[**NPM**](https://npmjs.com/package/yield-stream) |
+[**Docs**](https://yield-stream.vercel.app)
 
-A small library for switching between streams, generators, and arrays.
+A small library for switching between streams, generators, and arrays. See docs
+for details.
 
+### Note: Using `NodeJS.ReadableStream`
 
-```ts
-/**
- * `compose(f, g, h, ...)` returns a generator function `G(data)` that yields
- * all `(f · g · h · ...)(data)`.
- *
- * @note Used to compose multiple transforms into a `pipeline`.
- */
-export const compose = <T>(
-  ...generators: GeneratorFn<T>[]
-): GeneratorFn<T> => {
-  return generators.reduce(
-    (prev, next) => async function* (data) {
-      for await (const chunk of prev(data)) {
-        yield* next(chunk);
-      }
-    },
-  );
-};
-
-/**
- * Accepts a stream and transforms and returns a stream of the transformed
- * chunks. Transforms can yield multiple chunks per input chunk.
- */
-export const pipeline = <T>(
-  stream: ReadableStream<T>,
-  ...transforms: GeneratorFn<T>[]
-): ReadableStream<T> => {
-  const composed = compose(...transforms);
-  return generateStream(
-    async function* () {
-      for await (const chunk of yieldStream(stream)) {
-        yield* composed(chunk);
-      }
-    }
-  );
-};
-
-/**
- * Accepts a stream and yields all of its chunks.
- */
-export const yieldStream = async function* <T>(
-  stream: ReadableStream<T>,
-  controller?: AbortController
-) {
-  const reader = stream.getReader();
-  while (true) {
-    if (controller?.signal.aborted) {
-      break;
-    }
-
-    const { done, value } = await reader.read();
-    if (done) {
-      break;
-    }
-
-    yield value;
-  }
-};
+By default, this library uses WHATWG `ReadableStream`, which is only available
+on Node 18+. If you are on an older version of Node or otherwise need to use
+`NodeJS.ReadableStream`, import from:
 
-/**
- * Accepts a generator function and streams its outputs.
- */
-export const generateStream = <T, TReturn, D>(
-  G: StreamGenerator<D, T, TReturn>,
-  data?: D
-): ReadableStream<T> => {
-  return new ReadableStream<T>({
-    async start(controller) {
-      for await (const chunk of G(data)) {
-        controller.enqueue(chunk);
-      }
-      controller.close();
-    },
-  });
-};
-
-/**
- * Accepts an array and returns a stream of its items.
- */
-export const streamArray = <T>(array: T[]): ReadableStream<T> => {
-  return generateStream(function* () {
-    for (const item of array) {
-      yield item;
-    }
-  });
-};
-
-/**
- * Accepts a stream and yields a growing buffer of all chunks received.
- */
-export const buffer = async function* <T>(stream: ReadableStream<T>) {
-  const buffer: T[] = [];
-
-  for await (const chunk of yieldStream(stream)) {
-    buffer.push(chunk);
-    yield buffer;
-  }
-};
+```ts
+import { yieldStream } from "yield-stream/node";
 ```

node.d.ts

@@ -0,0 +1 @@
+export * from "./dist/platforms/node";

node.js

@@ -0,0 +1 @@
+export * from "./dist/platforms/node.js";

package.json

@@ -9,10 +9,13 @@
   },
   "types": "dist/index.d.ts",
   "files": [
-    "dist"
+    "dist",
+    "node.d.ts",
+    "node.js"
   ],
   "exports": {
     "./package.json": "./package.json",
+    "./node": "./node.js",
     ".": "./dist/index.js",
     "./*": "./dist/*/index.js"
   },
@@ -41,4 +44,4 @@
     "shim-streams": "^0.0.2",
     "web-streams-polyfill": "^3.2.1"
   }
-}
+}

src/index.ts

@@ -1,2 +1,5 @@
-export * from "./lib";
-export * from "./types";
+/**
+ * `.` entrypoint is Edge by default. NodeJS.ReadableStream version available at
+ * `./node`.
+ */
+export * from "./platforms/edge";

src/lib.ts renamed to src/lib/edge.ts

@@ -1,21 +1,21 @@
-import { GeneratorFn, StreamGenerator } from "./types";
+import { GeneratorFn, StreamGenerator } from "../types";
+import { compose, generateArray } from "./shared";
 
 /**
- * `compose(f, g, h, ...)` returns a generator function `G(data)` that yields
- * all `(f · g · h · ...)(data)`.
- *
- * @note Used to compose multiple transforms into a `pipeline`.
+ * Accepts a generator function and streams its outputs.
  */
-export const compose = <Chunk>(
-  ...generators: GeneratorFn<Chunk>[]
-): GeneratorFn<Chunk> => {
-  return generators.reduce(
-    (prev, next) => async function* (data) {
-      for await (const chunk of prev(data)) {
-        yield* next(chunk);
+export const generateStream = <Chunk, Return, Data>(
+  G: StreamGenerator<Data, Chunk, Return>,
+  data?: Data
+): ReadableStream<Chunk> => {
+  return new ReadableStream<Chunk>({
+    async start(controller) {
+      for await (const chunk of G(data)) {
+        controller.enqueue(chunk);
       }
+      controller.close();
     },
-  );
+  });
 };
 
 /**
@@ -36,6 +36,15 @@ export const pipeline = <Chunk>(
   );
 };
 
+/**
+ * Accepts an array and returns a stream of its items.
+ */
+export const streamArray = <Chunk>(
+  array: Chunk[]
+): ReadableStream<Chunk> => {
+  return generateStream(generateArray(array));
+};
+
 /**
  * Accepts a stream and yields all of its chunks.
  */
@@ -58,36 +67,6 @@ export const yieldStream = async function* <Chunk>(
   }
 };
 
-/**
- * Accepts a generator function and streams its outputs.
- */
-export const generateStream = <Chunk, Return, Data>(
-  G: StreamGenerator<Data, Chunk, Return>,
-  data?: Data
-): ReadableStream<Chunk> => {
-  return new ReadableStream<Chunk>({
-    async start(controller) {
-      for await (const chunk of G(data)) {
-        controller.enqueue(chunk);
-      }
-      controller.close();
-    },
-  });
-};
-
-/**
- * Accepts an array and returns a stream of its items.
- */
-export const streamArray = <Chunk>(
-  array: Chunk[]
-): ReadableStream<Chunk> => {
-  return generateStream(function* () {
-    for (const item of array) {
-      yield item;
-    }
-  });
-};
-
 /**
  * Accepts a stream and yields a growing buffer of all chunks received.
  */

src/lib/node.ts

@@ -0,0 +1,73 @@
+import { Readable } from "stream";
+import { GeneratorFn, StreamGenerator } from "../types";
+import { compose, generateArray } from "./shared";
+
+/**
+ * Accepts a generator function and returns a NodeJS.ReadableStream of its
+ * outputs.
+ */
+export const generateStream = <Chunk, Return, Data>(
+  G: StreamGenerator<Data, Chunk, Return>,
+  data?: Data
+): NodeJS.ReadableStream => {
+  const readable = Readable.from(G(data));
+  return readable;
+};
+
+
+/**
+ * Accepts a stream and yields all of its chunks.
+ */
+export const yieldStream = async function* <Chunk>(
+  stream: NodeJS.ReadableStream,
+  controller?: AbortController
+): AsyncGenerator<Chunk> {
+  for await (const chunk of stream) {
+    if (controller?.signal.aborted) {
+      break;
+    }
+
+    yield chunk as Chunk;
+  }
+};
+
+/**
+ * Accepts a stream and transforms and returns a stream of the transformed
+ * chunks. Transforms can yield multiple chunks per input chunk.
+ */
+export const pipeline = <Chunk>(
+  stream: NodeJS.ReadableStream,
+  ...transforms: GeneratorFn<Chunk>[]
+): NodeJS.ReadableStream => {
+  const composed = compose(...transforms);
+  return generateStream(
+    async function* () {
+      for await (const chunk of yieldStream<Chunk>(stream)) {
+        yield* composed(chunk);
+      }
+    }
+  );
+};
+
+/**
+ * Accepts an array and returns a stream of its items.
+ */
+export const streamArray = <Chunk>(
+  array: Chunk[]
+): NodeJS.ReadableStream => {
+  return generateStream(generateArray(array));
+};
+
+/**
+ * Accepts a stream and yields a growing buffer of all chunks received.
+ */
+export const buffer = async function* <Chunk>(
+  stream: NodeJS.ReadableStream
+) {
+  const buffer: Chunk[] = [];
+
+  for await (const chunk of yieldStream<Chunk>(stream)) {
+    buffer.push(chunk);
+    yield buffer;
+  }
+};

src/lib/shared.ts

@@ -0,0 +1,32 @@
+import { GeneratorFn } from "../types";
+
+/**
+ * `compose(f, g, h, ...)` returns a generator function `G(data)` that yields
+ * all `(f · g · h · ...)(data)`.
+ *
+ * @note Used to compose multiple transforms into a `pipeline`.
+ */
+export const compose = <Chunk>(
+  ...generators: GeneratorFn<Chunk>[]
+): GeneratorFn<Chunk> => {
+  return generators.reduce(
+    (prev, next) => async function* (data) {
+      for await (const chunk of prev(data)) {
+        yield* next(chunk);
+      }
+    },
+  );
+};
+
+/**
+ * Accepts an array and returns a generator function that yields its items.
+ */
+export const generateArray = <Chunk>(
+  array: Chunk[]
+) => {
+  return function* () {
+    for (const item of array) {
+      yield item;
+    }
+  };
+};