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');

fs.writeFileSync('./input.txt', 'hello world'); // demo fixture

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 fs = require('node:fs');
const zlib = require('node:zlib');

fs.writeFileSync('./file.txt', 'hello world'); // demo fixture

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.

// @skip-verify illustrative fragment: readable/writable continue the example above
// 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 — no callback, and a normal rejected promise on error:

Node.js

import { pipeline } from 'node:stream/promises';
import { Readable, Writable } from 'node:stream';
import { createGzip, createGunzip } from 'node:zlib';

var source = Readable.from(['hello world, compressed and decompressed via a stream/promises pipeline']);
var chunks = [];
var sink = new Writable({
  write(chunk, encoding, callback) { chunks.push(chunk); callback(); }
});

await pipeline(source, createGzip(), createGunzip(), sink);
console.log('round-tripped:', Buffer.concat(chunks).toString());
console.log('pipeline() resolved -- no callback needed with stream/promises');

Output (verified on Node 24)

round-tripped: hello world, compressed and decompressed via a stream/promises pipeline
pipeline() resolved -- no callback needed with stream/promises

Cancelling a pipeline with AbortSignal

pipeline() (both the callback and promise forms) accepts an options.signal. Aborting it — because a client disconnected, a request timed out, or the user cancelled — destroys every stream in the chain immediately instead of letting a slow source keep running to no purpose:

Node.js

import { pipeline } from 'node:stream/promises';
import { Readable, Writable } from 'node:stream';
import { setTimeout as delay } from 'node:timers/promises';

async function* slowSource() {
  for (let i = 0; i < 20; i++) {
    await delay(10); // simulate a slow upstream (network, DB cursor, ...)
    yield String(i);
  }
}

var controller = new AbortController();
var sink = new Writable({
  write(chunk, enc, cb) { cb(); }
});

// Simulates a client disconnect or a request timeout arriving mid-stream
setTimeout(function() { controller.abort(); }, 30);

try {
  await pipeline(Readable.from(slowSource()), sink, { signal: controller.signal });
  console.log('pipeline finished normally');
} catch (err) {
  console.log('pipeline aborted:', err.name);
}

Output (verified on Node 24)

pipeline aborted: AbortError

Transform streams: reshaping data in flight

The uppercase example above is the simplest possible Transform: one chunk in, one chunk out. A Transform can just as easily change the shape of data — for example, turning a stream of objects into newline-delimited JSON (a common format for logs and streaming APIs), by running in object mode on the input side and byte mode on the output side:

Node.js

import { Transform, Readable } from 'node:stream';

var toJsonLines = new Transform({
  objectMode: true, // accepts JS objects as input chunks
  transform(obj, encoding, callback) {
    callback(null, JSON.stringify(obj) + '\n'); // emits strings/bytes as output
  }
});

var source = Readable.from([{ id: 1 }, { id: 2 }, { id: 3 }], { objectMode: true });

var out = '';
source.pipe(toJsonLines);
for await (const chunk of toJsonLines) {
  out += chunk;
}
console.log(out.trim());

Output (verified on Node 24)

{"id":1}
{"id":2}
{"id":3}

.destroy() and finished() — cleanup on every path

.pipe()’s error-propagation gap (above) means you are responsible for destroying every stream in a chain when one of them errors. stream.finished(stream, callback) gives you one callback that fires exactly once, whether the stream ended successfully, errored, or was destroyed — the reliable way to know “this stream is done, one way or another”:

Node.js

import { Readable, Writable, finished } from 'node:stream';

var source = Readable.from((function* () {
  yield 'a'; yield 'b'; throw new Error('source broke');
})());

var sink = new Writable({ write(chunk, enc, cb) { cb(); } });

finished(sink, function(err) {
  console.log('sink finished callback, err:', err ? err.message : null);
});

source.on('error', function(err) {
  console.log('source error:', err.message);
  sink.destroy(err); // propagate manually -- .pipe() alone would not do this
});

source.pipe(sink);

Output (verified on Node 24)

source error: source broke
sink finished callback, err: source broke

Duplex and compose()

A Duplex stream has independent readable and writable sides — a TCP socket is the canonical example: what you write and what you read are unrelated data. stream.compose(...streams) (Node 16.9+) takes that idea further: it chains several Transform-like streams into a single Duplex, so a multi-step pipeline can be handed around and piped into like any other stream:

