stream.pipeline & Error Handling — Node.js

stream.pipeline

stream.pipeline is the correct, modern way to connect streams. It chains any number of streams together and — unlike .pipe() — propagates errors, destroys every stream on failure (no leaked file descriptors), and tells you when the whole flow finishes. If you remember one thing about streams, make it this: reach for pipeline, not pipe.

The promise form (preferred)

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

await pipeline(
  createReadStream('input.txt'),    // source
  createGzip(),                     // transform(s) — any number
  createWriteStream('input.txt.gz'),// destination
)
console.log('finished')             // only reached on full success

Import from `node:stream/promises` for async/await

There are two `pipeline`s: the callback version on `node:stream`, and the promise version on `node:stream/promises`. The promise form integrates with `async/await` and `try/catch` and is the cleanest choice for new code. The order of arguments is always **source → …transforms → destination**.

The callback form

import { pipeline } from 'node:stream'
import { createReadStream, createWriteStream } from 'node:fs'
import { createGzip } from 'node:zlib'

pipeline(
  createReadStream('input.txt'),
  createGzip(),
  createWriteStream('input.txt.gz'),
  (err) => {                        // final callback — always called
    if (err) console.error('pipeline failed:', err)
    else console.log('finished')
  },
)

Why it beats .pipe()

Capability	`.pipe()`	`pipeline()`
Backpressure	Yes	Yes
Propagates errors from any stream	No	Yes
Destroys all streams on failure	No	Yes — no FD leaks
Single completion signal	No	Yes (promise/callback)
Safe for production	Risky	Yes

`.pipe()` leaks resources on error — `pipeline` is the fix

When a stream in a `.pipe()` chain errors, the other streams aren't closed: their file descriptors and sockets stay open, and an unhandled `error` can crash the process. `pipeline` calls `.destroy()` on every stream the moment any one fails, releasing resources and surfacing the error in exactly one place. This is the whole reason it exists.

Mixing in functions and generators

pipeline accepts more than stream objects. An async generator as a middle stage acts as a transform; one as the last stage acts as a sink. This is the most ergonomic way to insert custom logic:

import { pipeline } from 'node:stream/promises'
import { createReadStream } from 'node:fs'

await pipeline(
  createReadStream('access.log', 'utf8'),
  async function* (source) {                 // transform stage
    let leftover = ''
    for await (const chunk of source) {
      const lines = (leftover + chunk).split('\n')
      leftover = lines.pop()
      for (const line of lines) {
        if (line.includes('ERROR')) yield line + '\n'
      }
    }
  },
  async function (errorsOnly) {              // sink stage (no yield)
    let count = 0
    for await (const _ of errorsOnly) count++
    console.log('error lines:', count)
  },
)

A generator that `yield`s is a transform; one that doesn't is a sink

If the final argument is an async function that consumes its source without yielding, it's the destination. A middle async generator must `yield` to pass data downstream. Either way, `pipeline` drives them with proper backpressure and tears everything down on error — the same guarantees as native streams.

Cancellation with AbortSignal

Pass an AbortSignal to abort an in-flight pipeline — for timeouts, request cancellation, or shutdown:

import { pipeline } from 'node:stream/promises'
import { createReadStream, createWriteStream } from 'node:fs'

const ac = new AbortController()
setTimeout(() => ac.abort(), 5000)          // cancel after 5s

try {
  await pipeline(
    createReadStream('huge.dat'),
    createWriteStream('out.dat'),
    { signal: ac.signal },                  // pass the signal as options
  )
} catch (err) {
  if (err.name === 'AbortError') console.log('cancelled')
  else throw err
}

Abort destroys the streams cleanly

When the signal fires, `pipeline` destroys every stream (closing files/sockets) and rejects with an `AbortError`. This gives you leak-free cancellation — far safer than trying to `destroy()` a hand-rolled `.pipe()` chain yourself mid-flight.

finished(): await a single stream's end

A close relative, stream.finished, resolves when one stream is done (or errors) — useful when you're not building a chain but need to know a single stream completed and was cleaned up:

import { finished } from 'node:stream/promises'
import { createReadStream } from 'node:fs'

const rs = createReadStream('file.txt')
rs.resume()                       // drain it somewhere

await finished(rs)                // resolves on 'end'/'close', rejects on 'error'
console.log('stream fully consumed')

Rules of thumb

Use pipeline (promise form) for connecting two or more streams — make it the default.
Insert custom logic as async generator stages instead of Transform subclasses.
Reserve raw .pipe() for trivial throwaway scripts; never in long-running services.
Use finished to await a single stream; use an AbortSignal for cancellable flows.

With buffers and streams in hand, move up the stack to talking over the network: [Networking in Node.js](/nodejs/networking-intro).