58 lines
1.2 KiB
JavaScript
58 lines
1.2 KiB
JavaScript
|
var stream = require("stream");
|
||
|
|
|
||
|
|
var ForkStream = module.exports = function ForkStream(options) {
|
||
|
|
options = options || {};
|
||
|
|
|
||
|
|
options.objectMode = true;
|
||
|
|
|
||
|
|
stream.Writable.call(this, options);
|
||
|
|
|
||
|
|
if (options.classifier) {
|
||
|
|
this._classifier = options.classifier;
|
||
|
|
}
|
||
|
|
|
||
|
|
this.a = new stream.Readable(options);
|
||
|
|
this.b = new stream.Readable(options);
|
||
|
|
|
||
|
|
var self = this;
|
||
|
|
|
||
|
|
var resume = function resume() {
|
||
|
|
if (self.resume) {
|
||
|
|
var r = self.resume;
|
||
|
|
self.resume = null;
|
||
|
|
r.call(null);
|
||
|
|
}
|
||
|
|
};
|
||
|
|
|
||
|
|
this.a._read = resume;
|
||
|
|
this.b._read = resume;
|
||
|
|
|
||
|
|
this.on("finish", function() {
|
||
|
|
self.a.push(null);
|
||
|
|
self.b.push(null);
|
||
|
|
});
|
||
|
|
};
|
||
|
|
ForkStream.prototype = Object.create(stream.Writable.prototype, {constructor: {value: ForkStream}});
|
||
|
|
|
||
|
|
ForkStream.prototype._classifier = function(e, done) {
|
||
|
|
return done(null, !!e);
|
||
|
|
};
|
||
|
|
|
||
|
|
ForkStream.prototype._write = function _write(input, encoding, done) {
|
||
|
|
var self = this;
|
||
|
|
|
||
|
|
this._classifier.call(null, input, function(err, res) {
|
||
|
|
if (err) {
|
||
|
|
return done(err);
|
||
|
|
}
|
||
|
|
|
||
|
|
var out = res ? self.a : self.b;
|
||
|
|
|
||
|
|
if (out.push(input)) {
|
||
|
|
return done();
|
||
|
|
} else {
|
||
|
|
self.resume = done;
|
||
|
|
}
|
||
|
|
});
|
||
|
|
};
|