Pipes and Transform Streams

A pipe is how Node.js moves data from one stream to another without ever loading the whole thing into memory. Instead of reading a file completely and then writing it back out, you connect a readable stream’s output straight into a writable stream’s input, and Node.js takes care of reading, writing, and pacing the flow automatically. A Transform stream sits in the middle of a pipe and reshapes data as it passes through — compressing it, changing its encoding, parsing it — one chunk at a time, without ever buffering the whole stream in memory. Together, pipes and Transform streams are the backbone of efficient file processing, HTTP handling, and real-time data pipelines in Node.js.

Overview: How Piping Works

Every stream in Node.js is one of four kinds: Readable, Writable, Duplex (both readable and writable), or Transform (a duplex stream that modifies data as it passes through, such as zlib.createGzip()). A pipe connects a source — something readable — to a destination — something writable — and Transform streams can be chained in between so data is reshaped at each stage before it reaches its final destination.

The original method for this is readable.pipe(writable). Calling .pipe() does three things automatically: it listens for data events on the source and calls write() on the destination for every chunk that arrives; it manages backpressure so a fast source cannot overwhelm a slow destination; and it ends the destination stream when the source emits its end event. Because .pipe() returns the destination stream, calls can be chained: a.pipe(b).pipe(c) sends data from a through b and into c.

Backpressure is the mechanism that keeps memory usage bounded no matter how large the data is. Every writable stream has an internal buffer sized by its highWaterMark option — 16 KB by default for byte streams, 16 objects for object-mode streams. When you call write() and that internal buffer is already full, write() returns false. A well-behaved pipe stops reading from the source at that moment, and resumes only after the writable stream emits a drain event signalling that room has freed up. .pipe() implements this pause-and-resume dance for you automatically; if you move data manually with data listeners and manual write() calls instead of piping, you have to implement backpressure handling yourself, or risk unbounded memory growth.

Since pipeline() was added to node:stream (and later to node:stream/promises), it has been the recommended way to wire up any chain of streams, not raw .pipe(). pipeline() connects a source, zero or more Transform streams, and a destination — and, critically, it forwards errors from any stream in the chain to a single place and destroys every stream if one of them fails. A chain of chained .pipe() calls does neither of those things on its own: an error raised by one stream is not automatically delivered to the others, and streams left half-open after a failure can leak open file descriptors or sockets.

A Transform stream is a class that implements a _transform(chunk, encoding, callback) method. Node.js calls _transform once for every chunk pushed into the stream; inside it, you process the chunk and call callback(error, transformedChunk) to emit the result downstream and to signal that the stream is ready to accept the next chunk. An optional _flush(callback) method runs exactly once, after the last chunk has been processed, useful for output that depends on having seen the entire stream — a trailing checksum or a closing bracket in a JSON array, for example.

Events you’ll see on a piped stream

Event	Fires on	Meaning
data	Readable	A chunk is available (only fires once you attach a listener or call .pipe(), which switches the stream to flowing mode).
end	Readable	No more data will be emitted.
drain	Writable	The internal buffer has room again after write() returned false.
finish	Writable	All data has been flushed to the underlying system.
error	Any	Something went wrong; if unhandled, Node.js throws and crashes the process.
close	Any	The underlying resource (file descriptor, socket) has been released.

Syntax

The two shapes you’ll use for wiring streams together, and the shape of a Transform subclass:

Form	What it does
source.pipe(destination)	Connects one readable to one writable (or Transform), managing backpressure automatically. Returns destination, so calls can be chained.
await pipeline(source, ...transforms, destination)	From node:stream/promises. Connects any number of streams, forwards errors from any of them, and destroys every stream in the chain if one fails. The returned promise resolves once the destination finishes.
class MyTransform extends Transform	Defines a custom Transform stream by implementing _transform(chunk, encoding, callback) and, optionally, _flush(callback).

Key parts of a Transform subclass:

• chunk — the incoming piece of data, a Buffer by default, or a string if the source stream was set to a text encoding.
• encoding — the string encoding in use, relevant only when chunk is a string rather than a Buffer.
• callback(error, data) — must be called exactly once per _transform invocation. Pass null as the error and the output chunk to push data downstream, or an Error to abort the whole pipeline.
• _flush(callback) — optional; called once, after all input chunks have been processed, to emit any trailing output.

Examples

Example 1: Copying a file with pipe()

