Node.js将相同的可读流管道化为多个(可写)目标

Question

我需要串行运行两个需要从同一个流中读取数据的命令.将流传输到另一个流后,缓冲区被清空,因此我无法再次从该流中读取数据,因此这不起作用:

var spawn = require('child_process').spawn;
var fs = require('fs');
var request = require('request');

var inputStream = request('http://placehold.it/640x360');
var identify = spawn('identify',['-']);

inputStream.pipe(identify.stdin);

var chunks = [];
identify.stdout.on('data',function(chunk) {
  chunks.push(chunk);
});

identify.stdout.on('end',function() {
  var size = getSize(Buffer.concat(chunks)); //width
  var convert = spawn('convert',['-','-scale',size * 0.5,'png:-']);
  inputStream.pipe(convert.stdin);
  convert.stdout.pipe(fs.createWriteStream('half.png'));
});

function getSize(buffer){
  return parseInt(buffer.toString().split(' ')[2].split('x')[0]);
}

请求抱怨这个

Error: You cannot pipe after data has been emitted from the response.

并改变inputStreamfs.createWriteStream当然会产生同样的问题.我不想写入文件,而是以某种方式重用请求生成的流(或其他任何内容).

有没有办法在完成管道后重复使用可读流？完成上述例子的最佳方法是什么？

Answer 1

您必须通过将流管道传输到两个流来创建流的副本.您可以使用PassThrough流创建一个简单的流,它只是将输入传递给输出.

const spawn = require('child_process').spawn;
const PassThrough = require('stream').PassThrough;

const a = spawn('echo', ['hi user']);
const b = new PassThrough();
const c = new PassThrough();

a.stdout.pipe(b);
a.stdout.pipe(c);

let count = 0;
b.on('data', function (chunk) {
  count += chunk.length;
});
b.on('end', function () {
  console.log(count);
  c.pipe(process.stdout);
});

输出:

8
hi user

Answer 2

第一个答案仅在流处理数据的时间大致相同时才有效.如果需要更长的时间,请求新数据的速度越快,因此覆盖较慢的数据仍在使用的数据(尝试使用重复流解决后我遇到此问题).

以下模式对我来说非常有效.它使用基于Stream2流,Streamz和Promises的库通过回调同步异步流.使用第一个答案中熟悉的示例:

spawn = require('child_process').spawn;
pass = require('stream').PassThrough;
streamz = require('streamz').PassThrough;
var Promise = require('bluebird');

a = spawn('echo', ['hi user']);
b = new pass;
c = new pass;   

a.stdout.pipe(streamz(combineStreamOperations)); 

function combineStreamOperations(data, next){
  Promise.join(b, c, function(b, c){ //perform n operations on the same data
  next(); //request more
}

count = 0;
b.on('data', function(chunk) { count += chunk.length; });
b.on('end', function() { console.log(count); c.pipe(process.stdout); });

Answer 3

您可以使用我创建的这个小 npm 包：

readable-stream-clone

这样，您可以根据需要多次重用可读流

Answer 4

对于一般问题，以下代码可以正常工作

var PassThrough = require('stream').PassThrough
a=PassThrough()
b1=PassThrough()
b2=PassThrough()
a.pipe(b1)
a.pipe(b2)
b1.on('data', function(data) {
  console.log('b1:', data.toString())
})
b2.on('data', function(data) {
  console.log('b2:', data.toString())
})
a.write('text')