Support ReadableStream from transformation

diff --git a/examples/README.md b/examples/README.md
index 5a7a9027be5d1dd319165b637e2bec4b46ac3765..40478874e2e496c4ce7248479e8a83cb81c5ef93 100644 (file)
--- a/examples/README.md
+++ b/examples/README.md
@@ -32,6 +32,8 @@ These examples demonstrate the most common, expected usage.
 
 [Example 1.9: Read and write a `.csv.gz` file](./ex1_9.js) ([input](./data/ex1_9-in.csv.gz), [output](./data/ex1_9-out.csv.gz))
 
+[Example 1.10: Return a ReadableStream from a transformation](./ex1_10.js) ([input](./data/ex1_10-in.csv), [output](./data/ex1_10-out.csv))
+
 ### Section 2: Error strategies
 
 These examples demonstrate using the `onError` option to handle transformations that may throw errors.

diff --git a/examples/data/ex1_10-in.csv b/examples/data/ex1_10-in.csv
new file mode 100644 (file)
index 0000000..1d2364a
--- /dev/null
+++ b/examples/data/ex1_10-in.csv
@@ -0,0 +1,6 @@
+columnA\r
+0\r
+1\r
+2\r
+3\r
+4
\ No newline at end of file

diff --git a/examples/data/ex1_10-out.csv b/examples/data/ex1_10-out.csv
new file mode 100644 (file)
index 0000000..a40ca3b
--- /dev/null
+++ b/examples/data/ex1_10-out.csv
@@ -0,0 +1,8 @@
+columnA\r
+0: streamed 1\r
+0: streamed 2\r
+0: streamed 3\r
+1: streamed 1\r
+1: streamed 2\r
+1: streamed 3\r
+2: streamed 1\r

diff --git a/examples/ex1_10.js b/examples/ex1_10.js
new file mode 100644 (file)
index 0000000..c7aea69
--- /dev/null
+++ b/examples/ex1_10.js
@@ -0,0 +1,33 @@
+// ## Example 1.10: Return a ReadableStream from a transformation
+// 
+// If the transformation itself involves a stream, you can directly return a ReadableStream to CSVTransformer.
+// This capability is useful if your transformation involves a network request; you can directly pass the
+// streaming response body as the output of your transformation.
+
+import { ReadableStream } from 'node:stream/web';
+import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js';
+/** @import { TransformationFunction } from '../src/index.js'; */
+
+let rowCount = 0;
+
+/** @type {TransformationFunction} */
+function stream(row) {
+  return ReadableStream.from((function* () {
+    yield [`${row[0]}: streamed 1`];
+    yield [`${row[0]}: streamed 2`];
+
+    // Errors produced by the ReadableStream are NOT be handled by CSVTransformer's onError, use .catch or implement
+    // your own error handling this instead. Note some rows yielded immediately before the error may be dropped.
+    if (rowCount === 2) {
+      throw new Error();
+    }
+
+    yield [`${row[0]}: streamed 3`];
+    rowCount++;
+  })());
+}
+
+await new CSVReader(new URL('./data/ex1_10-in.csv', import.meta.url))
+  .pipeThrough(new CSVTransformer(stream))
+  .pipeTo(new CSVWriter(new URL('./data/ex1_10-out.csv', import.meta.url)))
+  .catch(() => { }); // No-op to catch the intentional error for example purposes. Beware this will also catch other errors.