The simplest possible pipe connects a readable file stream directly to a writable one. Node streams the file through a small in-memory buffer instead of loading the whole thing at once, so this works the same way whether the file is 1 KB or 10 GB.

import { createReadStream, createWriteStream } from "node:fs";

const source = createReadStream("input.txt");
const destination = createWriteStream("output.txt");

source.on("error", (err) => console.error("Read error:", err.message));
destination.on("error", (err) => console.error("Write error:", err.message));

source.pipe(destination);

destination.on("finish", () => {
  console.log("Copy complete");
});

Output:

Copy complete

Each stream gets its own error listener before the pipe starts — this matters, because if input.txt does not exist, the error happens on source, not on destination, and an unhandled error event on any stream crashes the Node.js process. The finish event on destination fires once every byte has actually been written to disk, which is the correct signal that the copy is complete — not when .pipe() is called, which only sets the pipe up.

Example 2: Compressing a file with pipeline()

Real code should prefer pipeline() over manual .pipe() chaining, especially once a Transform stream like zlib.createGzip() is involved. Here the file is streamed straight into a gzip Transform and out to a .gz file, without ever holding the whole file in memory.

import { createReadStream, createWriteStream } from "node:fs";
import { createGzip } from "node:zlib";
import { pipeline } from "node:stream/promises";

async function compressFile(inputPath, outputPath) {
  await pipeline(
    createReadStream(inputPath),
    createGzip(),
    createWriteStream(outputPath)
  );
  console.log(`Compressed ${inputPath} to ${outputPath}`);
}

try {
  await compressFile("input.txt", "input.txt.gz");
} catch (err) {
  console.error("Pipeline failed:", err.message);
}

Output:

Compressed input.txt to input.txt.gz

pipeline() takes any number of streams and links them in order: a source, one Transform, and a destination here, though you can chain several Transforms in the middle. If any stream in the chain errors — a missing input file, a full disk, a corrupt write — the try/catch catches it, and pipeline() has already destroyed the other streams for you, so nothing is left half-open.

Example 3: A custom Transform stream

Custom Transform streams are how you plug your own logic into a pipe. This one upper-cases every chunk of text that flows through it, working equally well on a 100-byte file or a multi-gigabyte log file, because it never sees more than one chunk at a time.

import { Transform } from "node:stream";
import { pipeline } from "node:stream/promises";
import { createReadStream, createWriteStream } from "node:fs";

class UpperCaseTransform extends Transform {
  _transform(chunk, encoding, callback) {
    const upper = chunk.toString().toUpperCase();
    callback(null, upper);
  }
}

async function shout(inputPath, outputPath) {
  await pipeline(
    createReadStream(inputPath),
    new UpperCaseTransform(),
    createWriteStream(outputPath)
  );
  console.log("Done shouting");
}

await shout("input.txt", "shout.txt");

Output:

Done shouting

_transform is called once per chunk. Because text files are often split across chunk boundaries in the middle of a line, this particular transform is safe: upper-casing works correctly no matter where a chunk boundary falls. A transform that needs to see whole lines — like a CSV parser — would instead need to buffer partial lines between calls, which is a common reason to reach for a helper like readline or a line-splitting Transform instead of writing one from scratch.

Under the Hood: Order of Operations

When pipeline(source, transform, destination) runs, here is what actually happens, in order:

• Node.js puts source into flowing mode and starts reading chunks from the underlying file descriptor via libuv’s thread pool (file I/O is not truly async at the OS level on most platforms, so Node offloads it to a worker thread).
• Each chunk read from source is pushed into transform‘s internal readable buffer, which calls _transform(chunk, encoding, callback).
• _transform runs synchronously (unless you delay calling callback, e.g. to await an async operation), transforms the chunk, and calls callback(null, output), which pushes output into the transform’s internal writable-to-readable pipe and marks it ready for the next input chunk.
• The transformed chunk is written to destination with write(). If destination‘s internal buffer is full, write() returns false.
• On a false return, the pipe pauses source (and by extension stops feeding transform) until destination emits drain. This is backpressure in action — source is only ever as fast as the slowest stream in the chain.
• Once source emits end, Node calls transform‘s optional _flush(callback) for any trailing output, then ends destination.
• destination emits finish once the underlying resource confirms every byte was written, and pipeline()‘s returned promise resolves.

Notice that the whole chain runs chunk by chunk, interleaved — transform can be processing chunk 3 while destination is still flushing chunk 2 to disk. None of it blocks Node’s single JavaScript thread for long; the actual file and socket I/O happens off-thread in libuv, and your _transform callback is invoked back on the event loop as each result becomes available.

