Advanced Streams: pipe & pipeline

Reviewed & published by Brayan K

Streams let Node process data piece by piece as it flows, and pipe() and pipeline() connect readable, transform, and writable streams together while handling backpressure and cleanup for you.

Part of the free Node.js course at LearnCodingFast — hands-on lessons with worked examples and the output they print, plus practice exercises and a quick quiz.

In this lesson you'll go beyond the basics: connect streams with .pipe(), wire multi-stage chains safely with stream.pipeline(), build your own Transform, understand backpressure, and consume a stream with for await.

What You'll Learn in This Lesson

• How readable, writable, and transform streams fit together
• Connecting streams with .pipe() and what it does for you
• Why stream.pipeline() is the safe way to chain streams
• How pipeline() propagates errors and cleans up every stream
• What backpressure is and why pipe/pipeline handle it for you
• Writing a custom Transform and consuming a stream with for await

1️⃣ Connecting Streams with .pipe()

A readable stream is a source of data and a writable stream is a destination. The classic way to connect them is source.pipe(destination). Every chunk the source emits flows into the destination, and when the source ends, the destination is told to finish up.

Crucially, .pipe() manages backpressure automatically: if the writable can't keep up, pipe pauses the readable and resumes it once the writable drains. That single feature is why piping beats manually wiring .on('data') and .write() yourself.

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

// A readable stream that emits three chunks, then ends.
const source = Readable.from(['Buy milk\n', 'Walk dog\n', 'Write code\n']);

// A writable stream that just prints whatever it receives, uppercased.
const sink = new Writable({
  write(chunk, encoding, callback) {
    process.stdout.write(chunk.toString().toUpperCase());
    callback();             // tell Node this chunk is fully handled
  },
});

// .pipe() connects them: every chunk from source flows into sink,
// and pipe() automatically respects BACKPRESSURE — it pauses the
// readable when the writable can't keep up, then resumes it.
source.pipe(sink);

2️⃣ Safe Chains with pipeline() & a Transform

A Transform stream sits in the middle of a chain: it reads chunks and writes new ones. You implement transform(chunk, encoding, callback) and call callback(error, output) — pass null as the error and the data you want to send downstream.

stream.pipeline() (use the promise version from node:stream/promises) connects source → transform → destination, waits for the whole chain to finish, and — unlike pipe — destroys every stream if any stage fails. It's the production-grade way to combine streams.

import { pipeline } from 'node:stream/promises';
import { Readable, Transform } from 'node:stream';
import { createWriteStream, readFileSync } from 'node:fs';

// A Transform sits in the MIDDLE: it reads chunks and writes new ones.
let n = 0;
const numberLines = new Transform({
  transform(chunk, encoding, callback) {
    n += 1;
    // callback(error, dataToPushDownstream)
    callback(null, `${n}: ${chunk.toString().trim()}\n`);
  },
});

const source = Readable.from(['apples', 'pears', 'plums']);

// pipeline() wires the stages together left-to-right, waits for the
// whole thing to finish, and destroys EVERY stream if any stage errors.
await pipeline(source, numberLines, createWriteStream('out.txt'));
console.log('Done. File contents:');
console.log(readFileSync('out.txt', 'utf8'));

3️⃣ Errors, Cleanup & Async Iteration

The biggest reason to reach for pipeline() is error handling. If any stage throws or calls its callback with an error, pipeline rejects with that error and destroys every stream in the chain, so you never leak a half-open file or socket. With the promise API you just wrap it in try/catch.

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

const source = Readable.from(['ok-1', 'ok-2', 'BOOM', 'ok-3']);

const guard = new Transform({
  transform(chunk, encoding, callback) {
    const line = chunk.toString();
    if (line === 'BOOM') {
      callback(new Error('hit a bad chunk'));   // signal failure
      return;
    }
    callback(null, line + '\n');
  },
});

// The whole point of pipeline(): one place to catch errors, and it
// tears down every stream so you never leak a half-open file handle.
try {
  await pipeline(source, guard, process.stdout);
} catch (err) {
  console.log('pipeline failed:', err.message);
  console.log('streams were cleaned up automatically');
}

When you only need to consume a stream (not connect it to a destination), every readable is an async iterable — so a plain for await loop works and applies backpressure naturally.

import { Readable } from 'node:stream';

// Every readable stream is also an async iterable, so you can just
// loop over it with "for await". This is the cleanest way to CONSUME
// a stream when you don't need a full pipeline.
const numbers = Readable.from([10, 20, 30, 40]);

let total = 0;
for await (const n of numbers) {
  total += n;
  console.log('got', n);
}
console.log('total =', total);

Your turn. Fill in the two blanks marked ___ to make a Transform that shouts each word, then run it and compare with the expected output.

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

const words = Readable.from(['stream', 'pipe', 'node']);

// 🎯 YOUR TURN — fill in the two ___ blanks, then run it.
const shout = new Transform({
  transform(chunk, encoding, callback) {
    // 1) Uppercase the chunk and add "!\n" to the end.
    const out = chunk.toString().___() + '!\n';   // 👉 toUpperCase
    // 2) Push it downstream with no error.
    callback(___, out);                            // 👉 null
  },
});

