Node.js Piping the same readable stream into multiple (writable) targets
Solution 1
You have to create duplicate of the stream by piping it to two streams. You can create a simple stream with a PassThrough stream, it simply passes the input to the output.
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);
});
Output:
8
hi user
Solution 2
The first answer only works if streams take roughly the same amount of time to process data. If one takes significantly longer, the faster one will request new data, consequently overwriting the data still being used by the slower one (I had this problem after trying to solve it using a duplicate stream).
The following pattern worked very well for me. It uses a library based on Stream2 streams, Streamz, and Promises to synchronize async streams via a callback. Using the familiar example from the first answer:
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); });
Solution 3
You can use this small npm package I created:
readable-stream-clone
With this you can reuse readable streams as many times as you need
Solution 4
For general problem, the following code works fine
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')
Solution 5
If you have async operations on the PassThrough streams, the answers posted here won't work. A solution that works for async operations includes buffering the stream content and then creating streams from the buffered result.
-
To buffer the result you can use concat-stream
const Promise = require('bluebird'); const concat = require('concat-stream'); const getBuffer = function(stream){ return new Promise(function(resolve, reject){ var gotBuffer = function(buffer){ resolve(buffer); } var concatStream = concat(gotBuffer); stream.on('error', reject); stream.pipe(concatStream); }); }
-
To create streams from the buffer you can use:
const { Readable } = require('stream'); const getBufferStream = function(buffer){ const stream = new Readable(); stream.push(buffer); stream.push(null); return Promise.resolve(stream); }
Maroshii
Updated on March 25, 2020Comments
-
Maroshii about 4 years
I need to run two commands in series that need to read data from the same stream. After piping a stream into another the buffer is emptied so i can't read data from that stream again so this doesn't work:
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]); }
Request complains about this
Error: You cannot pipe after data has been emitted from the response.
and changing the inputStream to
fs.createWriteStream
yields the same issue of course. I don't want to write into a file but reuse in some way the stream that request produces (or any other for that matter).Is there a way to reuse a readable stream once it finishes piping? What would be the best way to accomplish something like the above example?
-
Admin over 10 yearsUsed this technique with the Haraka mailserver attachment hooks to pipe the incoming stream into multiple mail account databases. This answer works.
-
Jerome WAGNER almost 10 yearsNote that this technique only works if the spawned command outputs a number of bytes that does not fill the backpressure buffers. you can try and make it fail with a = spawn('head', ['-c', '200K', '/dev/urandom']);. If c is not piped out, at some point, a.stdout will pause piping out. b will drain and never end.
-
Daniel Beardsley over 9 yearsIs
a.stdout.pipe(c);
supposed to beb.stdout.pipe(c);
? -
user568109 over 9 years@DanielBeardsley No, it is copying streams. That would never work as you want to listen on b's data events.
-
B T over 9 yearsI'm confused, you say that you can't process the same stream twice, but your solution is to.. process the same stream twice (with the PassThrough transform). This seems contradictory. Is this something special about the stdout streams?
-
B T over 9 yearsI tested this and it certainly works. I think its not correct for you to say "you cannot process same [the] stream twice", since that's what you're doing. Your first statements about not being able to pipe a stream after its 'end' is the appropriate reason.
-
user568109 over 9 years@BT If by stream you mean flow of data. In the answer's context stream refers to an object. Also the processing as asked in question concerns causal or sequential processing. Specifically OP wanted to calculate size of the stream and then use the stream from the beginning.
-
Tola almost 9 yearsTested this answer with images, and like @Jerome WAGNER said, it works if files are small enough. It did not work for me as request never ended and I got "socket hang up" errors.
-
inf3rno over 8 yearsI think this is no longer a valid statement: github.com/nodejs/readable-stream/blob/master/lib/… readable streams can pipe data to multiple writeable streams.
-
user568109 over 8 years@inf3rno The same was valid when I answered. The OP was using the stream incorrectly. He was piping after consuming the data.
-
clacke about 8 years@inf3rno You mean github.com/nodejs/readable-stream/blob/… . :-)
-
sandip about 7 yearsin some way it helps me!
-
fider about 6 yearsSolution seems fine but watch out when operating on buffers - inplace buffer data modifications in b will be visible in c (if data chunk is string then no worry :) ).
-
Michael about 6 yearsi think you've identified a problem, but it's confusing because this isn't an answer.
-
kiwicomb123 over 5 yearsDon't use this method because it creates issues if the streams are read at different rates. Try this instead npmjs.com/package/readable-stream-clone worked well for me.
-
Avius over 5 years@kiwicomb123 nice, this happily worked out in my situation. Sadly I am not quite able to understand what the problem is, and you are not the first person to mention it. Are you able to elaborate? How is the stream reading rate measured? Is it about the actual time in seconds it takes to process stream data? The first stream is already finished by the time the final pipe starts reading, which is what confuses me. I guess I don't know something fundamental about how streams work. I've read the NodeJS Stream doc before but it's of little help. Do you know any other more useful resource?
-
kiwicomb123 over 5 years@Avius I think its a race condition. Under the assumption that you can't process the same stream twice, if one fork of the stream is finished processing the stream before the second one even starts then the second one will not work. However, I don't have formal documentation to back this up, I arrived at this conclusion by testing.
-
Jürg Lehni about 5 yearsI can confirm that @kiwicomb123's suggestion also works in my situation. I've tried all kinds of things before (e.g. npmjs.com/package/cloneable-readable) to no avail. Here's my version of the class, updated to modern JS: github.com/ditojs/dito/blob/master/packages/server/src/storage/…
-
Robert Siemer about 4 yearsWhich part is actually overwriting the data? The code which overwrites should naturally throw an error.
-
ShortFuse almost 4 yearsFor anybody doubting this, this is the suggested method of piping to two streams by the actual NodeJS team since 2011. nodejs.org/en/knowledge/advanced/streams/how-to-use-stream-pipe They have a sample of streaming a process output to a file as well as
stdout
..pipe()
supports multiple entries. github.com/nodejs/node/blob/… Also,.pipe()
has backpressuring built-in for convenience. nodejs.org/en/docs/guides/backpressuring-in-streams If in doubt, never call.write()
and only use.pipe()
. -
Rich Remer over 3 yearsPiping to both a file and stdout may well be a special case since stdout is handled a bit different than other streams since it can't be closed.
-
maganap almost 3 yearsdoes it suffer from the backpressure problem described above? What about producing an empty file from the second pipe? If you can elaborate a little that'd be awesome (to me and to your package reputation :-) ). Thanks in advance!
-
SleepWalker over 2 yearsThis lib does correct thing. It is so simple, that the entire source code can be copied here as an answer. This lib won't suffer from "backpressure problem" (see @maganap comment above). This lib will completely ignore backpressure mechanism.
-
SleepWalker over 2 yearsThere is also more smart alternative implementation: github.com/mcollina/cloneable-readable
-
maganap over 2 years@SleepWalker Thanks for the reference