Pipe & Pipeline

readable.pipe(writable)

.pipe() connects a Readable directly to a Writable, automatically managing the flow of data between them. It handles backpressure: when the writable signals its buffer is full (by returning false from write()), pipe pauses the readable and resumes it on drain.

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

const readable = fs.createReadStream('./input.txt');
const writable = fs.createWriteStream('./output.txt');

readable.pipe(writable);

You can chain .pipe() through Transform streams to build processing pipelines:

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

fs.createReadStream('./file.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('./file.txt.gz'));

The problem with .pipe()

.pipe() has a well-known flaw: it does not propagate errors. If any stream in the chain emits an error event, the other streams are not automatically destroyed or closed. You end up with open file handles and memory leaks.

// Dangerous: error on readable does NOT close writable
readable.pipe(writable);

readable.on('error', function(err) {
  console.error('read error — but writable is still open!', err.message);
  writable.destroy();   // you must clean up manually
});

stream.pipeline() — the safe alternative

stream.pipeline(...streams, callback) was added in Node 10 precisely to fix this. It:

Wires up all the streams end-to-end.
Propagates errors from any stream to the callback.
Automatically destroys all streams in the chain on error.

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

pipeline(
  fs.createReadStream('./input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('./input.txt.gz'),
  function(err) {
    if (err) {
      console.error('pipeline failed:', err.message);
    } else {
      console.log('pipeline complete');
    }
  }
);

Node 15+ also ships stream/promises which gives you an await-friendly version:

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

await pipeline(
  fs.createReadStream('./input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('./input.txt.gz')
);

Runnable demo

The demo below builds a short in-memory pipeline: a Readable source produces three strings, a Transform uppercases each chunk, and a Writable collects the results.

Node.js

const { Readable, Transform, Writable, pipeline } = require('node:stream');

const source = Readable.from(['hello', ' world', ' streams']);

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  }
});

const collected = [];
const sink = new Writable({
  write(chunk, encoding, callback) {
    collected.push(chunk.toString());
    callback();
  }
});

pipeline(source, upper, sink, function(err) {
  if (err) {
    console.error('pipeline error:', err.message);
  } else {
    console.log('result: ' + collected.join(''));
  }
});

Needs the Node.js runtime — open in StackBlitz to run.