Simplify Node.js Streams with stream/promises

This article explains how the stream/promises module simplifies stream handling in Node.js. By replacing event-listener boilerplate with native Promise-based APIs, this module allows developers to write cleaner, more readable, and less error-prone asynchronous code using async/await.

The Problem with Traditional Streams

Historically, Node.js streams relied heavily on event emitters. Handling data flow, completion, and errors required registering multiple event listeners:

const fs = require('fs');

const readStream = fs.createReadStream('input.txt');
const writeStream = fs.createWriteStream('output.txt');

readStream.on('data', (chunk) => {
  // Handle chunk
});

readStream.on('end', () => {
  console.log('Finished reading');
});

readStream.on('error', (err) => {
  console.error('Stream error:', err);
});

This event-driven approach often leads to verbose code, especially when chaining multiple streams. Managing error propagation across piped streams manually is notoriously difficult and frequently leads to memory leaks or unhandled exceptions.

How stream/promises Solves This

The stream/promises module, introduced as a stable feature in Node.js, provides Promise-wrapped alternatives to core stream utilities. Instead of managing events, you can treat stream lifecycle events as standard JavaScript Promises.

The module primarily exports two essential utility functions: pipeline and finished.

1. Simplified Chaining with pipeline

The promisified version of pipeline chains multiple streams together, forwards errors automatically, and cleans up destination streams when the source stream fails.

By returning a Promise, it integrates seamlessly into async/await blocks:

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function compressFile() {
  try {
    await pipeline(
      fs.createReadStream('input.txt'),
      zlib.createGzip(),
      fs.createWriteStream('input.txt.gz')
    );
    console.log('Pipeline succeeded: File compressed.');
  } catch (err) {
    console.error('Pipeline failed:', err);
  }
}

compressFile();

With this approach, any error that occurs in the read stream, the gzip transform stream, or the write stream is caught in a single catch block. The system automatically destroys all active streams in the pipeline, preventing memory leaks.

2. Tracking Completion with finished

The finished function returns a Promise that resolves when a stream is no longer readable or writable, or has encountered an error. This is incredibly useful for executing code only after a stream has fully processed its data.

const { finished } = require('node:stream/promises');
const fs = require('node:fs');

async function processStream() {
  const stream = fs.createReadStream('large-file.json');
  
  stream.resume(); // Start flowing the data

  try {
    await finished(stream);
    console.log('Stream has finished processing successfully.');
  } catch (err) {
    console.error('Stream failed during processing:', err);
  }
}

Benefits of stream/promises