101

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?

2
  • Seems like you are using imagemick. You can pass value like 50% to -scale for scaling. You can also use npmjs.org/package/gm
    – user568109
    Oct 24, 2013 at 6:45
  • 3
    @user568109 Yeah. That's not the issue here though. It's a more general question... it's imagemagick as it could be any other command/stream
    – Maroshii
    Oct 24, 2013 at 8:00

7 Answers 7

99

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
18
  • 5
    Used this technique with the Haraka mailserver attachment hooks to pipe the incoming stream into multiple mail account databases. This answer works.
    – user673046
    Dec 5, 2013 at 14:53
  • 24
    Note 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. Jul 18, 2014 at 13:27
  • 56
    I'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
    Sep 8, 2014 at 21:37
  • 13
    I 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.
    – B T
    Sep 8, 2014 at 21:47
  • 10
    Don'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. Aug 19, 2018 at 22:24
12

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); });
1
  • 1
    Which part is actually overwriting the data? The code which overwrites should naturally throw an error. Mar 29, 2020 at 10:23
5

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

3
  • 3
    does 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!
    – maganap
    May 8, 2021 at 11:34
  • This 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. Sep 29, 2021 at 9:47
  • There is also more smart alternative implementation: github.com/mcollina/cloneable-readable Sep 29, 2021 at 13:50
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')
1

I have a different solution to write to two streams simultaneously, naturally, the time to write will be the addition of the two times, but I use it to respond to a download request, where I want to keep a copy of the downloaded file on my server (actually I use a S3 backup, so I cache the most used files locally to avoid multiple file transfers)

/**
 * A utility class made to write to a file while answering a file download request
 */
class TwoOutputStreams {
  constructor(streamOne, streamTwo) {
    this.streamOne = streamOne
    this.streamTwo = streamTwo
  }

  setHeader(header, value) {
    if (this.streamOne.setHeader)
      this.streamOne.setHeader(header, value)
    if (this.streamTwo.setHeader)
      this.streamTwo.setHeader(header, value)
  }

  write(chunk) {
    this.streamOne.write(chunk)
    this.streamTwo.write(chunk)
  }

  end() {
    this.streamOne.end()
    this.streamTwo.end()
  }
}

You can then use this as a regular OutputStream

const twoStreamsOut = new TwoOutputStreams(fileOut, responseStream)

and pass it to to your method as if it was a response or a fileOutputStream

1
1

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.

  1. 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);
        });
    }
    
  2. 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);
    }
    
1
  • I feel like this defeats the purpose of using streams, since you're loading the stream into memory, rather than handling it as it comes
    – NotSimon
    Oct 28 at 16:30
-1

What about piping into two or more streams not at the same time ?

For example :

var PassThrough = require('stream').PassThrough;
var mybiraryStream = stream.start(); //never ending audio stream
var file1 = fs.createWriteStream('file1.wav',{encoding:'binary'})
var file2 = fs.createWriteStream('file2.wav',{encoding:'binary'})
var mypass = PassThrough
mybinaryStream.pipe(mypass)
mypass.pipe(file1)
setTimeout(function(){
   mypass.pipe(file2);
},2000)

The above code does not produce any errors but the file2 is empty

2
  • in some way it helps me!
    – sandip
    Apr 8, 2017 at 12:36
  • 8
    i think you've identified a problem, but it's confusing because this isn't an answer.
    – Michael
    Mar 31, 2018 at 21:00

Your Answer

Reminder: Answers generated by Artificial Intelligence tools are not allowed on Stack Overflow. Learn more

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.