Allow transformation function to consume input row without output

diff --git a/src/core.js b/src/core.js
index 18920cb25c7e31b5b53166d102c681ba982fb30a..1fad6a2cc5f5e1804fa0afdcafb3a8b3299fb5d6 100644 (file)
--- a/src/core.js
+++ b/src/core.js
@@ -118,13 +118,15 @@ export function createCSVReadableStream(path) {
 /**
  * Create a new [TransformStream](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream) to process CSV data.
  * 
- * @param {function(Array<string>|string):Array<any>|string} fn A function to process a row of CSV data from `createCSVReadableStream`.
+ * @param {function(Array<string>|string):Array<any>|string|null} fn A function to process a row of CSV data from `createCSVReadableStream`.
  * 
  * If `options.rawInput` is `false` (default), the input will be a `Array<string>` representing the CSV row. If `options.rawInput` is `true`,
  * the input will be a JSON `string` representing the CSV row.
  * 
  * If `options.rawOutput` is `false` (default), the expected output is a `Array<string>` which will be serialized with `JSON.stringify()`. If
  * `options.rawOutput` is `true`, the raw output will be sent the next stream unmodified.
+ * 
+ * Return `null` to consume an input row without emitting an output row.
  * @param {createCSVTransformStreamOptions} options Object containing flags to configure the stream logic.
  * @returns {TransformStream} A `TransformStream` where each chunk is one row of the CSV file, after transformations applied by `fn()`.
  */
@@ -145,6 +147,11 @@ export function createCSVTransformStream(fn, options = {}) {
       else {
         out = await fn(row);
       }
+
+      if (out === null || out === undefined) { // Input row is consumed without emitting any output row
+        return;
+      }
+
       if (options.rawOutput || typeof out === 'string') {
         controller.enqueue(out);
       }

diff --git a/test/core.test.js b/test/core.test.js
index e52b6bc5942f15f6f5c4f611194cbd4650e7ae2e..26eaae79658c7c8f7fc8db981777eb409835bac8 100644 (file)
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -232,6 +232,25 @@ suite('createCSVTransformStream', { concurrency: true }, () => {
         ['536368', '676147', 1212515]
       ]));
   });
+  test('consumes input row without output row when null is returned', { concurrency: true }, async () => {
+    await createCSVMockStream([
+      ['columnA', 'columnB'],
+      ['should be deleted', 'should be deleted'],
+      ['a', 'b'],
+      ['should be deleted', 'should be deleted'],
+      ['a', 'b'],
+      ['should be deleted', 'should be deleted'],
+      ['a', 'b'],
+      ['should be deleted', 'should be deleted']
+    ])
+      .pipeThrough(createCSVTransformStream(r => r.length % 2 ? r : null))
+      .pipeTo(csvStreamEqualWritable([
+        ['columnA', 'columnB'],
+        ['a', 'b'],
+        ['a', 'b'],
+        ['a', 'b'],
+      ]));
+  });
 });
 
 suite('createCSVWritableStream', { concurrency: true }, () => {