JavaScript Basics: Connect Stream Steps with pipeline

A stream pipeline connects a source, optional processing steps, and a destination. Awaiting pipeline() tells you when the whole operation has finished or failed. The examples below stay in memory and use fresh streams for every run. They show byte transfer, object transformation, and an expected processing error.

These examples were tested on Linux with Node.js v24.19.0. They use built-in APIs only, need no network access, and include assertions that check the results. See the Node.js stream documentation for the API contract.

Prepare a scratch directory

You need Bash, a text editor, Node.js 24 on your PATH, and the mktemp, rm, and rmdir utilities. Keep using the same terminal. The guarded commands stop the shell if setup fails. Save each named source file as UTF-8; .mjs selects ES modules.

node --version
start_dir="$PWD"
work_dir="$(mktemp -d)" || exit 1
cd -- "$work_dir" || exit 1

The tested runtime prints:

v24.19.0

1. Collect a byte stream

Save this as collect-stream-chunks.mjs:

import assert from 'node:assert/strict';
import { Readable, Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';

const chunks = [];
const destination = new Writable({
  write(chunk, encoding, callback) {
    chunks.push(Buffer.from(chunk));
    callback();
  },
});
await pipeline(
  Readable.from([Buffer.from('hello '), Buffer.from('stream')]),
  destination,
);
const text = Buffer.concat(chunks).toString('utf8');
assert.equal(text, 'hello stream');
console.log(text);

Run it:

node collect-stream-chunks.mjs

Output:

hello stream

A writable’s callback reports that it has finished handling a chunk. Calling it is essential for progress. The example collects all bytes solely to make the result easy to inspect; a real large-file pipeline would usually write to a file or another bounded destination. A chunk is not necessarily a complete word, line, or Unicode character. Combining bytes before decoding avoids assuming that each chunk is a complete UTF-8 string.

2. Transform complete objects

Save this as double-stream-values.mjs:

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

const result = [];
const double = new Transform({
  objectMode: true,
  transform(value, encoding, callback) {
    callback(null, value * 2);
  },
});
const destination = new Writable({
  objectMode: true,
  write(value, encoding, callback) {
    result.push(value);
    callback();
  },
});
await pipeline(Readable.from([2, 4, 6]), double, destination);
assert.deepEqual(result, [4, 8, 12]);
console.log(JSON.stringify(result));

Run it:

node double-stream-values.mjs

Output:

[4,8,12]

Object mode passes JavaScript values rather than requiring strings or buffers. Readable.from() uses object mode by default; the transform and destination enable it explicitly. callback(null, value) means success with one output value. Pipeline coordinates backpressure so upstream steps can slow down when downstream steps cannot keep up. Avoid accumulating every output in an array when the point is to process a large input with bounded memory.

3. Verify a pipeline failure

Save this as reject-invalid-stream-value.mjs:

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

const validate = new Transform({
  objectMode: true,
  transform(value, encoding, callback) {
    if (!Number.isFinite(value)) {
      callback(new TypeError('Expected a finite number'));
      return;
    }
    callback(null, value);
  },
});
const destination = new Writable({
  objectMode: true,
  write(value, encoding, callback) { callback(); },
});
await assert.rejects(
  pipeline(Readable.from([1, 'bad', 3]), validate, destination),
  { name: 'TypeError', message: 'Expected a finite number' },
);
console.log('Pipeline rejected invalid input');

Run it:

node reject-invalid-stream-value.mjs

Output:

Pipeline rejected invalid input

Pass an error to the stream callback instead of throwing from the callback implementation. Pipeline rejects its promise and tears down the connected streams on failure. The assertion deliberately handles the error, so the demonstration exits successfully. In an application, catch the rejection and report or recover appropriately. A failed pipeline does not roll back writes already completed, and streams from a failed run should not be reused for a retry.

Clean up

In the same terminal, remove only the example files, return to your original directory, and remove the now-empty scratch directory:

rm -- collect-stream-chunks.mjs double-stream-values.mjs reject-invalid-stream-value.mjs
cd -- "$start_dir" || exit 1
rmdir -- "$work_dir"
unset start_dir work_dir

Acknowledgements

This post was created with assistance from AI.