To illustrate a simple example, let's consider the case of the following two transform streams:

Using a library such as multipipe, we can easily build these streams by combining some of the streams that we already have available from the core libraries (file combinedStreams.js):

const zlib = require('zlib'); 
const crypto = require('crypto'); 
const combine = require('multipipe'); 
 
module.exports.compressAndEncrypt = password => { 
  return combine( 
    zlib.createGzip(), 
    crypto.createCipher('aes192', password) 
  ); 
}; 
 
module.exports.decryptAndDecompress = password => { 
  return combine( 
    crypto.createDecipher('aes192', password), 
    zlib.createGunzip() 
  ); 
}; 

We can now use these combined streams as if they were black boxes, for example, to create a small application that archives a file by compressing and encrypting it. Let's do that in a new module named archive.js:

const fs = require('fs'); 
const compressAndEncryptStream = 
  require('./combinedStreams').compressAndEncrypt; 
 
fs.createReadStream(process.argv[3]) 
  .pipe(compressAndEncryptStream(process.argv[2])) 
  .pipe(fs.createWriteStream(process.argv[3] + ".gz.enc")); 

We can further improve the preceding code by building a combined stream out of the pipeline that we created, this time not to obtain a reusable black box but only to take advantage of its aggregated error management. In fact, as we have already mentioned many times, writing something such as the following will only catch the errors that are emitted by the last stream:

fs.createReadStream(process.argv[3]) 
  .pipe(compressAndEncryptStream(process.argv[2])) 
  .pipe(fs.createWriteStream(process.argv[3] + ".gz.enc")) 
  .on('error', err => { 
    //Only errors from the last stream 
    console.log(err); 
  }); 

However, by combining all the streams together, we can fix the problem elegantly. Let's then rewrite the archive.js file as follows:

const combine = require('multipipe'); 
const fs = require('fs'); 
const compressAndEncryptStream = 
  require('./combinedStreams').compressAndEncrypt; 
 
combine( 
  fs.createReadStream(process.argv[3]) 
  .pipe(compressAndEncryptStream(process.argv[2]))
  .pipe(fs.createWriteStream(process.argv[3] + ".gz.enc")) 
).on('error', err => { 
  //this error may come from any stream in the pipeline 
  console.log(err); 
}); 

As we can see, we can now attach an error listener directly to the combined stream and it will receive any error event that is emitted by any of its internal streams.

Now, to run the archive module, simply specify a password and a file in the command-line argument:

node archive mypassword /path/to/a/file.txt

With this example, we have clearly demonstrated how important it is to combine streams; from one aspect, it allows us to create reusable compositions of streams, and from another, it simplifies the error management of a pipeline.

Let's create a small utility that outputs both the sha1 and md5 hashes of a given file. Let's call this new module generateHashes.js and let's start by initializing our checksum streams:

const fs = require('fs'); 
const crypto = require('crypto'); 
 
const sha1Stream = crypto.createHash('sha1'); 
sha1Stream.setEncoding('base64'); 
 
const md5Stream = crypto.createHash('md5'); 
md5Stream.setEncoding('base64'); 

Nothing special so far; the next part of the module is actually where we will create a Readable stream from a file and fork it to two different streams in order to obtain two other files, one containing the sha1 hash and the other containing the md5 checksum:

const inputFile = process.argv[2]; 
const inputStream = fs.createReadStream(inputFile); 
inputStream 
  .pipe(sha1Stream) 
  .pipe(fs.createWriteStream(inputFile + '.sha1')); 
 
inputStream 
  .pipe(md5Stream) 
  .pipe(fs.createWriteStream(inputFile + '.md5')); 

Very simple, right? The inputStream variable is piped into sha1Stream on one side and md5Stream on the other. There are a couple of things to note, though, that happen behind the scenes:

To make a simple example, let's implement a small program that creates a tarball from the contents of two different directories. For this purpose, we are going to introduce two new NPM packages:

Our new module is going to be called mergeTar.js; let's define its contents starting from some initialization steps:

const tar = require('tar'); 
const fstream = require('fstream'); 
const path = require('path'); 
 
const destination = path.resolve(process.argv[2]); 
const sourceA = path.resolve(process.argv[3]); 
const sourceB = path.resolve(process.argv[4]); 

In the preceding code, we are just loading all the dependencies and initializing the variables that contain the name of the destination file and the two source directories (sourceA and sourceB).

Next, we will create the tar stream and pipe it into its destination:

const pack = tar.Pack(); 
pack.pipe(fstream.Writer(destination)); 

Now it's time to initialize the source streams:

let endCount = 0; 
function onEnd() { 
  if(++endCount === 2) { 
    pack.end(); 
  } 
} 
 
const sourceStreamA = fstream.Reader({type: "Directory", path: sourceA}) 
  .on('end', onEnd); 
 
const sourceStreamB = fstream.Reader({type: "Directory", path: sourceB}) 
  .on('end', onEnd); 

In the preceding code, we created the streams that read from both the two source directories (sourceStreamA and sourceStreamB); then for each source stream we attach an end listener, which will terminate the pack stream only when both the directories are read completely.

Finally, it is time to perform the real merge:

sourceStreamA.pipe(pack, {end: false}); 
sourceStreamB.pipe(pack, {end: false}); 

