www.aleph1.co.uk Git - yaffs-website/blob - node_modules/fork-stream/index.js

   1 var stream = require("stream");
   2
   3 var ForkStream = module.exports = function ForkStream(options) {
   4   options = options || {};
   5
   6   options.objectMode = true;
   7
   8   stream.Writable.call(this, options);
   9
  10   if (options.classifier) {
  11     this._classifier = options.classifier;
  12   }
  13
  14   this.a = new stream.Readable(options);
  15   this.b = new stream.Readable(options);
  16
  17   var self = this;
  18
  19   var resume = function resume() {
  20     if (self.resume) {
  21       var r = self.resume;
  22       self.resume = null;
  23       r.call(null);
  24     }
  25   };
  26
  27   this.a._read = resume;
  28   this.b._read = resume;
  29
  30   this.on("finish", function() {
  31     self.a.push(null);
  32     self.b.push(null);
  33   });
  34 };
  35 ForkStream.prototype = Object.create(stream.Writable.prototype, {constructor: {value: ForkStream}});
  36
  37 ForkStream.prototype._classifier = function(e, done) {
  38   return done(null, !!e);
  39 };
  40
  41 ForkStream.prototype._write = function _write(input, encoding, done) {
  42   var self = this;
  43
  44   this._classifier.call(null, input, function(err, res) {
  45     if (err) {
  46       return done(err);
  47     }
  48
  49     var out = res ? self.a : self.b;
  50
  51     if (out.push(input)) {
  52       return done();
  53     } else {
  54       self.resume = done;
  55     }
  56   });
  57 };