|$ curl https://forge-ai.dev/api/markdown?path=docs/nodejs/streams

$cat docs/streams-&-backpressure.md

updated Recently·26 min read·published

Streams & Backpressure

◆Node.js◆Streams◆Advanced🎯Free Tools

Introduction

Streams are one of Node.js's most powerful abstractions. They let you process data incrementally instead of loading everything into memory. Whether you are copying files, compressing uploads, transforming CSV rows, or proxying HTTP responses, streams keep memory usage bounded and improve responsiveness.

A stream is an abstract interface for data that flows over time. Node.js provides four fundamental stream types: Readable, Writable, Duplex, and Transform. Each implements the EventEmitter interface and emits events like data, end, error, and finish.

This guide covers creating streams, composing them with pipeline, handling backpressure, using object mode, and avoiding the most common production mistakes.

Stream Types

Type	Direction	Use Case
Readable	Source	File reads, HTTP requests, generators
Writable	Destination	File writes, HTTP responses, databases
Duplex	Both	TCP sockets, TLS connections
Transform	Both, modified	Compression, encryption, parsing

All streams inherit from EventEmitter. This means you attach listeners for events, but it also means unhandled error events can crash the process. Always attach error handlers or use pipeline.

Readable Streams

A Readable stream produces data. You can consume it by attaching a data event listener, by piping it to a Writable, or by using async iteration. Async iteration is the cleanest modern approach.

readable-events.js

JavaScript

1	const fs = require('fs');
2	const readStream = fs.createReadStream('./large-file.txt', {
3	highWaterMark: 64 * 1024 // 64 KB chunks
4	});
5
6	// Event-based consumption
7	readStream.on('data', (chunk) => {
8	console.log('Received chunk:', chunk.length);
9	});
10
11	readStream.on('end', () => {
12	console.log('Finished reading');
13	});
14
15	readStream.on('error', (err) => {
16	console.error('Stream error:', err.message);
17	});

readable-async.js

JavaScript

1	// Async iteration (preferred)
2	const fs = require('fs');
3
4	async function processFile(path) {
5	const stream = fs.createReadStream(path);
6	for await (const chunk of stream) {
7	console.log('Chunk size:', chunk.length);
8	}
9	}
10
11	processFile('./large-file.txt').catch(console.error);

✓

best practice

Prefer for await...of over data events. Async iteration pauses naturally for backpressure, produces cleaner stack traces, and works with try/catch for errors.

Writable Streams

A Writable stream consumes data. The write() method returns a boolean indicating whether it is safe to continue writing. When it returns false, you should wait for the drain event before writing more.

writable-backpressure.js

JavaScript

1	const fs = require('fs');
2	const writeStream = fs.createWriteStream('./output.txt');
3
4	function writeLots() {
5	let i = 0;
6	const max = 1_000_000;
7
8	function write() {
9	let ok = true;
10	while (i < max && ok) {
11	ok = writeStream.write(`line ${i++}\n`);
12	}
13
14	if (i < max) {
15	// Wait for drain before continuing
16	writeStream.once('drain', write);
17	}
18	}
19
20	write();
21	}
22
23	writeStream.on('finish', () => console.log('Done writing'));
24	writeLots();

Ignoring backpressure causes memory to grow without bound. The Writable buffers incoming chunks faster than it can flush them, eventually exhausting process memory.

⚠

warning

Always respect the return value of write(). If it returns false, stop writing and wait for drain.

Transform Streams

Transform streams sit between a source and destination, modifying data as it passes through. They are both Readable and Writable. Implement the transform function to process each chunk and call callback() when done.

transform.js

JavaScript

1	const { Transform } = require('stream');
2
3	const csvParser = new Transform({
4	objectMode: true,
5	transform(chunk, encoding, callback) {
6	// chunk is a line string
7	const cells = chunk.toString().split(',');
8	this.push(cells);
9	callback();
10	}
11	});
12
13	// Line splitter transform
14	const splitLines = new Transform({
15	transform(chunk, encoding, callback) {
16	const data = this._buffer ? this._buffer + chunk.toString() : chunk.toString();
17	const lines = data.split('\n');
18	this._buffer = lines.pop(); // keep incomplete line
19
20	for (const line of lines) {
21	this.push(line);
22	}
23	callback();
24	},
25	flush(callback) {
26	if (this._buffer) {
27	this.push(this._buffer);
28	}
29	callback();
30	}
31	});

ℹ

info

Transform streams must handle chunks that split in the middle of logical records. Buffer incomplete records internally and flush them in the flush method.

pipeline and pipe

stream.pipeline() connects multiple streams and automatically handles errors, cleanup, and backpressure. It is the recommended way to compose streams in production. The older .pipe() method does not forward errors and can leak resources.

pipeline.js

JavaScript

1	const { pipeline } = require('stream/promises');
2	const fs = require('fs');
3	const zlib = require('zlib');
4
5	async function compress(inputPath, outputPath) {
6	await pipeline(
7	fs.createReadStream(inputPath),
8	zlib.createGzip(),
9	fs.createWriteStream(outputPath)
10	);
11	console.log('Compression complete');
12	}
13
14	compress('./large.txt', './large.txt.gz').catch((err) => {
15	console.error('Pipeline failed:', err);
16	process.exit(1);
17	});

pipeline returns a promise that resolves when all streams finish or rejects when any stream emits an error. It also destroys streams appropriately to release file descriptors and sockets.

✓

best practice

Never use .pipe() without manually handling errors on every stream. Use stream/promises pipeline instead.

Object Mode

By default streams operate on Buffers or strings. Object mode allows streams to pass JavaScript objects between stages. This is useful for ETL pipelines, database record processing, and message queues.

object-mode.js

JavaScript

1	const { Transform } = require('stream');
2
3	const parseUser = new Transform({
4	objectMode: true,
5	transform(chunk, encoding, callback) {
6	try {
7	const user = JSON.parse(chunk);
8	this.push(user);
9	callback();
10	} catch (err) {
11	callback(err);
12	}
13	}
14	});
15
16	const validateUser = new Transform({
17	objectMode: true,
18	transform(user, encoding, callback) {
19	if (!user.email \|\| !user.name) {
20	callback(new Error('Invalid user record'));
21	return;
22	}
23	this.push(user);
24	callback();
25	}
26	});

⚠

warning

Object mode changes backpressure semantics. Chunks are counted as objects, not bytes. Tune highWaterMark for object throughput rather than byte size.

Common Mistakes

Mistake	Consequence	Solution
Ignoring backpressure	Memory growth, OOM	Use pipeline / async iteration
Using .pipe() without error handling	Silent failures, leaked FDs	Use stream.pipeline
No error listener	Process crash	Attach error handlers or pipeline
Buffering entire stream	Defeats purpose of streams	Process incrementally

$Blueprint — Engineering Documentation·Section ID: NODE-04·Revision: 1.0

X GitHub YouTube Instagram Facebook Discord Slack Telegram Buy me a coffee Ko-fi Sponsor