从两个pipe道stream创build一个Node.jsstream

如果可能的话，我想将两个Node.jsstream合并为一个。 我正在使用变换stream。

换句话说，我想让我的图书馆返回myStream供人们使用。 例如，他们可以写：

 process.stdin.pipe(myStream).pipe(process.stdout); 

而在内部，我正在使用第三方vendorStream ，它执行一些工作，将其插入myInternalStream包含的自己的逻辑。 那么上面的内容会转化为：

 process.stdin.pipe(vendorStream).pipe(myInternalStream).pipe(process.stdout); 

我可以做那样的事吗？ 我试过var myStream = vendorStream.pipe(myInternalStream)但是显然不起作用。

为了与bash做个比喻，假设我想编写一个程序来检查字母h是否存在于某个stream的最后一行（ tail -n 1 | grep h ），我可以创build一个shell脚本：

 # myscript.sh tail -n 1 | grep h 

然后如果人们这样做：

 $ printf "abc\ndef\nghi" | . myscript.sh 

它只是工作。

这是我迄今为止：

 // 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); // $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node streams-combine.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); // $ (printf "abc\nd"; sleep 1; printf "ef\nghi\n") | node streams-combine.js // Outputs: // 1 abc // 2 def // 3 ghi // chunks1: ["abc\nd","ef\nghi\n"] // chunks2: ["abc","def","ghi"] 

这是可能吗？ 让我知道，如果我想做的事情不清楚。

谢谢！

Answer 1

您可以观看某些内容到您的videostream，然后unpipe其unpipe其传输到您感兴趣的videostream中：

 var PassThrough = require('stream').PassThrough; var stream3 = new PassThrough(); // When a source stream is piped to us, undo that pipe, and save // off the source stream piped into our internally managed streams. stream3.on('pipe', function(source) { source.unpipe(this); this.transformStream = source.pipe(stream1).pipe(stream2); }); // When we're piped to another stream, instead pipe our internal // transform stream to that destination. stream3.pipe = function(destination, options) { return this.transformStream.pipe(destination, options); }; stdin.pipe(stream3).pipe(stdout); 

您可以将此function提取到您自己的可构造的stream类中：

 var util = require('util'); var PassThrough = require('stream').PassThrough; var StreamCombiner = function() { this.streams = Array.prototype.slice.apply(arguments); 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); StreamCombiner.prototype.pipe = function(dest, options) { return this.transformStream.pipe(dest, options); }; var stream3 = new StreamCombiner(stream1, stream2); stdin.pipe(stream3).pipe(stdout); 

Answer 2

一种select是使用multipipe ，它可以将多个变换链接在一起，作为单个变换stream进行封装：

 // my-stream.js var multipipe = require('multipipe'); module.exports = function createMyStream() { return multipipe(vendorStream, myInternalStream); }; 

那你可以这样做：

 var createMyStream = require('./my-stream'); var myStream = createMyStream(); process.stdin.pipe(myStream).pipe(process.stdout);