Additional notes
History
With the support of async generators and iterators in JavaScript, async generators are effectively a first-class language-level stream construct at this point.
Some common interop cases of using Node.js streams with async generators and async iterators are provided below.
(async function() { for await (const chunk of readable) { console.log(chunk); } })();
Async iterators register a permanent error handler on the stream to prevent any unhandled post-destroy errors.
A Node.js readable stream can be created from an asynchronous generator using
the Readable.from() utility method:
const { Readable } = require('node:stream'); const ac = new AbortController(); const signal = ac.signal; async function * generate() { yield 'a'; await someLongRunningFn({ signal }); yield 'b'; yield 'c'; } const readable = Readable.from(generate()); readable.on('close', () => { ac.abort(); }); readable.on('data', (chunk) => { console.log(chunk); });
When writing to a writable stream from an async iterator, ensure correct
handling of backpressure and errors. stream.pipeline() abstracts away
the handling of backpressure and backpressure-related errors:
const fs = require('node:fs'); const { pipeline } = require('node:stream'); const { pipeline: pipelinePromise } = require('node:stream/promises'); const writable = fs.createWriteStream('./file'); const ac = new AbortController(); const signal = ac.signal; const iterator = createIterator({ signal }); // Callback Pattern pipeline(iterator, writable, (err, value) => { if (err) { console.error(err); } else { console.log(value, 'value returned'); } }).on('close', () => { ac.abort(); }); // Promise Pattern pipelinePromise(iterator, writable) .then((value) => { console.log(value, 'value returned'); }) .catch((err) => { console.error(err); ac.abort(); });
Prior to Node.js 0.10, the Readable stream interface was simpler, but also
less powerful and less useful.
- Rather than waiting for calls to the
stream.read()method,'data'events would begin emitting immediately. Applications that would need to perform some amount of work to decide how to handle data were required to store read data into buffers so the data would not be lost. - The
stream.pause()method was advisory, rather than guaranteed. This meant that it was still necessary to be prepared to receive'data'events even when the stream was in a paused state.
In Node.js 0.10, the Readable class was added. For backward
compatibility with older Node.js programs, Readable streams switch into
"flowing mode" when a 'data' event handler is added, or when the
stream.resume() method is called. The effect is that, even
when not using the new stream.read() method and
'readable' event, it is no longer necessary to worry about losing
'data' chunks.
While most applications will continue to function normally, this introduces an edge case in the following conditions:
- No
'data'event listener is added. - The
stream.resume()method is never called. - The stream is not piped to any writable destination.
For example, consider the following code:
// WARNING! BROKEN! net.createServer((socket) => { // We add an 'end' listener, but never consume the data. socket.on('end', () => { // It will never get here. socket.end('The message was received but was not processed.\n'); }); }).listen(1337);
Prior to Node.js 0.10, the incoming message data would be simply discarded. However, in Node.js 0.10 and beyond, the socket remains paused forever.
The workaround in this situation is to call the
stream.resume() method to begin the flow of data:
// Workaround. net.createServer((socket) => { socket.on('end', () => { socket.end('The message was received but was not processed.\n'); }); // Start the flow of data, discarding it. socket.resume(); }).listen(1337);
In addition to new Readable streams switching into flowing mode,
pre-0.10 style streams can be wrapped in a Readable class using the
readable.wrap() method.
read
read(0): void
There are some cases where it is necessary to trigger a refresh of the
underlying readable stream mechanisms, without actually consuming any
data. In such cases, it is possible to call readable.read(0), which will
always return null.
If the internal read buffer is below the highWaterMark, and the
stream is not currently reading, then calling stream.read(0) will trigger
a low-level stream._read() call.
While most applications will almost never need to do this, there are
situations within Node.js where this is done, particularly in the
Readable stream class internals.
readable.push(''): void
Use of readable.push('') is not recommended.
Pushing a zero-byte string, Buffer, TypedArray or DataView to a stream
that is not in object mode has an interesting side effect.
Because it is a call to
readable.push(), the call will end the reading process.
However, because the argument is an empty string, no data is added to the
readable buffer so there is nothing for a user to consume.
The use of readable.setEncoding() will change the behavior of how the
highWaterMark operates in non-object mode.
Typically, the size of the current buffer is measured against the
highWaterMark in bytes. However, after setEncoding() is called, the
comparison function will begin to measure the buffer's size in characters.
This is not a problem in common cases with latin1 or ascii. But it is
advised to be mindful about this behavior when working with strings that could
contain multi-byte characters.