feat: Add support for objectMode in the stream API (#325)

README.md

@@ -245,6 +245,23 @@ json2csv
   .on('error', err => console.log(err));
 ```
 
+The stream API can also work on object mode. This is useful when you have an input stream in object mode or if you are getting JSON objects one by one and want to convert them to CSV as they come.
+
+```javascript
+    const input = new Readable({ objectMode: true });
+    input._read = () => {};
+    // myObjectEmitter is just a fake example representing anything that emit objects.
+    myObjectEmitter.on('object', obj => input.push(obj));
+    // Pushing a null close the stream
+    myObjectEmitter.end(()) => input.push(null));
+
+    const opts = {};
+    const transformOpts = { objectMode: true };
+
+    const json2csv = new Json2csvTransform(opts, transformOpts);
+    const processor = input.pipe(transform).pipe(output);
+```
+
 ### Javascript module examples
 
 #### Example `fields` option

lib/JSON2CSVTransform.js

@@ -17,7 +17,9 @@ class JSON2CSVTransform extends Transform {
     this._data = '';
     this._hasWritten = false;
 
-    if (this.opts.ndjson) {
+    if (this._readableState.objectMode) {
+      this.initObjectModeParse();
+    } else if (this.opts.ndjson) {
       this.initNDJSONParse();
     } else {
       this.initJSONParser();
@@ -30,7 +32,23 @@ class JSON2CSVTransform extends Transform {
     if (this.opts.fields) {
       this.pushHeader();
     }
+  }
+
+  /**
+   * Init the transform with a parser to process data in object mode.
+   * It receives JSON objects one by one and send them to `pushLine for processing.
+   */
+  initObjectModeParse() {
+    const transform = this;
 
+    this.parser = {
+      write(line) {
+        transform.pushLine(line);
+      },
+      getPendingData() {
+        return undefined;
+      }
+    };
   }
 
   /**

test/JSON2CSVTransform.js

@@ -3,7 +3,31 @@
 const Readable = require('stream').Readable;
 const Json2csvTransform = require('../lib/json2csv').Transform;
 
-module.exports = (testRunner, jsonFixtures, csvFixtures) => {
+module.exports = (testRunner, jsonFixtures, csvFixtures, inMemoryJsonFixtures) => {
+  testRunner.add('should handle object mode', (t) => {
+    const input = new Readable({ objectMode: true });
+    input._read = () => {};
+    inMemoryJsonFixtures.default.forEach(item => input.push(item));
+    input.push(null);
+
+    const opts = {
+      fields: ['carModel', 'price', 'color', 'transmission']
+    };
+    const transformOpts = { readableObjectMode: true, writableObjectMode: true };
+
+    const transform = new Json2csvTransform(opts, transformOpts);
+    const processor = input.pipe(transform);
+
+    let csv = '';
+    processor
+      .on('data', chunk => (csv += chunk.toString()))
+      .on('end', () => {
+        t.equal(csv, csvFixtures.ndjson);
+        t.end();
+      })
+      .on('error', err => t.notOk(true, err.message));
+  });
+
   testRunner.add('should handle ndjson', (t) => {
     const opts = {
       fields: ['carModel', 'price', 'color', 'transmission'],

test/index.js

@@ -41,7 +41,7 @@ Promise.all([
 
   CLI(testRunner, jsonFixtures, csvFixtures);
   JSON2CSVParser(testRunner, jsonFixtures, csvFixtures);
-  JSON2CSVTransform(testRunner, jsonFixturesStreams, csvFixtures);
+  JSON2CSVTransform(testRunner, jsonFixturesStreams, csvFixtures, jsonFixtures);
   parseNdjson(testRunner, jsonFixtures);
 
   testRunner.run();