Merge pull request #3 from tilfin/feature/null-work

Fix when workPromise doesn't return promises

Author: tilfin

example/normal.js

@@ -1,13 +1,13 @@
 'use strict';
 
 const es = require('event-stream');
+const devnull = require('dev-null');
 
 const PacedWorkStream = require('../lib/main');
 
 const pwStream = new PacedWorkStream({
     concurrency: 2,
-    workMS: 1000,
-    highWaterMark: 5
+    workMS: 1000
   }, function(item) {
     console.log(new Date().toISOString(), 'Begin', item);
 
@@ -26,4 +26,4 @@ const pwStream = new PacedWorkStream({
   });
 
 const reader = es.readArray([11, 12, 21, 22, 31])
-reader.pipe(pwStream);
+reader.pipe(pwStream).pipe(devnull({ objectMode: true }));

example/with_promised_lifestream.js

@@ -7,8 +7,7 @@ const PacedWorkStream = require('../lib/main');
 
 const pacedWorker = new PacedWorkStream({
     concurrency: 2,
-    workMS: 1000,
-    highWaterMark: 5
+    workMS: 1000
   }, function(item) {
     console.log(new Date().toISOString(), 'Begin', item);
 

lib/main.js

@@ -89,7 +89,10 @@ PacedWorkStream.prototype._flushPromises = function(cb) {
 PacedWorkStream.prototype._toPromises = function(items) {
   let promises = [];
   items.forEach((item) => {
-    promises = promises.concat(this._workPromise(item));
+    const pr = this._workPromise(item);
+    if (pr) {
+      promises = promises.concat(pr);
+    }
   });
   return promises;
 }

package.json

@@ -28,6 +28,7 @@
   "devDependencies": {
     "chai": "^3.5.0",
     "coveralls": "^2.11.12",
+    "dev-null": "^0.1.1",
     "event-stream": "^3.3.4",
     "istanbul": "^0.4.5",
     "mocha": "^3.0.2",

test/work_promise.test.js

@@ -3,6 +3,8 @@
 const chai = require('chai');
 const assert = chai.assert;
 const es = require('event-stream');
+const devnull = require('dev-null');
+
 const PacedWorkStream = require('../lib/main');
 
 describe('PacedWorkStream', () => {
@@ -30,7 +32,7 @@ describe('PacedWorkStream', () => {
         });
 
       const reader = es.readArray([11, 12, 21, 22, 31])
-      reader.pipe(pwStream);
+      reader.pipe(pwStream).pipe(devnull({ objectMode: true }));
     });
 
     it('raises done event', (done) => {
@@ -67,4 +69,57 @@ describe('PacedWorkStream', () => {
       reader.pipe(pwStream).pipe(writer);
     });
   });
+
+  context('workPromise returns no promises sometimes', () => {
+    it('raises done event', (done) => {
+      const pwStream = new PacedWorkStream({
+          concurrency: 3,
+          workMS: 0
+        }, function(item) {
+          if (item % 10 === 2) {
+            return Promise.resolve(item.toString())
+                    .then((r) => { this.countTag('workDone'); return r; });
+          }
+        })
+        .on('done', function() {
+          assert.deepEqual(this.tagCounts, { workDone: 2 });
+          done();
+        }).on('error', (err) => {
+          assert.ifError(err);
+          done();
+        });
+
+      const reader = es.readArray([11, 12, 21, 22, 31])
+      const writer = es.writeArray(function(err, array) {
+        assert.deepEqual(array, ['12', '22']);
+      });
+
+      reader.pipe(pwStream).pipe(writer);
+    });
+  });
+
+  context('workPromise does not return promises at all', () => {
+    it('raises done event', (done) => {
+      const pwStream = new PacedWorkStream({
+          concurrency: 3,
+          workMS: 0
+        }, function(item) {
+          return;
+        })
+        .on('done', function() {
+          assert.deepEqual(this.tagCounts, {});
+          done();
+        }).on('error', (err) => {
+          assert.ifError(err);
+          done();
+        });
+
+      const reader = es.readArray([11, 12, 21, 22, 31])
+      const writer = es.writeArray(function(err, array) {
+        assert.deepEqual(array, []);
+      });
+
+      reader.pipe(pwStream).pipe(writer);
+    });
+  });
 });