Version 1

[yaffs-website] / node_modules / fork-stream / index.js

diff --git a/node_modules/fork-stream/index.js b/node_modules/fork-stream/index.js
new file mode 100644 (file)
index 0000000..f916b79
--- /dev/null
+++ b/node_modules/fork-stream/index.js
@@ -0,0 +1,57 @@
+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;
+    }
+  });
+};