Can now handle null in streams

lib/pjs.js

@@ -7,6 +7,7 @@ var stream        = require('stream');
 var Stringify     = require('streaming-json-stringify');
 var StringDecoder = require('string_decoder').StringDecoder;
 var utils         = require('./utils');
+var _nullObject   = utils._nullObject;
 
 /**
  * Accepts an expression to evaluate, and two booleans. The first indicates
@@ -63,7 +64,8 @@ exports.map = function(expression, outputString, explicit) {
   i = 0;
 
   update = function($, i) {
-    return eval(exp);
+    var _result = eval(exp);
+    return (_result === null) ? _nullObject : _result;
   };
 
   if (outputString) {
@@ -102,6 +104,13 @@ exports.reduce = function(expression, outputString) {
     }
 
     return accumulator.reduce(function(prev, curr, i, array) {
+      if (prev === _nullObject) {
+        prev = null;
+      }
+      if (curr === _nullObject) {
+        curr = null;
+      }
+
       return eval(expression);
     });
   };
@@ -115,7 +124,11 @@ exports.reduce = function(expression, outputString) {
   collectStream.on('pipe', function(src) {
     src.on('end', function() {
       var result = performReduce();
-      if (outputString) result = String(result) + "\n";
+      if (outputString) {
+        result = String(result) + "\n";
+      } else if (result === null) {
+        result = _nullObject;
+      }
 
       collectStream.push(result);
       collectStream.end();
@@ -165,21 +178,62 @@ exports.ignore = function(ignoreEmpty) {
  * @return {Transform} A transform stream in object mode
  */
 exports.json = function(streamArray) {
-  var stringify, jsonStream;
+  var stringify, jsonStream, i;
 
   // Stream array members
   if (streamArray) {
-    stringify = new Stringify();
-    stringify.seperator = new Buffer(',\n', 'utf8');
-    return stringify;
+    return getStringifyStream();
   }
 
   // Stringify entire object
   jsonStream = new stream.Transform({objectMode: true});
   jsonStream._transform = function(chunk, encoding, fn) {
+    if (chunk === _nullObject) {
+      chunk = null;
+    }
+
     this.push(JSON.stringify(chunk));
     return fn();
   };
 
   return jsonStream;
 };
+
+/**
+ * Returns an instance of json-array-stream, or Stringify. The instance has
+ * been updated to handle the ",\n" as a delimiter, as well as convert
+ * _nullObject to null, prior to calling JSON.stringify.
+ *
+ * @returns {Stringify} The stringify stream
+ */
+function getStringifyStream() {
+  stringify = new Stringify();
+  stringify.seperator = new Buffer(',\n', 'utf8');
+
+  // Overwrite default _transform to convert _nullObject
+  stringify._transform = function(doc, enc, fn) {
+    if (this.destroyed) return;
+
+    if (this.started) {
+      this.push(this.seperator);
+    } else {
+      this.push(this.open);
+      this.started = true;
+    }
+
+    if (doc === _nullObject) {
+      doc = null;
+    }
+
+    try {
+      doc = JSON.stringify(doc, this.replacer, this.space);
+    } catch (err) {
+      return fn(err);
+    }
+
+    this.push(new Buffer(doc, 'utf8'));
+    fn();
+  };
+
+  return stringify;
+}

lib/utils.js

@@ -1,10 +1,20 @@
 /**
- * Exports array utility functions.
+ * Exports array utility functions, as well as a null object.
  */
 
 var utils = {};
 module.exports = utils;
 
+/**
+ * A null object representing a null value in a stream, so as not to terminate
+ * an object stream.
+ *
+ * @private
+ *
+ * @var {object}
+ */
+var _nullObject = utils._nullObject = {value: null};
+
 /**
  * Returns the length of the array.
  *
@@ -16,23 +26,53 @@ utils.length = function(array) {
 };
 
 /**
- * Returns the minimum value in an array.
+ * Returns the minimum value in an array. Mimics Math.min by casting values
+ * to a number, including the nullObject/null.
  *
  * @param   {*[]}    array The array for which to find its min
  * @returns {Number} Its minimum value
  */
 utils.min = function(array) {
-  return Math.min.apply(null, array);
+  var i, min, curr;
+  min = (array[0] === _nullObject) ? 0 : Number(array[0]);
+
+  for (i = 1; i < array.length; i++) {
+    if (array[i] === _nullObject) {
+      array[i] = 0;
+    }
+
+    curr = Number(array[i]);
+    if (curr < min) {
+      min = curr;
+    }
+  }
+
+  return min;
 };
 
 /**
- * Returns the maximum value in an array.
+ * Returns the maximum value in an array. Mimics Math.max by casting values
+ * to a number, including the nullObject/null.
  *
  * @param   {*[]}    array The array for which to find its max
  * @returns {Number} Its maximum value
  */
 utils.max = function(array) {
-  return Math.max.apply(null, array);
+  var i, max, curr;
+  max = (array[0] === _nullObject) ? 0 : Number(array[0]);
+
+  for (i = 1; i < array.length; i++) {
+    if (array[i] === _nullObject) {
+      array[i] = 0;
+    }
+
+    curr = Number(array[i]);
+    if (curr > max) {
+      max = curr;
+    }
+  }
+
+  return max;
 };
 
 /**
@@ -43,6 +83,13 @@ utils.max = function(array) {
  */
 utils.sum = function(array) {
   return array.reduce(function(curr, prev) {
+    if (curr === _nullObject) {
+      curr = null;
+    }
+    if (prev === _nullObject) {
+      prev = null;
+    }
+
     return Number(curr) + Number(prev);
   });
 };
