Merge pull request #1 from tilfin/feature/work-promises

Support workPromise returns Promise array

Author: tilfin

README.md

@@ -12,7 +12,7 @@ Node.js transform stream working at constant pace and concurrent for object mode
 * Specify work time at once (opts.workMS)
 * Specify concurrent workers (opts.concurrency)
 * Fire `done` event after when all workers have finished asynchrous -processes
-* Counting tag system to call `this.countTag(_tag_)` in `_workPromise`, you can get summarized results `tagCounts` grouped by tag.
+* Counting tag system to call `this.countTag(<tag>)` in `_workPromise`, you can get summarized results `tagCounts` grouped by tag.
 * Node.js 4.3 or later
 
 ## Targets
@@ -28,6 +28,17 @@ $ npm install -save paced-work-stream
 
 ## How to Use
 
+### Creating a PacedWorkStream
+
+```
+new PacedWorkStream(opts, workPromise);
+```
+
+* `opts.concurrency` is the number of concurrent processes.
+* `opts.workMS` is milliseconds of work time at once that contains process-time and wait-time.
+* `opts.highWaterMark` is maximum object buffer size. If you use flow mode, you should set it the number of source items.
+* `workPromise` is `function(item)` must returns a promise processing the item.
+
 ```javascript
 const es = require('event-stream');
 

lib/main.js

@@ -9,7 +9,7 @@ const util = require('util');
  *
  * @param {Number} opts.concurrency - number of concurrent process, default is 10
  * @param {number} opts.workMS - work-time millseconds at once, default is 0 (no wait)
- * @param {Function} workPromise - work-function returns a Promise callwed with an item `function (item)`
+ * @param {Promise|[Promise]} workPromise - work-function returns a Promise callwed with an item `function (item)`
  */
 function PacedWorkStream(opts, workPromise) {
   if (!(this instanceof PacedWorkStream))
@@ -23,6 +23,7 @@ function PacedWorkStream(opts, workPromise) {
 
   this.tagCounts = {};
   this._bufItems = [];
+  this._bufPrmss = [];
   if (workPromise) {
     this._workPromise = workPromise;
   }
@@ -35,37 +36,64 @@ util.inherits(PacedWorkStream, Transform);
 PacedWorkStream.prototype._transform = function(data, encoding, cb) {
   this._bufItems.push(data);
 
+  if (this._bufPrmss.length >= this._concurrency) {
+    const doPromises = this._bufPrmss.splice(0, this._concurrency);
+    this._process(doPromises, cb);
+    return;
+  }
+
   if (this._bufItems.length < this._concurrency) {
     cb();
     return;
   }
 
   const items = this._bufItems.splice(0, this._concurrency);
-  this._process(items, cb);
+  try {
+    this._bufPrmss = this._bufPrmss.concat(this._toPromises(items));
+    const doPromises = this._bufPrmss.splice(0, this._concurrency);
+    this._process(doPromises, cb);
+  } catch (err) {
+    cb(err);
+  }
 }
 PacedWorkStream.prototype._flush = function(cb) {
-  if (this._bufItems.length) {
-    this._process(this._bufItems, (err) => {
+  const restItems = this._bufItems;
+  if (restItems.length) {
+    try {
+      this._bufPrmss = this._bufPrmss.concat(this._toPromises(restItems));
+    } catch (err) {
       cb(err);
-      this.emit('done');
-    });
-    this._bufItems = [];
-  } else {
-    cb();
-    this.emit('done');
+      return;
+    }
   }
+
+  this._flushPromises(cb);
 }
 
-PacedWorkStream.prototype._process = function(items, cb) {
-  let promises;
-  try {
-    promises = items.map((item) => {
-        return this._workPromise(item);
-      });
-  } catch (err) {
-    return cb(err);
+PacedWorkStream.prototype._flushPromises = function(cb) {
+  if (this._bufPrmss.length === 0) {
+    cb();
+    this.emit('done');
+    return;
   }
 
+  const doPromises = this._bufPrmss.splice(0, this._concurrency);
+  this._process(doPromises, (err) => {
+    if (err) {
+      cb(err);
+    } else {
+      this._flushPromises(cb);
+    }
+  });
+}
+PacedWorkStream.prototype._toPromises = function(items) {
+  let promises = [];
+  items.forEach((item) => {
+    promises = promises.concat(this._workPromise(item));
+  });
+  return promises;
+}
+PacedWorkStream.prototype._process = function(promises, cb) {
   const startTime = new Date().getTime();
   Promise.all(promises)
   .then((results) => {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "paced-work-stream",
-  "version": "0.8.1",
+  "version": "0.9.0",
   "description": "Node.js stream working at constant pace and concurrent",
   "main": "lib/main.js",
   "scripts": {

test/work_promise.test.js

@@ -0,0 +1,70 @@
+'use strict';
+
+const chai = require('chai');
+const assert = chai.assert;
+const es = require('event-stream');
+const PacedWorkStream = require('../lib/main');
+
+describe('PacedWorkStream', () => {
+  context('workPromise returns promises greater than 1', () => {
+    it('raises done event', (done) => {
+      const pwStream = new PacedWorkStream({
+          concurrency: 2,
+          workMS: 0
+        }, function(item) {
+          if (item === 21) {
+            return [
+              Promise.resolve('21-1').then(() => { this.countTag('workDone') }),
+              Promise.resolve('21-2').then(() => { this.countTag('workDone') })
+            ];
+          } else {
+            return Promise.resolve(item.toString()).then(() => { this.countTag('workDone') });
+          }
+        })
+        .on('done', function() {
+          assert.deepEqual(this.tagCounts, { workDone: 6 });
+          done();
+        }).on('error', (err) => {
+          assert.ifError(err);
+          done();
+        });
+
+      const reader = es.readArray([11, 12, 21, 22, 31])
+      reader.pipe(pwStream);
+    });
+
+    it('raises done event', (done) => {
+      const pwStream = new PacedWorkStream({
+          concurrency: 3,
+          workMS: 0
+        }, function(item) {
+          if (item % 10 === 2) {
+            const prmss = [];
+            for (let i = 1; i <= 5; i++) {
+              prmss.push(Promise.resolve(`${item}-${i}`)
+                         .then((r) => { this.countTag('workDone'); return r; }));
+            }
+            return prmss;
+          } else {
+            return Promise.resolve(item.toString())
+                    .then((r) => { this.countTag('workDone'); return r; });
+          }
+        })
+        .on('done', function() {
+          assert.deepEqual(this.tagCounts, { workDone: 13 });
+          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, ['11', '12-1', '12-2', '12-3', '12-4', '12-5',
+            '21', '22-1', '22-2', '22-3', '22-4', '22-5', '31']);
+      });
+
+      reader.pipe(pwStream).pipe(writer);
+    });
+  });
+});