Node.js Streams

Node.js - Streams

What are Streams?

Streams are one of the fundamental concepts in Node.js, providing an efficient way to handle reading or writing data sequentially. Think of them like conveyor belts for data: instead of waiting for the entire dataset to be available in memory (like with fs.readFile), streams allow you to process data in smaller chunks as it arrives or is ready to be sent.

This makes streams particularly useful for working with large amounts of data (like large files or network responses) because they minimize memory usage and improve performance.

Types of Streams in Node.js

There are four main types of streams in Node.js, all of which are instances of EventEmitter:

• Readable: Streams from which data can be read (e.g., fs.createReadStream, HTTP request).
• Writable: Streams to which data can be written (e.g., fs.createWriteStream, HTTP response).
• Duplex: Streams that are both Readable and Writable (e.g., a TCP socket, zlib streams).
• Transform: A type of Duplex stream where the output is computed based on the input (e.g., a stream that compresses or encrypts data as it passes through).

Stream Events

Since streams are EventEmitters, they emit various events during their lifecycle. Some common events include:

• 'data': Fired when a chunk of data is available to be read from a Readable stream.
• 'end': Fired when there is no more data to be read from a Readable stream.
• 'error': Fired if an error occurs during reading or writing.
• 'finish': Fired on a Writable stream when all data has been flushed to the underlying system (e.g., written to the file).
• 'close': Fired when the stream and its underlying resource (like a file descriptor) have been closed.

Reading from a Stream

Let’s read the content of a file using a Readable stream. First, create a file input.txt:

This is line one.
This is line two.
Streaming data with Node.js is efficient!

Now, create readStream.js:

const fs = require('fs');
let fileContent = '';

// Create a readable stream
// Specify encoding directly in options for automatic string conversion
const readerStream = fs.createReadStream('input.txt', { encoding: 'utf8' });

// Handle stream events:

// 'data' event: Fired when a chunk of data arrives
readerStream.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
  fileContent += chunk;
});

// 'end' event: Fired when there's no more data
readerStream.on('end', () => {
  console.log('\n--- End of Stream ---');
  console.log('Complete File Content:\n' + fileContent);
});

// 'error' event: Fired on error
readerStream.on('error', (err) => {
  console.error('Error reading stream:', err.stack);
});

console.log('Stream reading initiated...');

Run the script:

$ node readStream.js

Output (chunk size might vary):

Stream reading initiated...
Received 73 bytes of data.

--- End of Stream ---
Complete File Content:
This is line one.
This is line two.
Streaming data with Node.js is efficient!

Note: Concatenating chunks with += can be inefficient for very large files. For large data, consider processing chunks individually or using specialized stream consumers.

Writing to a Stream

Let’s write data to a file using a Writable stream. Create writeStream.js:

const fs = require('fs');
const dataToWrite = 'Writing data chunk by chunk using Writable streams.\nHello Streams!';

// Create a writable stream to output.txt
const writerStream = fs.createWriteStream('output.txt', { encoding: 'utf8' });

// Write data to the stream
writerStream.write(dataToWrite);

// Mark the end of writing - essential to trigger 'finish'
writerStream.end(); // Can also pass the last chunk: writerStream.end('last chunk');

// Handle stream events:

// 'finish' event: Fired when all data is flushed
writerStream.on('finish', () => {
  console.log('Write operation completed.');
});

// 'error' event: Fired on error
writerStream.on('error', (err) => {
  console.error('Error writing stream:', err.stack);
});

console.log('Stream writing initiated...');

Run the script:

$ node writeStream.js

Output:

Stream writing initiated...
Write operation completed.

Check the output.txt file created in the same directory. It should contain the content defined in dataToWrite.

Piping Streams

Piping is a powerful mechanism that connects the output of a Readable stream directly to the input of a Writable stream. It automatically handles the flow of data, backpressure (preventing the Writable stream from being overwhelmed), and error propagation.

The .pipe() method is available on Readable streams. It takes a Writable stream as an argument.

Example: Copying a file using pipe (pipeDemo.js):

const fs = require('fs');

// Create readable stream from input.txt
const readerStream = fs.createReadStream('input.txt');

// Create writable stream to output_piped.txt
const writerStream = fs.createWriteStream('output_piped.txt');

// Pipe the streams!
readerStream.pipe(writerStream);

// Optional: Listen for finish/error on the writer stream
writerStream.on('finish', () => {
  console.log('Piping finished: input.txt copied to output_piped.txt');
});

writerStream.on('error', (err) => {
  console.error('Error during pipe:', err);
});

readerStream.on('error', (err) => {
  console.error('Error reading input for pipe:', err);
});

console.log('Piping initiated...');

Run it:

$ node pipeDemo.js

Output:

Piping initiated...
Piping finished: input.txt copied to output_piped.txt

This automatically copies the content of input.txt to output_piped.txt efficiently.

Chaining Streams (Transform Streams)

You can chain multiple .pipe() calls together, often involving Transform streams, to create data processing pipelines.

Example: Compressing a file using zlib (a Transform stream) (compress.js):

const fs = require('fs');
const zlib = require('zlib'); // Built-in module for compression
const { pipeline } = require('stream'); // Recommended way for piping

// Create streams
const reader = fs.createReadStream('input.txt');
const gzip = zlib.createGzip(); // Gzip compression transform stream
const writer = fs.createWriteStream('input.txt.gz');

console.log('Starting file compression...');

// Use stream.pipeline for robust error handling and cleanup
pipeline(reader, gzip, writer, (err) => {
  if (err) {
    console.error('Pipeline failed:', err);
  } else {
    console.log('File compressed successfully to input.txt.gz');
  }
});

Run it:

$ node compress.js

Output:

Starting file compression...
File compressed successfully to input.txt.gz

This creates a compressed file input.txt.gz.

Example: Decompressing the file (decompress.js):

const fs = require('fs');
const zlib = require('zlib');
const { pipeline } = require('stream');

// Create streams
const reader = fs.createReadStream('input.txt.gz');
const gunzip = zlib.createGunzip(); // Gunzip decompression transform stream
const writer = fs.createWriteStream('input_decompressed.txt');

console.log('Starting file decompression...');

pipeline(reader, gunzip, writer, (err) => {
  if (err) {
    console.error('Pipeline failed:', err);
  } else {
    console.log('File decompressed successfully to input_decompressed.txt');
  }
});

Run it:

$ node decompress.js

Output:

Starting file decompression...
File decompressed successfully to input_decompressed.txt

This decompresses input.txt.gz back into a text file.

Using stream.pipeline is generally preferred over multiple .pipe() calls as it provides better error handling and automatically destroys streams in the pipeline if one fails.

Further Reading

Streams are a deep topic. Explore the official Node.js documentation for more details:

• Node.js Stream API Docs: https://nodejs.org/api/stream.html