add ParalleStream.js

Author: front-thinking

ParallelStream.js

@@ -0,0 +1,42 @@
+/*
+* 基于stream的异步并行处理解决方案
+* */
+
+"use strict";
+
+const stream = require('stream');
+
+class ParallelStream extends stream.Transform {
+    constructor(userTransform) {
+        super({objectMode: true});
+        this.userTransform = userTransform;
+        this.running = 0;
+        this.terminateCallback = null;
+    }
+
+    _transform(chunk, enc, done) {
+        this.running++;
+        this.userTransform(chunk, enc, this._onComplete.bind(this), this.push.bind(this));
+        done();
+    }
+
+    _flush(done) {
+        if(this.running > 0) {
+            this.terminateCallback = done;
+        } else {
+            done();
+        }
+    }
+
+    _onComplete(err) {
+        this.running--;
+        if(err) {
+            return this.emit('error', err);
+        }
+        if(this.running === 0) {
+            this.terminateCallback && this.terminateCallback();
+        }
+    }
+}
+
+module.exports = ParallelStream;