@@ -66,6 +113,13 @@ utils.avg = function(array) {
  */
 utils.concat = function(array) {
   return array.reduce(function(curr, prev) {
+    if (curr === _nullObject) {
+      curr = null;
+    }
+    if (prev === _nullObject) {
+      prev = null;
+    }
+
     return String(curr) + String(prev);
   });
 };

spec/pjsSpec.js

@@ -108,6 +108,21 @@ describe('pjs', function() {
         done();
       });
     });
+
+    it('encodes null values as an object to avoid ending the stream', function(done) {
+      var map = pjs.map('null', false, true);
+
+      testStream(lines, map, {}, function(err, res) {
+        expect(err).to.be(null);
+        expect(res).to.eql([
+          {value: null},
+          {value: null},
+          {value: null},
+          {value: null}
+        ]);
+        done();
+      });
+    });
   });
 
   describe('reduce', function() {
@@ -190,6 +205,16 @@ describe('pjs', function() {
         done();
       });
     });
+
+    it('encodes null values as an object to avoid ending the stream', function(done) {
+      var reduce = pjs.reduce('null');
+
+      testStream([1, 2, 3], reduce, {end: false}, function(err, res) {
+        expect(err).to.be(null);
+        expect(res[0]).to.eql({value: null});
+        done();
+      });
+    });
   });
 
   describe('ignore', function() {

spec/utilsSpec.js

@@ -1,22 +1,36 @@
-var expect = require('expect.js');
-var utils  = require('../lib/utils.js');
+var expect      = require('expect.js');
+var utils       = require('../lib/utils.js');
+var _nullObject = utils._nullObject;
 
 describe('utils', function() {
   describe('length', function() {
     it('returns the length of an array', function() {
-      expect(utils.length([2, 4, 8])).to.be(3);
+      var result = utils.length([2, 4, 8]);
+      expect(result).to.be(3);
     });
   });
 
   describe('min', function() {
     it('returns the min value in an array', function() {
-      expect(utils.min([2, 4, 8])).to.be(2);
+      var result = utils.min([2, 4, 8]);
+      expect(result).to.be(2);
+    });
+
+    it('converts a null object to null', function() {
+      var result = utils.min([_nullObject, 1, 2]);
+      expect(result).to.be(0);
     });
   });
 
   describe('max', function() {
     it('returns the max value in an array', function() {
-      expect(utils.max([2, 4, 8])).to.be(8);
+      var result = utils.max([2, 4, 8]);
+      expect(result).to.be(8);
+    });
+
+    it('converts a null object to null', function() {
+      var result = utils.max([-2, _nullObject, -1]);
+      expect(result).to.be(0);
     });
   });
 
@@ -30,6 +44,11 @@ describe('utils', function() {
       var result = utils.sum([true, true, '2.5']);
       expect(result).to.be(4.5);
     });
+
+    it('converts a null object to null', function() {
+      var result = utils.sum([2, _nullObject, -1]);
+      expect(result).to.be(1);
+    });
   });
 
   describe('avg', function() {
@@ -42,6 +61,11 @@ describe('utils', function() {
       var result = utils.avg([true, true, '1.5', '2.5']);
       expect(result).to.be(1.5);
     });
+
+    it('converts a null object to null', function() {
+      var result = utils.avg([2, 4, _nullObject]);
+      expect(result).to.be(2);
+    });
   });
 
   describe('concat', function() {
@@ -54,6 +78,11 @@ describe('utils', function() {
       var result = utils.concat(['foo', true, 0]);
       expect(result).to.be('footrue0');
     });
+
+    it('converts a null object to null', function() {
+      var result = utils.concat(['foo', 'bar', _nullObject]);
+      expect(result).to.be('foobarnull');
+    });
   });
 
   describe('getBoundExpression', function() {