Node.js

import { Transform, Readable, compose } from 'node:stream';

var upper = new Transform({ transform(chunk, enc, cb) { cb(null, chunk.toString().toUpperCase()); } });
var addBang = new Transform({ transform(chunk, enc, cb) { cb(null, chunk.toString() + '!'); } });

// compose() fuses two Transforms into one Duplex -- one unit you can pipe into
var combined = compose(upper, addBang);
var source = Readable.from(['node', 'streams']);

var out = '';
source.pipe(combined);
for await (const chunk of combined) {
  out += chunk;
}
console.log('composed result:', out);

Output (verified on Node 24)

composed result: NODE!STREAMS!

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(''));
  }
});

Output (verified on Node 24)

result: HELLO WORLD STREAMS

Mental Model

flowchart TB
  Pipe[".pipe(writable)"] -->|"error on any stream"| Gap["NOT propagated -- you must listen + destroy() manually"]
  Pipeline["pipeline(...streams, cb)"] -->|"error on any stream"| Auto["all streams destroyed automatically, cb(err) called"]
  PipelinePromise["stream/promises pipeline(...)"] -->|"error on any stream"| Reject["all streams destroyed, promise rejects"]
  Pipeline -->|"options.signal"| Abort["AbortSignal cancels the whole chain early"]
  Compose["compose(t1, t2, ...)"] -->|"fuses"| OneDuplex["a single Duplex you can pipe like any stream"]

pipe() leaves error propagation to you; pipeline() (callback or promise) does it automatically and accepts an AbortSignal to cancel early

Trade-offs

Choice	Benefit	Cost
`.pipe()`	Minimal, one-line connection between two streams	No automatic error propagation or cleanup — a leak or hang if you forget to wire it up yourself
`pipeline()` (callback)	Automatic error propagation and cleanup across the whole chain	Callback-style — one more level of nesting if you’re already in async/await code
`stream/promises` `pipeline()`	Same guarantees as the callback form, but `await`-able and composes with `try/catch`/`Promise.all`	Requires an `async` context
`pipeline()` with `AbortSignal`	Cancels a whole chain early on a timeout or client disconnect, freeing resources immediately	The abort surfaces as a rejected promise/callback error — every caller must handle that path, not just the “normal” error path
`compose()`	Turns a multi-step pipeline into one reusable `Duplex` you can export, test, and pipe like any other stream	Adds a layer of indirection — errors and backpressure still flow through every inner stream, just wrapped

Common Misconceptions

“.pipe() and pipeline() are interchangeable — pipeline is just a newer name” — fact: .pipe() never propagates errors between streams; pipeline() was added specifically to fix that, plus automatic cleanup and (later) AbortSignal support.
“An aborted pipeline() just stops silently” — fact: it destroys every stream in the chain and the callback fires (or the promise rejects) with an AbortError — you still need to handle that path.
“compose() creates a new kind of stream” — fact: it returns an ordinary Duplex — the same interface as a TCP socket — that happens to be built by chaining Transforms together internally.

Reading AI-generated code

Does code use .pipe() to connect more than one stream without any error listener on any of them? That’s a guaranteed leak the moment any stream in the chain fails.
Does a snippet build a pipeline() (or stream/promises version) that accepts user-controlled input with no AbortSignal/timeout at all, so a slow or hung client can pin resources indefinitely?
Does code implement a multi-step transformation as several separate .pipe() calls with individually-wired error handling, when compose() would express the same idea as one reusable, testable unit?

🐛 Spot the bug

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

function compressFile(src, dest) {
fs.createReadStream(src)
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream(dest));
}

compressFile('./input.txt', './input.txt.gz');

Show the answer

None of the three streams in this .pipe() chain has an error listener. If input.txt doesn’t exist, or the disk fills up mid-write, or gzip encounters corrupt input, that stream emits 'error' — and because .pipe() doesn’t propagate errors, the other streams in the chain are never destroyed. The read stream’s file descriptor (or the partially-written destination file) can leak. The fix is stream.pipeline(fs.createReadStream(src), zlib.createGzip(), fs.createWriteStream(dest), function(err) { if (err) { /* handle it */ } }), which destroys every stream in the chain automatically on any error.