Chain Data Streams in Node.js

Piping is a mechanism that connects a readable stream to a writable stream, allowing data to flow directly between them without manual handling of the data chunks.

This is particularly useful for tasks like reading from a file and writing to another file, or streaming data over the network.

To pipe streams, you can use the pipe() method of the readable stream instance:

readable.pipe(writable);

Example

Let's consider this file named original.txt located in the current directory:

$ cat original.txt
Lorem ipsum dolor sit amet, consectetur adipiscing elit.
Aliquam at ullamcorper magna, non tempus neque.
Pellentesque habitant morbi tristique senectus et netus et malesuada fames ac turpis egestas.

Let's consider this script, that copies a file using the piping mechanism:

const fs = require('node:fs');

const source = fs.createReadStream('original.txt', {
  encoding: 'utf8'
});

const destination = fs.createWriteStream('copy.txt', {
  encoding: 'utf8'
});

console.log('Copying file...');

source.pipe(destination);

destination.on('finish', () => console.log('Done.'));

source.on('error', e => console.error(e.toString()));

destination.on('error', e => console.error(e.toString()));

When executed, it will:

1. Create a readable stream named source from the file named original.txt.
2. Create a writable stream named destination to the file named copy.txt.
3. Pipe the readable stream into the writable stream.
4. Output a message indicating that the writable stream has finished writing data.

Which will produce this output:

Copying file...
Done.

And will produce no difference when compared to the copy.txt file:

$ diff original.txt copy.txt
$

Handle backpressure

Writable streams maintain an internal buffer to temporarily hold data before it's written to the underlying resource, such as a file.

When working with large datasets, it may happen that too much data is written too quickly, causing the buffer to reach its high watermark and the write() method to return a false value, indicating that it cannot currently accept more data.

In this case, you can:

1. Pause the readable stream using the pause() method.
2. Wait for the writable stream to emit a 'drain' event, indicating that the stream's internal buffer has been cleared, and that the stream is ready to accept more data.
3. Resume the readable stream using the resume() method.

Which translates to this implementation:

readStream.on('data', chunk => {
  if (!writeStream.write(chunk)) {
    readStream.pause();

    writeStream.once('drain', () => {
      readStream.resume();
    });
  }
});

Note: You typically won't need to handle the 'drain' event manually if you use the pipe() method, as it internally manages backpressure.

Summary

Here's a summary of what you've learned in this lesson:

• The pipe() method of a readable or duplex stream instance is used to connect its output to the input of a writable or duplex stream.