From 6ae221fb146b0f5801c06444cc4020ca77f36aa4 Mon Sep 17 00:00:00 2001
From: Juanjo Diaz <juanjo.diazmo@gmail.com>
Date: Sat, 22 Sep 2018 14:39:24 +0300
Subject: [PATCH] Add support for objectMode in the stream API

---
 README.md                 | 17 +++++++++++++++++
 lib/JSON2CSVTransform.js  | 20 +++++++++++++++++++-
 test/JSON2CSVTransform.js | 26 +++++++++++++++++++++++++-
 test/index.js             |  2 +-
 4 files changed, 62 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index dc079580..81cc48fa 100644
--- a/README.md
+++ b/README.md
@@ -231,6 +231,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
diff --git a/lib/JSON2CSVTransform.js b/lib/JSON2CSVTransform.js
index 3a506860..e5fce797 100644
--- a/lib/JSON2CSVTransform.js
+++ b/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;
+      }
+    };
   }
 
   /**
diff --git a/test/JSON2CSVTransform.js b/test/JSON2CSVTransform.js
index c735a984..23bccc17 100644
--- a/test/JSON2CSVTransform.js
+++ b/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'],
diff --git a/test/index.js b/test/index.js
index 4ea03aec..01d1b054 100644
--- a/test/index.js
+++ b/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();