stream/promises.pipeline

stream/promises.pipeline connects Readable → Transform → Writable with correct error propagation, stream destruction on failure, and a Promise API suited to async/await handlers.

Recipe

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
 
await pipeline(
  createReadStream('input.log'),
  createGzip(),
  createWriteStream('input.log.gz'),
);

When to reach for this:

• Any multi-stage file or HTTP processing chain
• Replacing .on('error') spaghetti on manual pipes
• Express/Fastify download endpoints streaming files
• ETL scripts run via cron or worker processes

Working Example

import { pipeline } from 'node:stream/promises';
import { createReadStream } from 'node:fs';
import { createServer } from 'node:http';
import { Transform } from 'node:stream';
import { createGzip } from 'node:zlib';
 
const lineCount = new Transform({
  transform(chunk, _enc, cb) {
    const lines = String(chunk).split('\n').length - 1;
    (this as Transform & { lines?: number }).lines =
      ((this as Transform & { lines?: number }).lines ?? 0) + lines;
    cb(null, chunk);
  },
});
 
const server = createServer(async (req, res) => {
  if (req.url !== '/export') {
    res.writeHead(404).end();
    return;
  }
 
  try {
    res.writeHead(200, {
      'content-type': 'application/gzip',
      'content-disposition': 'attachment; filename="export.log.gz"',
    });
 
    await pipeline(
      createReadStream('app.log'),
      lineCount,
      createGzip(),
      res,
    );
    console.log('lines processed', (lineCount as Transform & { lines?: number }).lines);
  } catch (err) {
    if (!res.headersSent) res.writeHead(500);
    res.end();
    console.error('pipeline failed', err);
  }
});
 
server.listen(3000);

What this demonstrates:

• pipeline accepts Writable HTTP res as final sink
• Errors after headersSent require ending response without second status line
• Intermediate Transform can accumulate metrics while passing bytes through
• await integrates with try/catch like any async I/O

Deep Dive

How It Works

• On error, pipeline destroys participating streams with the error.
• On completion, streams are closed in order - Writable gets end.
• Supports AbortSignal in Node 24 for cancellation mid-flight.
• Callback version exists in node:stream - Promise variant preferred in async code.

pipeline vs pipe

TypeScript Notes

import { pipeline } from 'node:stream/promises';
import type { Readable, Writable } from 'node:stream';
 
export async function safePump(
  source: Readable,
  sink: Writable,
  signal?: AbortSignal,
): Promise<void> {
  await pipeline(source, sink, { signal });
}

Gotchas

• Not catching pipeline rejection - unhandled rejection crashes process. Fix: try/catch in route handlers.
• Headers after piping started - must writeHead before pipeline to sink response. Fix: set headers first.
• Client abort mid-stream - treat as non-fatal; listen req.on('aborted') optionally. Fix: pass AbortSignal linked to request.
• Mixing pipeline and manual write to same Writable - corrupted output. Fix: one pipeline per response.
• Transform not flushing partial state - implement _flush for trailing buffered data.

Alternatives

FAQs

Related

• Streams Basics - pipe introduction
• Backpressure - flow control
• Streaming HTTP Responses - proxies and headers
• Streams Best Practices - always pipeline rule

Stack versions: This page was written for Node.js 24.18.0 (Active LTS), npm 10+, TypeScript 5.6+, Express 5, Fastify 5, and NestJS 11.