From 1417e038532308e43779c6546a639c15e2a58655 Mon Sep 17 00:00:00 2001 From: Ardis Lu Date: Sun, 19 Oct 2025 22:39:11 -0700 Subject: [PATCH] Support ReadableStream from transformation --- examples/README.md | 2 ++ examples/data/ex1_10-in.csv | 6 ++++ examples/data/ex1_10-out.csv | 8 +++++ examples/ex1_10.js | 33 +++++++++++++++++ src/core.js | 17 ++++++--- test/core.test.js | 69 ++++++++++++++++++++++++++++++++++++ 6 files changed, 130 insertions(+), 5 deletions(-) create mode 100644 examples/data/ex1_10-in.csv create mode 100644 examples/data/ex1_10-out.csv create mode 100644 examples/ex1_10.js diff --git a/examples/README.md b/examples/README.md index 5a7a902..4047887 100644 --- 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 index 0000000..1d2364a --- /dev/null +++ b/examples/data/ex1_10-in.csv @@ -0,0 +1,6 @@ +columnA +0 +1 +2 +3 +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 index 0000000..a40ca3b --- /dev/null +++ b/examples/data/ex1_10-out.csv @@ -0,0 +1,8 @@ +columnA +0: streamed 1 +0: streamed 2 +0: streamed 3 +1: streamed 1 +1: streamed 2 +1: streamed 3 +2: streamed 1 diff --git a/examples/ex1_10.js b/examples/ex1_10.js new file mode 100644 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 4e49125..8490099 100644 --- 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} row A row of input CSV data before transformation. - * @returns {TransformationOutput|Promise} + * @returns {TransformationOutput|Promise|ReadableStream} */ /** * A function to process multiple rows of CSV data from `CSVReader`. * @callback TransformationFunctionBatch * @param {Array>} rows Multiple rows of input CSV data. - * @returns {TransformationOutput|Promise} + * @returns {TransformationOutput|Promise|ReadableStream} */ /** @@ -240,7 +240,7 @@ export class CSVReader { * @param {Array} 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} + * @returns {TransformationOutput|Promise|ReadableStream} */ /** @@ -249,7 +249,7 @@ export class CSVReader { * @param {Array>} 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} + * @returns {TransformationOutput|Promise|ReadableStream} */ /** @@ -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 cead01f..0d3f708 100644 --- 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); -- 2.47.3