ParallelStream.js

/*
* 基于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;