We pipe both the sources into the pack stream and take care to disable the auto ending of the destination stream by providing the option {end: false} to the two pipe() invocations.

With this, we have completed our simple TAR utility. We can try this utility by providing the destination file as the first command-line argument, followed by the two source directories:

node mergeTar dest.tar /path/to/sourceA /path/to/sourceB

To conclude this section, it's worth mentioning that, on npm, we can find a few modules that can simplify the merging of streams, for example:

As for the last comment on the stream merge pattern, it's worth reminding that the data piped into the destination stream is randomly intermingled; this is a property that can be acceptable in some types of object streams (as we saw in the last example) but it is often an undesired effect when dealing with binary streams.

However, there is one variation of the pattern that allows us to merge streams in order; it consists of consuming the source streams one after the other, when the previous one ends, the next one starts emitting chunks (it is like concatenating the output of all the sources). As always, on NPM we can find some packages that also deal with this situation. One of them is multistream (https://npmjs.org/package/multistream).

Let's use an example to drive our discussion. We want to have a small program that starts a child process and redirects both its standard output and standard error to a remote server, which in turn saves the two streams into two separate files. So, in this case, the shared medium is a TCP connection, while the two channels to be multiplexed are the stdout and stderr of a child process. We will leverage a technique called packet switching, the same technique that is used by protocols such as IP, TCP, or UDP and that consists of wrapping the data into packets allowing us to specify various meta information, useful for multiplexing, routing, controlling the flow, checking for corrupted data, and so on. The protocol that we are going to implement for our example is very minimalist; in fact, we will simply wrap our data into packets with the following structure:

As shown in the preceding figure, the packet contains the actual data, but also a header (Channel ID + Data length), which will make it possible to differentiate the data of each stream and enable the demultiplexer to route the packet to the right channel.

const child_process = require('child_process'); 
const net = require('net'); 

function multiplexChannels(sources, destination) { 
  let totalChannels = sources.length; 
  for(let i = 0; i <sources.length; i++) { 
    sources[i] 
      .on('readable', function() {                           //[1] 
        let chunk; 
        while((chunk = this.read()) !== null) { 
          const outBuff = new Buffer(1 + 4 + chunk.length);  //[2] 
          outBuff.writeUInt8(i, 0);        
          outBuff.writeUInt32BE(chunk.length, 1); 
          chunk.copy(outBuff, 5); 
          console.log('Sending packet to channel: ' + i); 
          destination.write(outBuff);                        //[3] 
        } 
      } 
      .on('end', () => {                                     //[4] 
        if(--totalChannels === 0) { 
           destination.end(); 
        } 
      }); 
  } 
} 

const socket = net.connect(3000, () => {                    //[1] 
  const child = child_process.fork(                         //[2] 
  process.argv[2], 
  process.argv.slice(3), 
    {silent: true} 
  ); 
  multiplexChannels([child.stdout, child.stderr], socket);  //[3] 
}); 

const net = require('net'); 
const fs = require('fs'); 
 
function demultiplexChannel(source, destinations) { 
  let currentChannel = null; 
  let currentLength = null; 
 
  source 
    .on('readable', () => {                                      //[1] 
      let chunk; 
      if(currentChannel === null) {                              //[2] 
        chunk = source.read(1); 
        currentChannel = chunk && chunk.readUInt8(0); 
      } 
 
      if(currentLength === null) {                               //[3] 
        chunk = source.read(4); 
        currentLength = chunk && chunk.readUInt32BE(0); 
        if(currentLength === null) { 
          return; 
        } 
      } 
 
      chunk = source.read(currentLength);                         //[4] 
      if(chunk === null) { 
        return; 
      } 
      console.log('Received packet from: ' + currentChannel); 
      destinations[currentChannel].write(chunk);                  //[5] 
      currentChannel = null; 
      currentLength = null; 
    }) 
    .on('end', ()=> {                                             //[6] 
      destinations.forEach(destination => destination.end()); 
      console.log('Source channel closed'); 
    }); 
} 

net.createServer(socket => { 
  const stdoutStream = fs.createWriteStream('stdout.log'); 
  const stderrStream = fs.createWriteStream('stderr.log');  
  demultiplexChannel(socket, [stdoutStream, stderrStream]); 
}).listen(3000, () => console.log('Server started')); 

console.log("out1"); 
console.log("out2"); 
console.error("err1"); 
console.log("out3"); 
console.error("err2"); 

node server

node client generateData.js

The example that we have just shown demonstrated how to multiplex and demultiplex a binary/text stream, but it's worth mentioning that the same rules apply also to object streams. The greatest difference is that, using objects, we already have a way to transmit the data using atomic messages (the objects), so multiplexing would be as easy as setting a property channelID into each object, while demultiplexing would simply involve reading the channelID property and routing each object towards the right destination stream.

Another pattern involving only demultiplexing consists of routing the data coming from a source depending on some condition. With this pattern, we can implement complex flows, such as the one shown in the following diagram:

The demultiplexer used in the system described by the preceding diagram, takes a stream of objects representing animals and distributes each of them to the right destination stream based on the class of the animal: reptiles, amphibians, and mammals.

Using the same principle, we can also implement an if...else statement for streams; for some inspiration, take a look at the ternary-stream package (https://npmjs.org/package/ternary-stream) that allows us to do exactly that.