There are two ways to receive the data from a Readable stream: non-flowing and flowing. Let's analyze these modes in more detail.

readable.read([size])

process.stdin 
  .on('readable', () => { 
    let chunk; 
    console.log('New data available'); 
    while((chunk = process.stdin.read()) !== null) { 
      console.log(
        `Chunk read: (${chunk.length}) "${chunk.toString()}"` 
      ); 
    } 
  }) 
  .on('end', () => process.stdout.write('End of stream')); 

cat <path to a file> | node readStdin

process.stdin 
  .on('data', chunk => {  
    console.log('New data available');  
    console.log( 
      `Chunk read: (${chunk.length}) "${chunk.toString()}"`  
    );  
  })  
  .on('end', () => process.stdout.write('End of stream')); 

Now that we know how to read from a stream, the next step is to learn how to implement a new Readable stream. To do this, it's necessary to create a new class by inheriting the prototype of stream.Readable. The concrete stream must provide an implementation of the _read() method, which has the following signature:

readable._read(size) 

The internals of the Readable class will call the _read() method, which in turn will start to fill the internal buffer using push():

readable.push(chunk) 

To demonstrate how to implement new Readable streams, we can try to implement a stream that generates random strings. Let's create a new module called randomStream.js that will contain the code of our string generator:

const stream = require('stream'); 
const Chance = require('chance'); 
 
const chance = new Chance(); 
 
class RandomStream extends stream.Readable { 
  constructor(options) { 
    super(options); 
  } 
 
  _read(size) { 
    const chunk = chance.string();                        //[1] 
    console.log(`Pushing chunk of size: ${chunk.length}`); 
    this.push(chunk, 'utf8');                             //[2] 
    if(chance.bool({likelihood: 5})) {                    //[3] 
      this.push(null); 
    } 
  } 
} 
 
module.exports = RandomStream; 

