Readable Streams, Deep
Flowing vs paused mode, object mode, Readable.from, and for-await consumption.
A Readable stream has two consumption modes and choosing wrong causes lost or stuck data.
- Flowing — you attach a
'data'listener (orpipe) and chunks are pushed at you as fast as they arrive. - Paused — you call
read()yourself, pulling when ready.
Modern code sidesteps the choice with for await … of, which consumes a readable with correct backpressure and error propagation built in.
const fs = require('node:fs');
async function countLines(path) {
let bytes = 0;
for await (const chunk of fs.createReadStream(path)) {
bytes += chunk.length; // chunk is a Buffer
}
return bytes;
}Object mode
By default streams carry Buffers/strings. In object mode each chunk is an arbitrary JS value — the basis for record pipelines (rows from a DB cursor, parsed CSV objects). Readable.from(iterable) turns any (async) iterable or generator into an object-mode stream, which is the easiest way to feed a pipeline.
Example
// A custom object-mode Readable that paginates an API into a record stream,
// consumable with for-await like any other source. push(null) signals EOF.
const { Readable } = require('node:stream');
function paginate(fetchPage) {
let page = 0;
let done = false;
return new Readable({
objectMode: true,
// _read is called when the consumer wants more; respect backpressure by
// pushing exactly one page per call and waiting to be asked again.
async read() {
if (done) return;
try {
const rows = await fetchPage(page++);
if (rows.length === 0) { done = true; this.push(null); return; }
for (const row of rows) this.push(row);
} catch (err) {
this.destroy(err); // propagate to the consumer's catch
}
},
});
}
// Fake API: 3 pages of 2 rows, then empty.
const fetchPage = async (p) => (p < 3 ? [{ p, i: 0 }, { p, i: 1 }] : []);
for await (const row of paginate(fetchPage)) {
console.log('row', row);
}
console.log('drained all pages with constant memory');When to use it
- A CSV importer uses a Readable stream in object mode to push one parsed row at a time through a processing pipeline without loading the whole file.
- An HTTP server creates a Readable stream from an S3 object and pipes it directly to the response, never buffering the file in Node memory.
- A database cursor wraps query results in a Readable stream so downstream consumers can apply backpressure and pause fetching when they are busy.
More examples
Create a custom Readable stream
Extends `Readable` with a `_read` method that pushes numbers until the max is reached, then signals end with `null`.
const { Readable } = require('stream');
class CounterStream extends Readable {
constructor(max) {
super();
this.current = 1;
this.max = max;
}
_read() {
if (this.current > this.max) return this.push(null);
this.push(String(this.current++));
}
}
new CounterStream(5).pipe(process.stdout);Consume a Readable with for-await
Iterates a file stream with `for await...of` -- all Readable streams are async iterables in Node 10+.
const { createReadStream } = require('fs');
async function countLines(file) {
const stream = createReadStream(file, { encoding: 'utf8' });
let lines = 0;
for await (const chunk of stream) {
lines += chunk.split('\n').length - 1;
}
return lines;
}
countLines('large.log').then(n => console.log(n, 'lines'));Readable in object mode
Uses `Readable.from()` to turn an array of objects into an async-iterable object-mode Readable stream.
const { Readable } = require('stream');
const objectStream = Readable.from([
{ id: 1, name: 'Alice' },
{ id: 2, name: 'Bob' },
{ id: 3, name: 'Carol' },
]);
for await (const user of objectStream) {
console.log(user.name);
}
Discussion