diff --git a/src/core.js b/src/core.js
index 4e4912523194db81040db3471224695fc87c5106..84900991c9580109a4d5be3acf203b45a544095e 100644 (file)
--- a/src/core.js
+++ b/src/core.js
@@ -224,14 +224,14 @@ export class CSVReader {
  * A function to process a row of CSV data from `CSVReader`.
  * @callback TransformationFunction
  * @param {Array<any>} row A row of input CSV data before transformation.
- * @returns {TransformationOutput|Promise<TransformationOutput>}
+ * @returns {TransformationOutput|Promise<TransformationOutput>|ReadableStream<TransformationOutput>}
  */
 
 /**
  * A function to process multiple rows of CSV data from `CSVReader`.
  * @callback TransformationFunctionBatch
  * @param {Array<Array<any>>} rows Multiple rows of input CSV data.
- * @returns {TransformationOutput|Promise<TransformationOutput>}
+ * @returns {TransformationOutput|Promise<TransformationOutput>|ReadableStream<TransformationOutput>}
  */
 
 /**
@@ -240,7 +240,7 @@ export class CSVReader {
  * @param {Array<any>} row The row passed to the transformation function which threw the error.
  * @param {Error} error The error thrown by the transformation function.
  * @param {TransformationFunction} fn The transformation function itself. This argument can be used to retry a transformation.
- * @returns {TransformationOutput|Promise<TransformationOutput>}
+ * @returns {TransformationOutput|Promise<TransformationOutput>|ReadableStream<TransformationOutput>}
  */
 
 /**
@@ -249,7 +249,7 @@ export class CSVReader {
  * @param {Array<Array<any>>} rows The rows of input CSV data passed to the transformation function which threw the error.
  * @param {Error} error The error thrown by the transformation function.
  * @param {TransformationFunctionBatch} fn The transformation function itself. This argument can be used to retry a transformation.
- * @returns {TransformationOutput|Promise<TransformationOutput>}
+ * @returns {TransformationOutput|Promise<TransformationOutput>|ReadableStream<TransformationOutput>}
  */
 
 /**
@@ -387,7 +387,14 @@ export class CSVTransformer extends TransformStream {
         }
       }
       else {
-        this.#enqueueRow(r.value, controller);
+        if (typeof r.value?.[Symbol.asyncIterator] === 'function') { // ReadableStream or ReadableStream-like
+          for await (const v of r.value) {
+            this.#enqueueRow(v, controller);
+          }
+        }
+        else {
+          this.#enqueueRow(r.value, controller);
+        }
       }
     }
     // Only clear these arrays AFTER all transform processing is complete. Otherwise downstream functions

diff --git a/test/core.test.js b/test/core.test.js
index cead01f036091de2c15942946cb5b7f5b98ef2e6..0d3f70818e0aa933abe736d087d197b1ab00e176 100644 (file)
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -563,6 +563,75 @@ suite('CSVTransformer', { concurrency: true }, () => {
       ]));
     deepStrictEqual(fn.mock.callCount(), 2);
   });
+  test('handles ReadableStream', { concurrency: true }, async (t) => {
+    await createCSVMockStream([
+      ['columnA'],
+      ['1'],
+      ['2']
+    ])
+      .pipeThrough(new CSVTransformer(row => {
+        return ReadableStream.from((function* () {
+          yield [`${row[0]}a`];
+          yield [`${row[0]}b`];
+          yield [`${row[0]}c`];
+        })());
+      }))
+      .pipeTo(csvStreamEqualWritable([
+        ['columnA'],
+        ['1a'],
+        ['1b'],
+        ['1c'],
+        ['2a'],
+        ['2b'],
+        ['2c']
+      ]));
+  });
+  test('handles ReadableStream with batches', { concurrency: true }, async (t) => {
+    await createCSVMockStream([
+      ['columnA'],
+      ['1'],
+      ['2']
+    ])
+      .pipeThrough(new CSVTransformer(row => {
+        return ReadableStream.from((function* () {
+          yield [`${row[0][0]} + ${row[1][0]} a`];
+          yield [`${row[0][0]} + ${row[1][0]} b`];
+          yield [`${row[0][0]} + ${row[1][0]} c`];
+        })());
+      }, { maxBatchSize: 2 }))
+      .pipeTo(csvStreamEqualWritable([
+        ['columnA'],
+        ['1 + 2 a'],
+        ['1 + 2 b'],
+        ['1 + 2 c']
+      ]));
+  });
+  test('handles ReadableStream with maxConcurrent', { concurrency: true }, async (t) => {
+    await createCSVMockStream([
+      ['columnA'],
+      ['1'],
+      ['2'],
+      ['3'],
+      ['4']
+    ])
+      .pipeThrough(new CSVTransformer(row => {
+        return ReadableStream.from((function* () {
+          yield [`${row[0]}a`];
+          yield [`${row[0]}b`];
+        })());
+      }, { maxConcurrent: 4 }))
+      .pipeTo(csvStreamEqualWritable([
+        ['columnA'],
+        ['1a'],
+        ['1b'],
+        ['2a'],
+        ['2b'],
+        ['3a'],
+        ['3b'],
+        ['4a'],
+        ['4b']
+      ]));
+  });
   test('works with instanceof', { concurrency: true }, async () => {
     const s = new CSVTransformer(r => r);
     deepStrictEqual(s instanceof CSVTransformer, true);