fix: handle exceptions in transforms properly in streaming APIs

Author: juanjoDiaz

build.js

@@ -27,7 +27,7 @@ const replaceDependenciesByJsdelivr = {
     });
 
     const dependencies = {
-      '@streamparser/json': 'https://cdn.jsdelivr.net/npm/@streamparser/[email protected].8/dist/mjs/index.mjs',
+      '@streamparser/json': 'https://cdn.jsdelivr.net/npm/@streamparser/[email protected].9/dist/mjs/index.mjs',
       'lodash.get': 'https://cdn.jsdelivr.net/gh/lodash/lodash@master/get.js'
     };
 

dist/cdn/plainjs/StreamParser.js

@@ -1,5 +1,5 @@
 // packages/plainjs/src/StreamParser.js
-import { Tokenizer, TokenParser, TokenType } from "https://cdn.jsdelivr.net/npm/@streamparser/[email protected].8/dist/mjs/index.mjs";
+import { Tokenizer, TokenParser, TokenType } from "https://cdn.jsdelivr.net/npm/@streamparser/[email protected].9/dist/mjs/index.mjs";
 import JSON2CSVBase from "./BaseParser.js";
 var JSON2CSVStreamParser = class extends JSON2CSVBase {
   constructor(opts, asyncOpts) {

package-lock.json

@@ -1,4 +1,5 @@
 import os from 'os';
+import { Writable } from 'stream';
 
 import TestRunner from '@json2csv/test-helpers/TestRunner.js';
 import { forceLfEol } from '@json2csv/test-helpers/utils.js';
@@ -890,6 +891,39 @@ export default function (jsonFixtures, csvFixtures) {
     t.equal(csv, csvFixtures.defaultCustomTransform);
   });
 
+  testRunner.add('should handle errors in transforms correctly', async (t) => {
+    const outputStream = new Writable({
+      write(chunk, encoding, callback) {
+        callback();
+      },
+    });
+
+    const opts = {
+      transforms: [
+        (row) => {
+          if (row.carModel === 'Mercedes') {
+            throw new Error('Mercerdes not allowed');
+          }
+
+          return row;
+        },
+      ],
+    };
+
+    const parser = new Parser(opts);
+    const transformation = parser.parse(jsonFixtures.default());
+    const promise = new Promise((res) => {
+      transformation.on('end', () => {
+        t.fail('Exception expected');
+        res();
+      });
+      transformation.on('error', (err) => res(err));
+    });
+
+    transformation.pipe(outputStream);
+    await promise;
+  });
+
   // Formatters
 
   // Number

packages/node/test/AsyncParser.js

@@ -1,4 +1,5 @@
 import os from 'os';
+import { Writable } from 'stream';
 
 import TestRunner from '@json2csv/test-helpers/TestRunner.js';
 import { forceLfEol } from '@json2csv/test-helpers/utils.js';
@@ -890,6 +891,39 @@ export default function (jsonFixtures, csvFixtures) {
     t.equal(csv, csvFixtures.defaultCustomTransform);
   });
 
+  testRunner.add('should handle errors in transforms correctly', async (t) => {
+    const outputStream = new Writable({
+      write(chunk, encoding, callback) {
+        callback();
+      },
+    });
+
+    const opts = {
+      transforms: [
+        (row) => {
+          if (row.carModel === 'Mercedes') {
+            throw new Error('Mercerdes not allowed');
+          }
+
+          return row;
+        },
+      ],
+    };
+
+    const parser = new Parser(opts);
+    const transformation = parser.parse(jsonFixtures.default());
+    const promise = new Promise((res) => {
+      transformation.on('end', () => {
+        t.fail('Exception expected');
+        res();
+      });
+      transformation.on('error', (err) => res(err));
+    });
+
+    transformation.pipe(outputStream);
+    await promise;
+  });
+
   // Formatters
 
   // Number

packages/node/test/AsyncParserInMemory.js

@@ -1,4 +1,5 @@
 import os from 'os';
+import { Writable } from 'stream';
 
 import TestRunner from '@json2csv/test-helpers/TestRunner.js';
 import { forceLfEol } from '@json2csv/test-helpers/utils.js';
@@ -885,6 +886,39 @@ export default function (jsonFixtures, csvFixtures) {
     t.equal(csv, csvFixtures.defaultCustomTransform);
   });
 
+  testRunner.add('should handle errors in transforms correctly', async (t) => {
+    const outputStream = new Writable({
+      write(chunk, encoding, callback) {
+        callback();
+      },
+    });
+
+    const opts = {
+      transforms: [
+        (row) => {
+          if (row.carModel === 'Mercedes') {
+            throw new Error('Mercerdes not allowed');
+          }
+
+          return row;
+        },
+      ],
+    };
+
+    const parser = new Parser(opts);
+    const transformation = jsonFixtures.default().pipe(parser);
+    const promise = new Promise((res) => {
+      transformation.on('end', () => {
+        t.fail('Exception expected');
+        res();
+      });
+      transformation.on('error', (err) => res(err));
+    });
+
+    transformation.pipe(outputStream);
+    await promise;
+  });
+
   // Formatters
 
   // Number

packages/node/test/Transform.js

@@ -38,7 +38,7 @@
   },
   "dependencies": {
     "@json2csv/formatters": "^6.0.0-alpha.4",
-    "@streamparser/json": "^0.0.8",
+    "@streamparser/json": "^0.0.9",
     "lodash.get": "^4.4.2"
   }
 }

packages/plainjs/package.json

@@ -857,6 +857,33 @@ export default function (jsonFixtures, csvFixtures) {
     t.equal(csv, csvFixtures.defaultCustomTransform);
   });
 
+  testRunner.add('should handle errors in transforms correctly', async (t) => {
+    const opts = {
+      transforms: [
+        (row) => {
+          if (row.carModel === 'Mercedes') {
+            throw new Error('Mercerdes not allowed');
+          }
+
+          return row;
+        },
+      ],
+    };
+
+    const parser = new Parser(opts);
+
+    const promise = new Promise((res) => {
+      parser.onEnd = () => {
+        t.fail('Exception expected');
+        res();
+      };
+      parser.onError = (err) => res(err);
+    });
+
+    parser.write('{ "carModel": "Mercedes" }');
+    await promise;
+  });
+
   // Formatters
 
   // Number