At the top of the file, we will load our dependencies. There is nothing special here, except that we are loading a npm module called chance (https://npmjs.org/package/chance), which is a library for generating all sorts of random values, from numbers to strings to entire sentences.

The next step is to create a new class called RandomStream and that specifies stream.Readable as its parent. In the preceding code, we call the constructor of the parent class to initialize its internal state, and forward the options argument received as input. The possible parameters passed through the options object include the following:

Okay, now let's explain the _read() method:

We can also see that the size argument given in input to the _read() function is ignored, as it is an advisory parameter. We can simply just push all the available data, but if there are multiple pushes inside the same invocation, then we should check whether push() returns false, as this would mean that the internal buffer has reached the highWaterMark limit and we should stop adding more data to it.

That's it for RandomStream; we are now ready to use it. Let's create a new module named generateRandom.js in which we instantiate a new RandomStream object and pull some data from it:

const RandomStream = require('./randomStream'); 
const randomStream = new RandomStream(); 
 
randomStream.on('readable', () => { 
  let chunk; 
  while((chunk = randomStream.read()) !== null) { 
    console.log(`Chunk received: ${chunk.toString()}`); 
  } 
}); 

Now, everything is ready for us to try our new custom stream. Simply execute the generateRandom module as usual and watch a random set of strings flowing on the screen.

Pushing some data down a Writable stream is a straightforward business; all we need to do is to use the write() method, which has the following signature:

writable.write(chunk, [encoding], [callback]) 

The encoding argument is optional and can be specified if chunk is String (it defaults to utf8, and is ignored if chunk is Buffer); the callback function instead is called when the chunk is flushed into the underlying resource and is optional as well.

To signal that no more data will be written to the stream, we have to use the end() method:

writable.end([chunk], [encoding], [callback]) 

We can provide a final chunk of data through the end() method; in this case, the callback function is equivalent to registering a listener to the finish event, which is fired when all the data written in the stream has been flushed into the underlying resource.

Now, let's show how this works by creating a small HTTP server that outputs a random sequence of strings:

const Chance = require('chance'); 
const chance = new Chance(); 
 
require('http').createServer((req, res) => { 
  res.writeHead(200, {'Content-Type': 'text/plain'});        //[1] 
  while(chance.bool({likelihood: 95})) {                     //[2] 
    res.write(chance.string() + '
');                       //[3] 
  } 
  res.end('
The end...
');                                 //[4] 
  res.on('finish', () => console.log('All data was sent'));  //[5] 
}).listen(8080, () => console.log('Listening on http://localhost:8080')); 

The HTTP server that we created writes into the res object, which is an instance of http.ServerResponse and also a Writable stream. What happens is explained as follows:

We can call this small module entropyServer.js, and then execute it. To test the server, we can open a browser at the address http://localhost:8080, or use curl from the terminal as follows:

curl localhost:8080 

At this point, the server should start sending random strings to the HTTP client that you chose (please bear in mind that some browsers might buffer the data, and the streaming behavior might not be apparent).

Similar to a liquid flowing in a real piping system, Node.js streams can also suffer from bottlenecks, where data is written faster than the stream can consume it. The mechanism to cope with this problem consists of buffering the incoming data; however, if the stream doesn't give any feedback to the writer, we might incur a situation where more and more data is accumulated into the internal buffer, leading to undesired levels of memory usage.

To prevent this from happening, writable.write() will return false when the internal buffer exceeds the highWaterMark limit. Writable streams have a highWaterMark property, which is the limit of the internal buffer size beyond which the write() method starts returning false, indicating that the application should now stop writing. When the buffer is emptied, the drain event is emitted, communicating that it's safe to start writing again. This mechanism is called back-pressure.

We can quickly demonstrate how to take into account the back-pressure of a Writable stream by modifying the entropyServer that we created before:

const Chance = require('chance'); 
const chance = new Chance(); 
 
require('http').createServer((req, res) => { 
  res.writeHead(200, {'Content-Type': 'text/plain'}); 
 
  function generateMore() {                           //[1] 
    while(chance.bool({likelihood: 95})) { 
      let shouldContinue = res.write( 
      chance.string({length: (16 * 1024) - 1})        //[2] 
    ); 
      if(!shouldContinue) {                           //[3] 
        console.log('Backpressure'); 
        return res.once('drain', generateMore); 
      } 
    } 
    res.end('
The end...
',() => console.log('All data was sent')); 
  } 
  generateMore(); 
}).listen(8080, () => console.log('Listening on http://localhost:8080')); 

The most important steps of the previous code can be summarized as follows:

If we now try to run the server again, and then generate a client request with curl, there is a high probability that there will be some back-pressure, as the server produces data at a very high rate, faster than the underlying socket can handle.

We can implement a new Writable stream by inheriting the prototype of stream.Writable and providing an implementation for the _write() method. Let's try to do it immediately while discussing the details along the way.

Let's build a Writable stream that receives objects in the following format:

{ 
  path: <path to a file> 
  content: <string or buffer> 
} 

For each one of these objects, our stream has to save the content part into a file created at the given path. We can immediately see that the inputs of our stream are objects, and not strings or buffers; this means that our stream has to work in object mode.

Let's call the module toFileStream.js:

const stream = require('stream'); 
const fs = require('fs'); 
const path = require('path'); 
const mkdirp = require('mkdirp'); 
 
class ToFileStream extends stream.Writable { 
  constructor() { 
    super({objectMode: true}); 
  } 
 
  _write (chunk, encoding, callback) { 
    mkdirp(path.dirname(chunk.path), err => { 
      if (err) { 
        return callback(err); 
      } 
      fs.writeFile(chunk.path, chunk.content, callback); 
    }); 
  } 
} 
module.exports = ToFileStream; 

As the first step, let's load all the dependencies that we are going to use. Beware that we are requiring the module mkdirp and, as you should know from the previous chapters, it should be installed with NPM.

We created a new class for our new stream, which extends from stream.Writable.

We had to invoke the parent constructor to initialize its internal state; we also provide an options object that specifies that the stream works in object mode (objectMode: true). Other options accepted by stream.Writable are as follows:

Finally, we provided an implementation for the _write() method. As you can see, the method accepts a data chunk, an encoding (which makes sense only if we are in binary mode and the stream option decodeStrings is set to false). Also, the method accepts a callback function, which needs to be invoked when the operation completes; it's not necessary to pass the result of the operation but, if needed, we can still pass an error that will cause the stream to emit an error event.

Now, to try the stream that we just built, we can create a new module called, for example, writeToFile.js, and perform some write operations against the stream:

const ToFileStream = require('./toFileStream.js'); 
const tfs = new ToFileStream(); 
 
tfs.write({path: "file1.txt", content: "Hello"}); 
tfs.write({path: "file2.txt", content: "Node.js"}); 
tfs.write({path: "file3.txt", content: "Streams"}); 
tfs.end(() => console.log("All files created")); 

With this, we created and used our first custom Writable stream. Run the new module as usual to check its output; you will see that after the execution, three new files will be created.

Let's implement a Transform stream that replaces all the occurrences of a given string. To do this, we have to create a new module named replaceStream.js. Let's jump directly to the implementation:

const stream = require('stream'); 
const util = require('util'); 
 
class ReplaceStream extends stream.Transform { 
  constructor(searchString, replaceString) { 
    super(); 
    this.searchString = searchString; 
    this.replaceString = replaceString; 
    this.tailPiece = ''; 
  } 
 
  _transform(chunk, encoding, callback) { 
    const pieces = (this.tailPiece + chunk)                   //[1] 
      .split(this.searchString); 
    const lastPiece = pieces[pieces.length - 1]; 
    const tailPieceLen = this.searchString.length - 1; 
 
    this.tailPiece = lastPiece.slice(-tailPieceLen);          //[2] 
    pieces[pieces.length - 1] = lastPiece.slice(0,-tailPieceLen); 
 
    this.push(pieces.join(this.replaceString));               //[3] 
    callback(); 
  } 
 
  _flush(callback) { 
    this.push(this.tailPiece); 
    callback(); 
  } 
} 
 
module.exports = ReplaceStream; 

As always, we will start building the module from its dependencies. This time we are not using third-party modules.

Then we created a new class extending from the stream.Transform base class. The constructor of the class accepts two arguments: searchString and replaceString. As you can imagine, they allow us to define the text to match and the string to use as a replacement. We also initialize an internal tailPiece variable that will be used by the _transform() method.

Now, let's analyze the _transform() method, which is the core of our new class. The _transform() method has practically the same signature as that of the _write() method of the Writable stream but, instead of writing data into an underlying resource, it pushes it into the internal buffer using this.push(), exactly as we would do in the _read() method of a Readable stream. This confirms how the two sides of a Transform stream are actually connected.

The _transform() method of ReplaceStream implements the core of our algorithm. To search and replace a string in a buffer is an easy task; however, it's a totally different story when the data is streaming, and possible matches might be distributed across multiple chunks. The procedure followed by the code can be explained as follows:

When the stream ends, we might still have a last tailPiece variable not pushed into the internal buffer. That's exactly what the _flush() method is for; it is invoked just before the stream is ended, and this is where we have one final chance to finalize the stream or push any remaining data before completely ending the stream.

The _flush() method only takes in a callback that we have to make sure to invoke when all the operations are complete, causing the stream to be terminated. With this, we have completed our ReplaceStream class.

Now, it's time to try the new stream. We can create another module called replaceStreamTest.js that writes some data and then reads the transformed result:

const ReplaceStream = require('./replaceStream'); 
 
const rs = new ReplaceStream('World', 'Node.js'); 
rs.on('data', chunk => console.log(chunk.toString())); 
 
rs.write('Hello W'); 
rs.write('orld!'); 
rs.end(); 

To make life a little bit harder for our stream, we spread the search term (which is World) across two different chunks; then, using flowing mode, we read from the same stream, logging each transformed chunk. Running the preceding program should produce the following output:

Hel
lo Node.js
!

The way we created custom streams so far does not exactly follow the Node way; in fact, inheriting from a base stream class violates the small surface area principle and requires some boilerplate code. This does not mean that the streams were badly designed; in fact, we should not forget that since they are a part of the Node.js core they must be as flexible as possible in order to enable userland modules to extend them for a broad range of purposes.

However, most of the time we don't need all the power and extensibility that prototypal inheritance can give, but usually what we want is just a quick and an expressive way to define new streams. The Node.js community, of course, created a solution also for this. A perfect example is through2 (https://npmjs.org/package/through2), a small library which simplifies the creation of Transform streams. With through2, we can create a new Transform stream by invoking a simple function:

const transform = through2([options], [_transform], [_flush]) 

In a similar way, from2 (https://npmjs.org/package/from2) allows us to easily and succinctly create Readable streams with code such as the following:

const readable = from2([options], _read) 

The advantages of using these little libraries will be immediately clear as soon as we start showing their usage in the rest of the chapter.