Piping a Readable through a Transform stream with backpressure

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

const upperCaseChunks = new Transform({
  highWaterMark: 16 * 1024, // downstream buffer fills before upstream is paused
  transform(chunk: Buffer, _encoding, callback) {
    callback(null, chunk.toString("utf8").toUpperCase());
  },
});

export const upperCaseFile = async (sourcePath: string, destPath: string) => {
  const source = createReadStream(sourcePath, { highWaterMark: 64 * 1024 });
  const destination = createWriteStream(destPath);
  await pipeline(source, upperCaseChunks, destination);
};

What

upperCaseFile reads a file, uppercases it chunk by chunk through a Transform stream, and writes the result — using pipeline() so the whole chain respects backpressure automatically instead of buffering the file in memory.

Why it matters

readable.on("data", chunk => ...) reads as fast as the source can produce bytes, regardless of whether whatever's downstream can keep up. Point that at a slow network write or a CPU-heavy transform and Node happily buffers the entire backlog in memory — for a large file, that's the process getting OOM-killed for something a stream was supposed to make cheap.

The fix isn't a bigger buffer, it's respecting the signal each stream already gives you: a Writable's write() returns false when its internal buffer is full, and it emits 'drain' when it's ready for more. Transform streams (which are Duplex under the hood) propagate that signal both ways — they won't pull more from upstream until their own downstream has room. pipeline() from node:stream/promises wires all of that up correctly, plus destroys every stream in the chain and forwards the first error if any stage fails.

How it works

• upperCaseChunks extends Transform with a transform() method that receives each chunk, uppercases it, and calls callback(null, result) — the callback is what actually signals "I'm done with this chunk," which is the internal mechanism that makes backpressure work.
• highWaterMark: 16 * 1024 caps how much transformed output can sit in the Transform's internal buffer before it stops accepting new input from upstream — deliberately small here to make the effect visible on a large file.
• pipeline(source, upperCaseChunks, destination) connects all three streams and returns a promise that resolves when the last stream finishes and rejects if any stream errors — no manual 'error' listeners on each stream needed.
• Because pipeline() is awaited, upperCaseFile itself only resolves once every byte has actually landed in destPath, not just once it's been read from sourcePath.

Gotchas

• Never call .pipe() manually and skip error handling "for now" — an unhandled 'error' on any stream in an unpiloted chain crashes the process, and the source stream never gets .destroy()ed if the destination errors. pipeline() exists specifically to close that gap.
• Transform's callback accepts a second argument for transformed output or you can call this.push() from inside transform() — mixing both for the same chunk double-emits it.
• A Transform that does synchronous CPU-heavy work in transform() still blocks the event loop like any other synchronous code; backpressure controls memory, not concurrency.

Related

• stream.Readable.from() — the other common stream entry point, for turning a plain (async) iterable into a Readable instead of a file.
• events.once(stream, "finish") — a lower-level way to await stream completion when you need more control than pipeline() gives you.