Add onError callback option

diff --git a/examples/README.md b/examples/README.md
index 808c82a6a39b3ad8901c243396da7dc067a6828f..ec0724086022a6e0d59366a83a760a7105c61407 100644 (file)
--- a/examples/README.md
+++ b/examples/README.md
@@ -18,4 +18,6 @@ node ./examples/ex1.js
 
 [Example 4: Using `createCSVNormalizationStream` and `createCSVDenormalizationStream`](./ex4.js)
 
-[Example 5: Handling an irregularly-shaped CSV file](./ex5.js)
\ No newline at end of file
+[Example 5: Handling an irregularly-shaped CSV file](./ex5.js)
+
+[Example 6: Handling a flaky transformation (basic retry strategy)](./ex6.js)
\ No newline at end of file

diff --git a/examples/data/ex6-in.csv b/examples/data/ex6-in.csv
new file mode 100644 (file)
index 0000000..4dad4af
--- /dev/null
+++ b/examples/data/ex6-in.csv
@@ -0,0 +1,6 @@
+columnA\r
+1\r
+2\r
+3\r
+4\r
+5\r

diff --git a/examples/data/ex6-out.csv b/examples/data/ex6-out.csv
new file mode 100644 (file)
index 0000000..9bcd49e
--- /dev/null
+++ b/examples/data/ex6-out.csv
@@ -0,0 +1,6 @@
+columnA\r
+1: pass\r
+2: pass\r
+3: pass\r
+4: pass\r
+5: pass\r

diff --git a/examples/ex6.js b/examples/ex6.js
new file mode 100644 (file)
index 0000000..039de8c
--- /dev/null
+++ b/examples/ex6.js
@@ -0,0 +1,30 @@
+// ## Example 6: Handling a flaky transformation (basic retry strategy)
+// 
+// Pass a function to the `onError` option of `createCSVTransformStream` to catch errors thrown by the transformation
+// function.
+// 
+// The `onError` function may be used to implement graceful failure strategies (e.g., set the output row to a
+// placeholder), retry strategies (e.g., a truncated exponential backoff for flaky network requests), logging, etc.
+
+import { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/index.js';
+
+// Will immediately retry calling `fn` a maximum of `iterations` times
+function retry(row, fn, iterations) {
+  while (iterations--) {
+    try { return fn(row); }
+    catch { }
+  }
+}
+
+// Will fail 90% of the time
+function flaky(row) {
+  if (Math.random() < 0.9) {
+    throw new Error();
+  }
+  const [v] = row;
+  return [`${v}: pass`];
+}
+
+await createCSVReadableStream(new URL('./data/ex6-in.csv', import.meta.url))
+  .pipeThrough(createCSVTransformStream(flaky, { onError: (row, e, fn) => retry(row, fn, 1000) }))
+  .pipeTo(createCSVWritableStream(new URL('./data/ex6-out.csv', import.meta.url)));
\ No newline at end of file

diff --git a/src/core.js b/src/core.js
index 6cd25b567f016d9c9c02f3f5356857ec3574a4cc..63e9888676bbd578b9587db4819502ecd8b1c241 100644 (file)
--- a/src/core.js
+++ b/src/core.js
@@ -138,6 +138,9 @@ export function createCSVReadableStream(path) {
  * using `JSON.parse()` before being sent to `fn()`. The default value is `false`.
  * @property {boolean} [rawOutput=false] Set to `true` to send the raw return value of `fn()` to the next stream. Otherwise, the return value of
  * `fn()` will be serialized using `JSON.stringify()` before being sent to the next stream. The default value is `false`.
+ * @property {null|function(Array<string>|string,Error,function(Array<string>|string):Array<Array<any>>|Array<any>|string|null):Array<Array<any>>|Array<any>|string|null} [onError=null]
+ * Set to a function to catch errors thrown by the transformation function. The CSV row, the error that was thrown, and the transformation function
+ * itself will be passed to the `onError` function. The default value is `null` (errors will not be caught).
  */
 
 /**
@@ -160,6 +163,13 @@ export function createCSVTransformStream(fn, options = {}) {
   options.includeHeaders ??= false;
   options.rawInput ??= false;
   options.rawOutput ??= false;
+  options.onError ??= null;
+
+  async function wrappedFn(row) {
+    if (options.onError === null) { return await fn(row); }
+    try { return await fn(row); }
+    catch (e) { return await options.onError(row, e, fn); }
+  }
 
   let firstChunk = true;
   return new TransformStream({
@@ -168,10 +178,10 @@ export function createCSVTransformStream(fn, options = {}) {
       let out;
       if (firstChunk) {
         firstChunk = false;
-        out = options.includeHeaders ? await fn(row) : row;
+        out = options.includeHeaders ? await wrappedFn(row) : row;
       }
       else {
-        out = await fn(row);
+        out = await wrappedFn(row);
       }
 
       if (out === null || out === undefined) { // Input row is consumed without emitting any output row

diff --git a/test/core.test.js b/test/core.test.js
index 98256e531c74010b9c66279b2b4277b44a3b2c5f..35ae2ddbc7912bd8ee14adab67f86a9bf6697927 100644 (file)
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -1,5 +1,5 @@
 import { suite, test } from 'node:test';
-import { deepStrictEqual, throws } from 'node:assert/strict';
+import { deepStrictEqual, throws, rejects } from 'node:assert/strict';
 import { unlink } from 'node:fs/promises';
 import { normalize, resolve } from 'node:path';
 import { pathToFileURL } from 'node:url';
@@ -245,6 +245,33 @@ suite('createCSVTransformStream', { concurrency: true }, () => {
         ['a', 'b']
       ]));
   });
+  test('passes errors to onError function', { concurrency: true }, async () => {
+    await createCSVMockStream([
+      ['columnA'],
+      ['1'],
+      ['2']
+    ])
+      .pipeThrough(createCSVTransformStream(() => { throw new Error(); }, { onError: r => [`${r[0]}: caught`] }))
+      .pipeTo(csvStreamEqualWritable([
+        ['columnA'],
+        ['1: caught'],
+        ['2: caught']
+      ]));
+  });
+  test('bubbles up errors re-raised by onError function', { concurrency: true }, async () => {
+    await rejects(
+      createCSVMockStream([
+        ['columnA'],
+        ['1']
+      ])
+        .pipeThrough(createCSVTransformStream(() => { throw new Error('outer'); }, { onError: () => { throw new Error('inner') } }))
+        .pipeTo(csvStreamEqualWritable([
+          ['columnA'],
+          ['1']
+        ])),
+      { name: 'Error', message: 'inner' }
+    );
+  });
 });
 
 suite('createCSVWritableStream', { concurrency: true }, () => {