stream.Readable
Create a source stream that produces data on demand for consumers to read.
new stream.Readable([options])- Added in Node.js v0.9.4
Parameters
| Parameter | Type | Description |
|---|---|---|
$optionsoptional | Object | Configuration for the stream. All fields are optional. |
$options.highWaterMarkoptional | number= See stream.getDefaultHighWaterMark() | The maximum number of bytes to store in the internal buffer before ceasing to read from the underlying resource. |
$options.encodingoptional | string= null | If specified, buffers are decoded to strings using this encoding. |
$options.objectModeoptional | boolean= false | Whether this stream should behave as a stream of objects, meaning stream.read(n) returns a single value instead of a Buffer of size n. |
$options.emitCloseoptional | boolean= true | Whether the stream should emit 'close' after it has been destroyed. |
$options.readoptional | Function | Implementation for the stream._read() method, called as read(size). |
$options.destroyoptional | Function | Implementation for the stream._destroy() method, called as destroy(error, callback). |
$options.constructoptional | Function | Implementation for the stream._construct() method, called as construct(callback). |
$options.autoDestroyoptional | boolean= true | Whether this stream should automatically call .destroy() on itself after ending. |
$options.signaloptional | AbortSignal | A signal representing possible cancellation; aborting it destroys the stream. |
Return value
stream.ReadableA new readable stream instance.
Errors
Reading from a Readable that provides no _read() implementation emits an ERR_METHOD_NOT_IMPLEMENTED error, and push(null) followed by another push() emits ERR_STREAM_PUSH_AFTER_EOF.
Changelog
- PHP v22.0.0The default highWaterMark was raised, so code that counted on a 16 KB buffer now buffers more before _read() stops being called.
- PHP v15.5.0Added the signal option for destroying the stream from an AbortController.
- PHP v14.0.0The autoDestroy option now defaults to true, so streams destroy themselves after 'end' and emit 'close' where older code saw neither.
- PHP v11.2.0Added the autoDestroy option.
Readable is the base class for anything that produces data: files, sockets, HTTP responses, and your own sources. You subclass it (or pass a read function to the constructor) and implement _read(size). Node calls _read() whenever the internal buffer has room; you answer by calling this.push(chunk) one or more times, and this.push(null) when there is nothing left.
When to reach for it
Only when you are wrapping a real source that Node does not already model — a database cursor, a message queue, a custom binary protocol. If your data is already an array, a generator, or anything iterable, use Readable.from() instead; it is shorter and correct by construction.
What trips people up
The pull model is the usual surprise. _read() is an invitation, not a command to produce everything — push what you have and return. Push too much and you defeat backpressure; push nothing and never push again and the stream simply stalls, with no error.
The second surprise is that a Readable produces nothing until something consumes it. Attaching a 'data' handler, calling pipe(), calling resume(), or iterating it with for await all start the flow. Creating a stream and only listening for 'end' gives you a stream that never ends.
Set objectMode: true to emit values other than strings and buffers.
Example
When to use it
- Wrap a paginated API cursor so downstream code can consume records with for await
- Expose rows from a database driver as a stream that respects backpressure
- Feed synthetic or fixture data into a pipeline during tests
- Adapt an event-emitting source into something pipe() and pipeline() accept
Frequently asked questions
How do I create a custom readable stream in Node.js?
Extend stream.Readable and implement _read(), pushing chunks with this.push(chunk) and this.push(null) when the source is exhausted; alternatively pass a read function directly to the constructor. Reach for Readable.from() first if your data is already an array, iterable, or generator.
Why does my readable stream never emit data?
A Readable stays paused until something consumes it, so nothing flows until you call pipe(), attach a 'data' listener, call resume(), or iterate it with for await. Listening only for 'end' leaves the stream paused forever.
What does push(null) do in a readable stream?
push(null) signals end-of-stream, causing the 'end' event to fire once buffered data has been read. Pushing anything after push(null) raises ERR_STREAM_PUSH_AFTER_EOF.
More examples
Object mode without subclassing
Passing a read function to the constructor avoids a subclass entirely. Because objectMode is on, each chunk arrives as the original object rather than a Buffer, so row.id is a number. Output is: row: 1 ada / row: 2 grace / row: 3 alan / done.
const { Readable } = require('node:stream');
const records = [
{ id: 1, name: 'ada' },
{ id: 2, name: 'grace' },
{ id: 3, name: 'alan' },
];
let i = 0;
const source = new Readable({
objectMode: true,
read() {
// Push one record per call; push(null) once the array is exhausted.
this.push(i < records.length ? records[i++] : null);
},
});
source.on('data', (row) => console.log('row:', row.id, row.name));
source.on('end', () => console.log('done'));Wrapping a paginated source with async _read
Shows the pattern for an asynchronous _read(): a re-entry guard, pushing a whole page at once, and routing failures to destroy(err) so consumers see an 'error' event instead of a hung stream. Prints a, b, c, then end.
const { Readable } = require('node:stream');
// Pretend this hits a network API that returns one page at a time.
async function fetchPage(cursor) {
const pages = {
null: { items: ['a', 'b'], next: 'p2' },
p2: { items: ['c'], next: null },
};
return pages[cursor];
}
class PagedSource extends Readable {
constructor() {
super({ objectMode: true });
this.cursor = null;
this.started = false;
}
_read() {
// Guard against re-entry: _read() can be called again while awaiting.
if (this.fetching) return;
this.fetching = true;
fetchPage(this.started ? this.cursor : null)
.then((page) => {
this.started = true;
this.fetching = false;
for (const item of page.items) this.push(item);
this.cursor = page.next;
if (!page.next) this.push(null);
})
.catch((err) => this.destroy(err));
}
}
new PagedSource().on('data', console.log).on('end', () => console.log('end'));
Discussion