Let's demonstrate this immediately with an example; let's create a module called parallelStream.js and define a generic Transform stream that executes a given transform function in parallel:

const stream = require('stream'); 
 
class ParallelStream extends stream.Transform { 
  constructor(userTransform) { 
    super({objectMode: true}); 
    this.userTransform = userTransform; 
    this.running = 0; 
    this.terminateCallback = null; 
  } 
 
  _transform(chunk, enc, done) { 
    this.running++; 
    this.userTransform(chunk, enc, this.push.bind(this), 
    this._onComplete.bind(this)); 
    done(); 
  } 
 
  _flush(done) { 
    if(this.running> 0) { 
      this.terminateCallback = done; 
    } else { 
      done(); 
    } 
  } 
 
  _onComplete(err) { 
    this.running--; 
    if(err) { 
      return this.emit('error', err); 
    } 
    if(this.running === 0) { 
      this.terminateCallback && this.terminateCallback(); 
    } 
  } 
 
} 
 
module.exports = ParallelStream; 

Let's analyze this new class. As you can see, the constructor accepts a userTransform() function, which is then saved as an instance variable; we also invoke the parent constructor and for convenience we enable the object mode by default.

Next, it is the turn of the _transform() method. In this method, we execute the userTransform() function, then we increment the count of running tasks; finally, we notify that the current transformation step is complete by invoking done(). The trick for triggering the processing of another item in parallel is exactly this; we are not waiting for the userTransform() function to complete before invoking done(); instead, we do it immediately. On the other hand, we provide a special callback to userTransform(), which is the this._onComplete() method; this allows us to get notified when the userTransform() completes.

The _flush() method is invoked just before the stream terminates, so if there are still tasks running we can put on hold the release of the finish event by not invoking the done() callback immediately; instead, we assign it to the this.terminateCallback variable. To understand how the stream is then properly terminated, we have to look into the _onComplete() method. This last method is invoked every time an asynchronous task completes. It checks whether there are any more tasks running and, if there are none, it invokes the this.terminateCallback() function, which will cause the stream to end, releasing the finish event that was put on hold in the _flush() method.

The ParallelStream class we just built allows us to easily create a Transform stream that executes its tasks in parallel, but there is a caveat: it does not preserve the order of the items as they are received. In fact, asynchronous operations can complete and push data at any time, regardless of when they are started. We immediately understand that this property does not play well with binary streams where the order of data usually matters, but it can surely be useful with some types of object streams.

Now, let's apply our ParallelStream to a concrete example. Let's imagine that we wanted to build a simple service to monitor the status of a big list of URLs. Let's imagine all these URLs are contained in a single file and are newline separated.

Streams can offer a very efficient and elegant solution to this problem, especially if we use our ParallelStream class to parallelize the checking of the URLs.

Let's build this simple application immediately in a new module called checkUrls.js:

const fs = require('fs'); 
const split = require('split'); 
const request = require('request'); 
const ParallelStream = require('./parallelStream'); 
 
fs.createReadStream(process.argv[2])                       //[1] 
  .pipe(split())                                           //[2] 
  .pipe(new ParallelStream((url, enc, push, done) => {     //[3] 
    if(!url) return done(); 
    request.head(url, (err, response) => { 
      push(url + ' is ' + (err ? 'down' : 'up') + '
'); 
      done(); 
    }); 
  })) 
  .pipe(fs.createWriteStream('results.txt'))   //[4] 
  .on('finish', () => console.log('All urls were checked')); 

As we can see, with streams our code looks very elegant and straightforward; let's see how it works:

Now, we can run the checkUrls module with a command such as this:

node checkUrlsurlList.txt

Here the file urlList.txt contains a list of URLs, for example:

When the command finishes running, we will see that a file, results.txt, was created. This contains the results of the operation, for example:

    http://thiswillbedownforsure.com is down 
    http://loige.co is up 
    http://www.mariocasciaro.me is up 

There is a good probability that the order in which the results are written is different from the order in which the URLs were specified in the input file. This is clear evidence that our stream executes its tasks in parallel, and it does not enforce any order between the various data chunks in the stream.

The parallel streams that we created previously might shuffle the order of the emitted data, but there are situations where this is not acceptable; sometimes, in fact, it is necessary to have each chunk emitted in the same order in which it was received. However, not all hopes is lost, we can still run the transform function in parallel; all we have to do is to sort the data emitted by each task so that it follows the same order in which the data was received.

This technique involves the use of a buffer to reorder the chunks while they are emitted by each running task. For brevity, we are not going to provide an implementation of such a stream, as it's quite verbose for the scope of this book; what we are going to do instead is reuse one of the available packages on NPM built for this specific purpose, for example, through2-parallel (https://npmjs.org/package/through2-parallel).

We can quickly check the behavior of an ordered parallel execution by modifying our existing checkUrls module. Let's say that we want our results to be written in the same order as the URLs in the input file, while executing our checks in parallel. We can do this using through2-parallel:

//... 
const throughParallel = require('through2-parallel'); 
 
fs.createReadStream(process.argv[2]) 
  .pipe(split()) 
  .pipe(throughParallel.obj({concurrency: 2},(url, enc, done) => { 
      //... 
    }) 
  ) 
  .pipe(fs.createWriteStream('results.txt')) 
  .on('finish', () => console.log('All urls were checked')); 

As we can see, the interface of through2-parallel is very similar to that of through2; the only difference is that we can also specify a concurrency limit for the transform function that we provide. If we try to run this new version of checkUrls, we will now see that the results.txt file lists the results in the same order as the URLs appear in the input file.

With this, we conclude our analysis of the asynchronous control flow with streams; next, we are going to focus on some piping patterns.