SimonJang · July 29, 2020 18:32 · Jul 5, 2013 · Jul 5, 2013 · Jul 5, 2013 · Jul 4, 2013
diff --git a/streamcombiner.js b/streamcombiner.js
@@ -22,7 +22,7 @@ var StreamCombiner = function() {
   // off the source stream piped into our internally managed streams.
   this.on('pipe', function(source) {
     source.unpipe(this);
-    for(i in this.streams) {
+    for(var i in this.streams) {
       source = source.pipe(this.streams[i]);
     }
     this.transformStream = source;

diff --git a/example.js b/example.js
@@ -1,5 +1,3 @@
-// Combine a pipe of two streams into one stream
-
 var util = require('util')
   , Transform = require('stream').Transform
   , StreamCombiner = require('./streamcombiner');

diff --git a/streams-combine.js → example.js b/streams-combine.js → example.js
@@ -1,7 +1,8 @@
 // Combine a pipe of two streams into one stream
 
 var util = require('util')
-  , Transform = require('stream').Transform;
+  , Transform = require('stream').Transform
+  , StreamCombiner = require('./streamcombiner');
 
 var chunks1 = [];
 var stream1 = new Transform();
@@ -36,27 +37,25 @@ process.on('exit', function () {
 });
 process.stdout.on('error', process.exit);
 
-
+// Test with `stream1` and `stream2`
 // stdin.pipe(stream1).pipe(stream2).pipe(stdout);
 
-// $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node streams-combine.js
+// $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node example.js
 // Outputs:
 // 1 abc
 // 2 def
 // 3 ghi
 // chunks1: ["abc\nd","ef\nghi\n"]
 // chunks2: ["abc","def","ghi"]
 
-// Best working solution I could find
-var stream3 = function(src) {
-  return src.pipe(stream1).pipe(stream2);
-};
-stream3(stdin).pipe(stdout);
+// Now combine into `stream3` to "hide" `stream1` and `stream2` from user
+var stream3 = new StreamCombiner(stream1, stream2);
+stdin.pipe(stream3).pipe(stdout);
 
-// $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node streams-combine.js
+// $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node example.js
 // Outputs:
 // 1 abc
 // 2 def
 // 3 ghi
 // chunks1: ["abc\nd","ef\nghi\n"]
-// chunks2: ["abc","def","ghi"]
+// chunks2: ["abc","def","ghi"]
diff --git a/streamcombiner.js b/streamcombiner.js
@@ -0,0 +1,40 @@
+/* StreamCombiner
+Combine a pipe of multiple streams into one stream.
+
+Example:
+
+  var stream3 = new StreamCombiner(stream1, stream2);
+  process.stdin.pipe(stream3).pipe(process.stdout);
+  // The line above will do this:
+  // process.stdin.pipe(stream1).pipe(stream2).pipe(process.stdout);
+
+Thanks to Brandon Tilley (https://github.com/BinaryMuse)
+for this code snippet.
+*/
+
+var util = require('util')
+  , PassThrough = require('stream').PassThrough;
+
+var StreamCombiner = function() {
+  this.streams = Array.prototype.slice.apply(arguments);
+
+  // When a source stream is piped to us, undo that pipe, and save
+  // off the source stream piped into our internally managed streams.
+  this.on('pipe', function(source) {
+    source.unpipe(this);
+    for(i in this.streams) {
+      source = source.pipe(this.streams[i]);
+    }
+    this.transformStream = source;
+  });
+};
+
+util.inherits(StreamCombiner, PassThrough);
+
+// When we're piped to another stream, instead pipe our internal
+// transform stream to that destination.
+StreamCombiner.prototype.pipe = function(dest, options) {
+  return this.transformStream.pipe(dest, options);
+};
+
+module.exports = StreamCombiner;
diff --git a/streams-combine.js b/streams-combine.js
@@ -47,15 +47,16 @@ process.stdout.on('error', process.exit);
 // chunks1: ["abc\nd","ef\nghi\n"]
 // chunks2: ["abc","def","ghi"]
 
-var stream3 = stream1.pipe(stream2);
-stdin.pipe(stream3).pipe(stdout);
+// Best working solution I could find
+var stream3 = function(src) {
+  return src.pipe(stream1).pipe(stream2);
+};
+stream3(stdin).pipe(stdout);
 
 // $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node streams-combine.js
 // Outputs:
 // 1 abc
-// d
-// 2 ef
-// ghi
-
-// chunks1: []
-// chunks2: ["abc\nd","ef\nghi\n"]
+// 2 def
+// 3 ghi
+// chunks1: ["abc\nd","ef\nghi\n"]
+// chunks2: ["abc","def","ghi"]
diff --git a/streams-combine.js b/streams-combine.js
@@ -39,6 +39,7 @@ process.stdout.on('error', process.exit);
 
 // stdin.pipe(stream1).pipe(stream2).pipe(stdout);
 
+// $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node streams-combine.js
 // Outputs:
 // 1 abc
 // 2 def
@@ -49,6 +50,7 @@ process.stdout.on('error', process.exit);
 var stream3 = stream1.pipe(stream2);
 stdin.pipe(stream3).pipe(stdout);
 
+// $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node streams-combine.js
 // Outputs:
 // 1 abc
 // d

diff --git a/streams-combine.js b/streams-combine.js
@@ -0,0 +1,59 @@
+// Combine a pipe of two streams into one stream
+
+var util = require('util')
+  , Transform = require('stream').Transform;
+
+var chunks1 = [];
+var stream1 = new Transform();
+var soFar = '';
+stream1._transform = function(chunk, encoding, done) {
+  chunks1.push(chunk.toString());
+  var pieces = (soFar + chunk).split('\n');
+  soFar = pieces.pop();
+  for (var i = 0; i < pieces.length; i++) {
+    var piece = pieces[i];
+    this.push(piece);
+  }
+  return done();
+};
+
+var chunks2 = [];
+var count = 0;
+var stream2 = new Transform();
+stream2._transform = function(chunk, encoding, done) {
+  chunks2.push(chunk.toString());
+  count = count + 1;
+  this.push(count + ' ' + chunk.toString() + '\n');
+  done();
+};
+
+var stdin = process.stdin;
+var stdout = process.stdout;
+
+process.on('exit', function () {
+    console.error('chunks1: ' + JSON.stringify(chunks1));
+    console.error('chunks2: ' + JSON.stringify(chunks2));
+});
+process.stdout.on('error', process.exit);
+
+
+// stdin.pipe(stream1).pipe(stream2).pipe(stdout);
+
+// Outputs:
+// 1 abc
+// 2 def
+// 3 ghi
+// chunks1: ["abc\nd","ef\nghi\n"]
+// chunks2: ["abc","def","ghi"]
+
+var stream3 = stream1.pipe(stream2);
+stdin.pipe(stream3).pipe(stdout);
+
+// Outputs:
+// 1 abc
+// d
+// 2 ef
+// ghi
+
+// chunks1: []
+// chunks2: ["abc\nd","ef\nghi\n"]