Streams in Depth

Why streams need a deeper look#

The Async Node.js page introduced streams briefly: process data in chunks instead of loading a whole file into memory. That's the right mental model to start with, but building a real pipeline — reading a file, transforming its contents, writing the result, all without blowing up memory usage on a multi-gigabyte input — needs three more pieces: the four stream types, backpressure, and stream.pipeline for wiring them together safely.

The four stream types#

Every stream in Node is one of four base classes, all extending EventEmitter:

A Transform stream is the one you'll write yourself most often — it's a Duplex stream where whatever comes in on the writable side gets processed and pushed out the readable side, which is exactly the shape of "parse this," "compress this," or "filter these lines."

import { Transform } from 'stream';

class UpperCaseTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback(); // MUST be called — it's how this stream signals "ready for the next chunk"
  }
}

Backpressure: why pipe/pipeline exist at all#

Backpressure is what happens when a writable destination can't keep up with a readable source — writing to a slow disk, a slow network socket, or a downstream service, while data keeps arriving faster than it can be flushed. Handle it wrong and memory usage climbs unbounded as unconsumed chunks pile up waiting to be written.

writable.write(chunk) returns false the moment its internal buffer is full, which is Node's own signal to stop pushing more data until a 'drain' event says it's safe again:

// The manual, error-prone way — shown to make backpressure's mechanics explicit,
// not as the recommended way to actually write this code
function writeAll(readable, writable) {
  readable.on('data', (chunk) => {
    const canContinue = writable.write(chunk);
    if (!canContinue) {
      readable.pause();               // stop reading until the writable catches up
      writable.once('drain', () => readable.resume());
    }
  });
}

Getting this exactly right by hand — pausing, resuming, handling errors on both sides, closing both streams at the right moment — is fiddly enough that Node ships a built-in solution: .pipe() handles backpressure automatically, and stream.pipeline() (its modern successor) does the same while also handling errors and cleanup correctly across the whole chain.

import { pipeline } from 'stream/promises';

await pipeline(readable, writable); // backpressure handled for you, on every chunk

A complete file-processing pipeline#

Reading a large log file, keeping only lines that contain "ERROR", and writing the filtered result out compressed — three streams chained together, each handling exactly its own piece, with backpressure managed automatically end to end:

import { createReadStream, createWriteStream } from 'fs';
import { createGzip } from 'zlib';
import { Transform } from 'stream';
import { pipeline } from 'stream/promises';

// A Transform stream that filters lines, working on raw chunks rather than
// full lines for simplicity — good enough as long as a line never spans a chunk boundary
class FilterErrorLines extends Transform {
  _transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    const errorLines = lines.filter(line => line.includes('ERROR'));

    if (errorLines.length > 0) {
      this.push(errorLines.join('\n') + '\n');
    }
    callback();
  }
}

async function processLogFile(inputPath, outputPath) {
  await pipeline(
    createReadStream(inputPath),      // Readable — reads the file in chunks
    new FilterErrorLines(),           // Transform — keeps only error lines
    createGzip(),                     // Transform — compresses the result
    createWriteStream(outputPath),    // Writable — writes the compressed output
  );
  console.log('Done — output written to', outputPath);
}

processLogFile('app.log', 'errors.log.gz').catch((err) => {
  console.error('Pipeline failed:', err);
});

Regardless of whether app.log is 10KB or 10GB, this uses roughly the same, small amount of memory — at any instant, only the chunk currently being read, filtered, compressed, and written exists in memory, because pipeline() applies backpressure across the entire chain: if the disk write to errors.log.gz is slow, gzip stops accepting new input, which stops the filter from pushing more, which stops the file read from producing more, automatically, with no manual pause()/resume() anywhere in this code.

Common mistakes#

• Attaching a 'data' listener directly to a Readable stream to "manually" pipe it into a Writable, instead of using .pipe() or pipeline() — it's easy to get backpressure handling subtly wrong this way, and memory usage climbs under load exactly when it matters most.
• Forgetting to call the callback() argument inside a custom _transform() — the stream simply stalls forever, since that call is the stream's only signal that it's ready for the next chunk.
• Using .pipe() instead of pipeline() in new code and not attaching an 'error' listener to every stream in the chain — .pipe() does not forward errors between streams automatically, so an error in one stream can leave the others open and the process silently stuck; pipeline() handles this correctly by default.
• Splitting on '\n' inside a single chunk (as the example above does, for simplicity) without accounting for a line that happens to be split across two chunk boundaries — a production-grade line-splitting transform typically buffers a partial trailing line until the next chunk completes it.

Interview questions#

Q: What is backpressure, and how does pipeline() handle it automatically? Backpressure is what happens when a writable destination can't consume data as fast as a readable source produces it — without handling it, unconsumed chunks pile up in memory. writable.write() returns false when its internal buffer is full, and pipeline() (like .pipe()) uses that signal to automatically pause the upstream source until a 'drain' event says it's safe to resume, so memory usage stays roughly constant regardless of a mismatch in speed between the two ends.

Q: What's the difference between a Duplex stream and a Transform stream? A Duplex stream has independent readable and writable sides that aren't necessarily related to each other (a TCP socket, for instance). A Transform stream is a specific kind of Duplex stream where the readable side's output is derived from whatever was written to the writable side — a gzip compressor or a CSV parser, where what comes out is a direct function of what went in.