await pipeline(words, shout, process.stdout);

// ✅ Expected output:
//    STREAM!
//    PIPE!
//    NODE!

Mini-Challenge: A Tax-Adding Stream

No blanks this time — just a brief and an outline. Build a Transform that adds 10% tax to each price and prints a grand total at the end using flush(). Run it and check your output against the example in the comments.

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

const prices = Readable.from(['9.99', '19.50', '4.00']);

// 🎯 MINI-CHALLENGE: a price-with-tax Transform
// 1. In transform(): read the chunk as a Number, multiply by 1.1 (10% tax),
//    keep a running sum, and push "<price> -> <withTax>\n" (2 decimals).
// 2. In flush(): push one final line "TOTAL: <sum>\n" after the last chunk.
//    (flush runs once, right before the stream ends — perfect for totals.)
//
// ✅ Example output:
//    9.99 -> 10.99
//    19.50 -> 21.45
//    4.00 -> 4.40
//    TOTAL: 36.84

let sum = 0;
const addTax = new Transform({
  transform(chunk, encoding, callback) {
    // your code here
  },
  flush(callback) {
    // your code here
  },
});

await pipeline(prices, addTax, process.stdout);

Common Errors (and the fix)

• ❌ Your Transform hangs and the program never exits — ✅ You forgot to call callback() in transform. Every chunk MUST end with exactly one callback call (or one this.push() + callback()), or the stream stalls waiting for you.
• ❌ "Cannot pipe, not readable" / chunks come out as [object Object] — ✅ You piped an object-mode stream into one expecting strings/Buffers. Either set { objectMode: true } consistently, or stringify objects in your Transform before pushing.
• ❌ A file handle stays open after an error with .pipe() — ✅ .pipe() doesn't clean up on failure. Switch to pipeline(), which destroys every stream when any stage errors.
• ❌ Memory climbs and the process is killed (OOM) — ✅ You're buffering instead of streaming — e.g. pushing every chunk into an array. Process one chunk at a time in a Transform and let backpressure do its job.
• ❌ await pipeline(...) is a SyntaxError — ✅ Top-level await only works in an ES module. Use a .mjs file (or "type": "module"), and import from node:stream/promises.

Pro Tips

• 💡 Default to pipeline() for any chain you ship. The error propagation and automatic cleanup are worth it even for two stages.
• 💡 Reach for node:stream/promises so you can await and try/catch instead of juggling callbacks.
• 💡 Use flush() for footers/summaries — it's the one hook that runs after the last chunk but before the stream closes.

📋 Quick Reference — Streams

🎉 Lesson Complete!

• ✅ .pipe() connects a readable to a writable and handles backpressure
• ✅ pipeline() chains streams, propagates errors, and cleans up on failure
• ✅ A Transform reads chunks and writes new ones via callback(err, data)
• ✅ flush() lets a Transform emit a final summary chunk
• ✅ Any readable is an async iterable — loop it with for await

Practice quiz

Continue this course

• Previous: Checkpoint: Node Core
• Next: Express Routers & Route Organization — Structure routes with express.Router, params, and mounted sub-apps
• Quick reference: Node.js cheat sheet

Frequently asked questions

What is the difference between .pipe() and stream.pipeline()?

Both connect a readable stream to a writable one (with optional transforms in between), but they differ on error handling. .pipe() does NOT forward errors or clean up the other streams when one fails — if the destination errors, the source can be left open, leaking file handles or sockets. stream.pipeline() (especially the promise version from 'node:stream/promises') wires the stages together, propagates the first error to a single place, and destroys every stream in the chain on failure. For anything beyond a throwaway script, prefer pipeline().

What is backpressure and do I have to handle it myself?

Backpressure is what happens when a readable produces data faster than a writable can consume it. If you ignore it, chunks pile up in memory and your process can run out of RAM. The good news: both .pipe() and pipeline() handle backpressure for you — they pause the readable when the writable's internal buffer is full and resume it once it drains. You only manage backpressure manually if you write raw .on('data') handlers and call .write() yourself, which is exactly why pipe/pipeline are recommended.

When should I use a Transform stream instead of just an array map?

Use a Transform when the data is too big to hold in memory all at once, or arrives over time (a large file, an HTTP response, a database cursor). A Transform processes one chunk at a time and never loads the whole dataset, so it scales to gigabytes with a tiny, constant memory footprint. If your data comfortably fits in an array, a plain .map() is simpler — streams shine specifically for large or continuous data.

Can I just loop over a stream with for await?

Yes. Every readable stream in modern Node is an async iterable, so 'for await (const chunk of stream) { ... }' is a clean, readable way to consume one. It naturally applies backpressure (the loop body must finish before the next chunk is pulled) and integrates with try/catch. Use it when you're consuming a stream and want plain control flow; use pipeline() when you're connecting a readable through transforms into a writable destination.