Common Mistakes

Mistake 1: Chaining .pipe() without handling errors

It’s tempting to chain .pipe() calls the way you’d chain array methods. But .pipe() does not forward errors between streams, and an unhandled error event crashes the process:

import { createReadStream, createWriteStream } from "node:fs";
import { createGzip } from "node:zlib";

createReadStream("input.txt")
  .pipe(createGzip())
  .pipe(createWriteStream("output.txt.gz"));

// If input.txt does not exist, the 'error' event fires on the readable
// stream with no listener attached. Node throws and crashes the process,
// and the gzip and write streams are left open, never cleaned up.

If input.txt is missing, source emits error with no listener attached, Node throws, and the process exits — meanwhile the gzip Transform and the destination write stream are left open, having never been told to clean up. The fix is to use pipeline(), as in Example 2, which attaches error forwarding and guarantees every stream in the chain is destroyed if any one of them fails. If you must use raw .pipe() (for example, targeting an older Node.js codebase), attach an explicit error listener to every single stream in the chain, not just the first one.

Mistake 2: Forgetting to call the Transform callback

Every _transform call must invoke its callback exactly once, even if you already called this.push() yourself:

class BrokenTransform extends Transform {
  _transform(chunk, encoding, callback) {
    const upper = chunk.toString().toUpperCase();
    this.push(upper);
    // BUG: callback() is never called, so the stream never signals that
    // it is ready for the next chunk. pipeline() hangs forever and the
    // process never exits.
  }
}

Without the callback, the Transform stream never tells its internal state machine that it’s ready for more input. The stream stalls silently — no error, no crash, just a process that hangs forever and a pipeline() promise that never resolves or rejects. The fix, shown in Example 3, is to always finish with callback(null, output) (or just callback(error) if something went wrong) instead of calling this.push() and forgetting the callback, or calling this.push() in addition to it unnecessarily.

Best Practices

• Prefer pipeline() from node:stream/promises over manual .pipe() chains for anything beyond a quick script — it forwards errors and cleans up automatically.
• Always attach an error listener to every stream you create manually, even ones in the middle of a pipe chain, unless you’re already using pipeline() to handle that for you.
• Never load a whole file into memory with readFile just to write it back out, compress it, or hash it — stream it instead. A pipe’s memory footprint stays roughly constant regardless of file size.
• Keep _transform callbacks synchronous-fast where possible; if a transform needs to do async work (a database lookup, an API call), await it before calling callback, and understand that this will slow the whole pipe down to match, by design.
• Set highWaterMark higher for large, high-throughput streams (bigger chunks, fewer event-loop round trips) and leave it at the default for typical file and network work.
• When writing a custom Transform, call callback exactly once per invocation — never twice, and never zero times.
• Use object mode ({ objectMode: true }) when a Transform’s chunks are JavaScript objects rather than bytes or text, such as parsed database rows flowing toward a JSON serializer.

Practice Exercises

• Write a program that reads a text file, pipes it through a custom Transform stream that counts the number of newline characters seen, and logs the total once the pipeline finishes. (Hint: keep a running count in the Transform instance and log it in _flush.)
• Build a pipe chain with pipeline() that reads a .gz file, decompresses it with zlib.createGunzip(), and writes the decompressed contents to a new file. Confirm it correctly reports an error if you point it at a file that isn’t actually gzipped.
• Write a custom Transform stream that redacts every occurrence of the word “secret” in a text stream, replacing it with “[REDACTED]”, and pipe a file through it with pipeline(). Consider and note, in a comment, what happens if the word “secret” is split across two chunk boundaries.

Summary

• A pipe connects a readable stream’s output to a writable stream’s input, moving data in bounded chunks instead of loading it all into memory.
• .pipe() manages backpressure automatically but does not forward errors between streams — an unhandled error event crashes the process.
• pipeline() from node:stream/promises is the modern, recommended way to connect streams: it forwards errors and destroys every stream in the chain on failure.
• A Transform stream implements _transform(chunk, encoding, callback) to reshape data one chunk at a time, and an optional _flush(callback) for trailing output.
• Backpressure is enforced by write() returning false when a writable’s internal buffer is full, and resolved by the drain event.
• Forgetting to call a Transform’s callback silently hangs the pipeline forever, with no error and no crash — always call it exactly once.