Event:'close'#

The'close' event is emitted when the stream and any of its underlyingresources (a file descriptor, for example) have been closed. The event indicatesthat no more events will be emitted, and no further computation will occur.

AWritable stream will always emit the'close' event if it iscreated with theemitClose option.

Event:'drain'#

If a call tostream.write(chunk) returnsfalse, the'drain' event will be emitted when it is appropriate to resume writing datato the stream.

// Write the data to the supplied writable stream one million times.// Be attentive to back-pressure.functionwriteOneMillionTimes(writer, data, encoding, callback) {let i =1000000;write();functionwrite() {let ok =true;do {      i--;if (i ===0) {// Last time!        writer.write(data, encoding, callback);      }else {// See if we should continue, or wait.// Don't pass the callback, because we're not done yet.        ok = writer.write(data, encoding);      }    }while (i >0 && ok);if (i >0) {// Had to stop early!// Write some more once it drains.      writer.once('drain', write);    }  }}copy

Event:'error'#

• <Error>

The'error' event is emitted if an error occurred while writing or pipingdata. The listener callback is passed a singleError argument when called.

The stream is closed when the'error' event is emitted unless theautoDestroy option was set tofalse when creating thestream.

After'error', no further events other than'close'should be emitted(including'error' events).

Event:'finish'#

The'finish' event is emitted after thestream.end() methodhas been called, and all data has been flushed to the underlying system.

const writer =getWritableStreamSomehow();for (let i =0; i <100; i++) {  writer.write(`hello, #${i}!\n`);}writer.on('finish',() => {console.log('All writes are now complete.');});writer.end('This is the end\n');copy

Event:'pipe'#

• src<stream.Readable> source stream that is piping to this writable

The'pipe' event is emitted when thestream.pipe() method is called ona readable stream, adding this writable to its set of destinations.

const writer =getWritableStreamSomehow();const reader =getReadableStreamSomehow();writer.on('pipe',(src) => {console.log('Something is piping into the writer.');  assert.equal(src, reader);});reader.pipe(writer);copy

Event:'unpipe'#

• src<stream.Readable> The source stream thatunpiped this writable

The'unpipe' event is emitted when thestream.unpipe() method is calledon aReadable stream, removing thisWritable from its set ofdestinations.

This is also emitted in case thisWritable stream emits an error when aReadable stream pipes into it.

const writer =getWritableStreamSomehow();const reader =getReadableStreamSomehow();writer.on('unpipe',(src) => {console.log('Something has stopped piping into the writer.');  assert.equal(src, reader);});reader.pipe(writer);reader.unpipe(writer);copy

writable.cork()#

Thewritable.cork() method forces all written data to be buffered in memory.The buffered data will be flushed when either thestream.uncork() orstream.end() methods are called.

The primary intent ofwritable.cork() is to accommodate a situation in whichseveral small chunks are written to the stream in rapid succession. Instead ofimmediately forwarding them to the underlying destination,writable.cork()buffers all the chunks untilwritable.uncork() is called, which will pass themall towritable._writev(), if present. This prevents a head-of-line blockingsituation where data is being buffered while waiting for the first small chunkto be processed. However, use ofwritable.cork() without implementingwritable._writev() may have an adverse effect on throughput.

See also:writable.uncork(),writable._writev().

writable.destroy([error])#

• error<Error> Optional, an error to emit with'error' event.
• Returns:<this>

Destroy the stream. Optionally emit an'error' event, and emit a'close'event (unlessemitClose is set tofalse). After this call, the writablestream has ended and subsequent calls towrite() orend() will result inanERR_STREAM_DESTROYED error.This is a destructive and immediate way to destroy a stream. Previous calls towrite() may not have drained, and may trigger anERR_STREAM_DESTROYED error.Useend() instead of destroy if data should flush before close, or wait forthe'drain' event before destroying the stream.

const {Writable } =require('node:stream');const myStream =newWritable();const fooErr =newError('foo error');myStream.destroy(fooErr);myStream.on('error',(fooErr) =>console.error(fooErr.message));// foo errorcopy

const {Writable } =require('node:stream');const myStream =newWritable();myStream.destroy();myStream.on('error',functionwontHappen() {});copy

const {Writable } =require('node:stream');const myStream =newWritable();myStream.destroy();myStream.write('foo',(error) =>console.error(error.code));// ERR_STREAM_DESTROYEDcopy

Oncedestroy() has been called any further calls will be a no-op and nofurther errors except from_destroy() may be emitted as'error'.

Implementors should not override this method,but instead implementwritable._destroy().

writable.closed#

• <boolean>

Istrue after'close' has been emitted.

writable.destroyed#

• <boolean>

Istrue afterwritable.destroy() has been called.

const {Writable } =require('node:stream');const myStream =newWritable();console.log(myStream.destroyed);// falsemyStream.destroy();console.log(myStream.destroyed);// truecopy

writable.end([chunk[, encoding]][, callback])#

• chunk<string> |<Buffer> |<TypedArray> |<DataView> |<any> Optional data to write. Forstreams not operating in object mode,chunk must be a<string>,<Buffer>,<TypedArray> or<DataView>. For object mode streams,chunk may be anyJavaScript value other thannull.
• encoding<string> The encoding ifchunk is a string
• callback<Function> Callback for when the stream is finished.
• Returns:<this>

Calling thewritable.end() method signals that no more data will be writtento theWritable. The optionalchunk andencoding arguments allow onefinal additional chunk of data to be written immediately before closing thestream.

Calling thestream.write() method after callingstream.end() will raise an error.

// Write 'hello, ' and then end with 'world!'.const fs =require('node:fs');const file = fs.createWriteStream('example.txt');file.write('hello, ');file.end('world!');// Writing more now is not allowed!copy

writable.setDefaultEncoding(encoding)#

• encoding<string> The new default encoding
• Returns:<this>

Thewritable.setDefaultEncoding() method sets the defaultencoding for aWritable stream.

writable.uncork()#

Thewritable.uncork() method flushes all data buffered sincestream.cork() was called.

When usingwritable.cork() andwritable.uncork() to manage the bufferingof writes to a stream, defer calls towritable.uncork() usingprocess.nextTick(). Doing so allows batching of allwritable.write() calls that occur within a given Node.js event loop phase.

stream.cork();stream.write('some ');stream.write('data ');process.nextTick(() => stream.uncork());copy

If thewritable.cork() method is called multiple times on a stream, thesame number of calls towritable.uncork() must be called to flush the buffereddata.

stream.cork();stream.write('some ');stream.cork();stream.write('data ');process.nextTick(() => {  stream.uncork();// The data will not be flushed until uncork() is called a second time.  stream.uncork();});copy

See also:writable.cork().

writable.writable#

• <boolean>

Istrue if it is safe to callwritable.write(), which meansthe stream has not been destroyed, errored, or ended.

writable.writableAborted#

• <boolean>

Returns whether the stream was destroyed or errored before emitting'finish'.

writable.writableEnded#

• <boolean>

Istrue afterwritable.end() has been called. This propertydoes not indicate whether the data has been flushed, for this usewritable.writableFinished instead.

writable.writableCorked#

• <integer>

Number of timeswritable.uncork() needs to becalled in order to fully uncork the stream.

writable.errored#

• <Error>

Returns error if the stream has been destroyed with an error.

writable.writableFinished#

• <boolean>

Is set totrue immediately before the'finish' event is emitted.

writable.writableHighWaterMark#

• <number>

Return the value ofhighWaterMark passed when creating thisWritable.

writable.writableLength#

• <number>

This property contains the number of bytes (or objects) in the queueready to be written. The value provides introspection data regardingthe status of thehighWaterMark.

writable.writableNeedDrain#

• <boolean>

Istrue if the stream's buffer has been full and stream will emit'drain'.

writable.writableObjectMode#

• <boolean>

Getter for the propertyobjectMode of a givenWritable stream.

writable[Symbol.asyncDispose]()#

Callswritable.destroy() with anAbortError and returnsa promise that fulfills when the stream is finished.

writable.write(chunk[, encoding][, callback])#

• chunk<string> |<Buffer> |<TypedArray> |<DataView> |<any> Optional data to write. Forstreams not operating in object mode,chunk must be a<string>,<Buffer>,<TypedArray> or<DataView>. For object mode streams,chunk may be anyJavaScript value other thannull.
• encoding<string> |<null> The encoding, ifchunk is a string.Default:'utf8'
• callback<Function> Callback for when this chunk of data is flushed.
• Returns:<boolean>false if the stream wishes for the calling code towait for the'drain' event to be emitted before continuing to writeadditional data; otherwisetrue.

Thewritable.write() method writes some data to the stream, and calls thesuppliedcallback once the data has been fully handled. If an erroroccurs, thecallback will be called with the error as itsfirst argument. Thecallback is called asynchronously and before'error' isemitted.

The return value istrue if the internal buffer is less than thehighWaterMark configured when the stream was created after admittingchunk.Iffalse is returned, further attempts to write data to the stream shouldstop until the'drain' event is emitted.

While a stream is not draining, calls towrite() will bufferchunk, andreturn false. Once all currently buffered chunks are drained (accepted fordelivery by the operating system), the'drain' event will be emitted.Oncewrite() returns false, do not write more chunksuntil the'drain' event is emitted. While callingwrite() on a stream thatis not draining is allowed, Node.js will buffer all written chunks untilmaximum memory usage occurs, at which point it will abort unconditionally.Even before it aborts, high memory usage will cause poor garbage collectorperformance and high RSS (which is not typically released back to the system,even after the memory is no longer required). Since TCP sockets may neverdrain if the remote peer does not read the data, writing a socket that isnot draining may lead to a remotely exploitable vulnerability.

Writing data while the stream is not draining is particularlyproblematic for aTransform, because theTransform streams are pausedby default until they are piped or a'data' or'readable' event handleris added.

If the data to be written can be generated or fetched on demand, it isrecommended to encapsulate the logic into aReadable and usestream.pipe(). However, if callingwrite() is preferred, it ispossible to respect backpressure and avoid memory issues using the'drain' event:

functionwrite(data, cb) {if (!stream.write(data)) {    stream.once('drain', cb);  }else {    process.nextTick(cb);  }}// Wait for cb to be called before doing any other write.write('hello',() => {console.log('Write completed, do more writes now.');});copy

AWritable stream in object mode will always ignore theencoding argument.

Event:'close'#

The'close' event is emitted when the stream and any of its underlyingresources (a file descriptor, for example) have been closed. The event indicatesthat no more events will be emitted, and no further computation will occur.

AReadable stream will always emit the'close' event if it iscreated with theemitClose option.

Event:'data'#

• chunk<Buffer> |<string> |<any> The chunk of data. For streams that are notoperating in object mode, the chunk will be either a string orBuffer.For streams that are in object mode, the chunk can be any JavaScript valueother thannull.

The'data' event is emitted whenever the stream is relinquishing ownership ofa chunk of data to a consumer. This may occur whenever the stream is switchedin flowing mode by callingreadable.pipe(),readable.resume(), or byattaching a listener callback to the'data' event. The'data' event willalso be emitted whenever thereadable.read() method is called and a chunk ofdata is available to be returned.

Attaching a'data' event listener to a stream that has not been explicitlypaused will switch the stream into flowing mode. Data will then be passed assoon as it is available.

The listener callback will be passed the chunk of data as a string if a defaultencoding has been specified for the stream using thereadable.setEncoding() method; otherwise the data will be passed as aBuffer.

const readable =getReadableStreamSomehow();readable.on('data',(chunk) => {console.log(`Received${chunk.length} bytes of data.`);});copy

Event:'end'#

The'end' event is emitted when there is no more data to be consumed fromthe stream.

The'end' eventwill not be emitted unless the data is completelyconsumed. This can be accomplished by switching the stream into flowing mode,or by callingstream.read() repeatedly until all data has beenconsumed.

const readable =getReadableStreamSomehow();readable.on('data',(chunk) => {console.log(`Received${chunk.length} bytes of data.`);});readable.on('end',() => {console.log('There will be no more data.');});copy

Event:'error'#

• <Error>

The'error' event may be emitted by aReadable implementation at any time.Typically, this may occur if the underlying stream is unable to generate datadue to an underlying internal failure, or when a stream implementation attemptsto push an invalid chunk of data.

The listener callback will be passed a singleError object.

Event:'pause'#

The'pause' event is emitted whenstream.pause() is calledandreadableFlowing is notfalse.

Event:'readable'#

The'readable' event is emitted when there is data available to be read fromthe stream, up to the configured high water mark (state.highWaterMark). Effectively,it indicates that the stream has new information within the buffer. If data is availablewithin this buffer,stream.read() can be called to retrieve that data.Additionally, the'readable' event may also be emitted when the end of the stream has beenreached.

const readable =getReadableStreamSomehow();readable.on('readable',function() {// There is some data to read now.let data;while ((data =this.read()) !==null) {console.log(data);  }});copy

If the end of the stream has been reached, callingstream.read() will returnnull and trigger the'end'event. This is also true if there never was any data to be read. For instance,in the following example,foo.txt is an empty file:

const fs =require('node:fs');const rr = fs.createReadStream('foo.txt');rr.on('readable',() => {console.log(`readable:${rr.read()}`);});rr.on('end',() => {console.log('end');});copy

The output of running this script is:

$node test.jsreadable: nullendcopy

In some cases, attaching a listener for the'readable' event will cause someamount of data to be read into an internal buffer.

In general, thereadable.pipe() and'data' event mechanisms are easier tounderstand than the'readable' event. However, handling'readable' mightresult in increased throughput.

If both'readable' and'data' are used at the same time,'readable'takes precedence in controlling the flow, i.e.'data' will be emittedonly whenstream.read() is called. ThereadableFlowing property would becomefalse.If there are'data' listeners when'readable' is removed, the streamwill start flowing, i.e.'data' events will be emitted without calling.resume().

Event:'resume'#

The'resume' event is emitted whenstream.resume() iscalled andreadableFlowing is nottrue.

readable.destroy([error])#

• error<Error> Error which will be passed as payload in'error' event
• Returns:<this>

Destroy the stream. Optionally emit an'error' event, and emit a'close'event (unlessemitClose is set tofalse). After this call, the readablestream will release any internal resources and subsequent calls topush()will be ignored.

Oncedestroy() has been called any further calls will be a no-op and nofurther errors except from_destroy() may be emitted as'error'.

Implementors should not override this method, but instead implementreadable._destroy().

readable.closed#

• <boolean>

Istrue after'close' has been emitted.

readable.destroyed#

• <boolean>

Istrue afterreadable.destroy() has been called.

readable.isPaused()#

• Returns:<boolean>

Thereadable.isPaused() method returns the current operating state of theReadable. This is used primarily by the mechanism that underlies thereadable.pipe() method. In most typical cases, there will be no reason touse this method directly.

const readable =new stream.Readable();readable.isPaused();// === falsereadable.pause();readable.isPaused();// === truereadable.resume();readable.isPaused();// === falsecopy

readable.pause()#

• Returns:<this>

Thereadable.pause() method will cause a stream in flowing mode to stopemitting'data' events, switching out of flowing mode. Any data thatbecomes available will remain in the internal buffer.

const readable =getReadableStreamSomehow();readable.on('data',(chunk) => {console.log(`Received${chunk.length} bytes of data.`);  readable.pause();console.log('There will be no additional data for 1 second.');setTimeout(() => {console.log('Now data will start flowing again.');    readable.resume();  },1000);});copy

Thereadable.pause() method has no effect if there is a'readable'event listener.

readable.pipe(destination[, options])#

• destination<stream.Writable> The destination for writing data
• options<Object> Pipe options
  • end<boolean> End the writer when the reader ends.Default:true.
• Returns:<stream.Writable> Thedestination, allowing for a chain of pipes ifit is aDuplex or aTransform stream

Thereadable.pipe() method attaches aWritable stream to thereadable,causing it to switch automatically into flowing mode and push all of its datato the attachedWritable. The flow of data will be automatically managedso that the destinationWritable stream is not overwhelmed by a fasterReadable stream.

The following example pipes all of the data from thereadable into a filenamedfile.txt:

const fs =require('node:fs');const readable =getReadableStreamSomehow();const writable = fs.createWriteStream('file.txt');// All the data from readable goes into 'file.txt'.readable.pipe(writable);copy

It is possible to attach multipleWritable streams to a singleReadablestream.

Thereadable.pipe() method returns a reference to thedestination streammaking it possible to set up chains of piped streams:

const fs =require('node:fs');const zlib =require('node:zlib');const r = fs.createReadStream('file.txt');const z = zlib.createGzip();const w = fs.createWriteStream('file.txt.gz');r.pipe(z).pipe(w);copy

By default,stream.end() is called on the destinationWritablestream when the sourceReadable stream emits'end', so that thedestination is no longer writable. To disable this default behavior, theendoption can be passed asfalse, causing the destination stream to remain open:

reader.pipe(writer, {end:false });reader.on('end',() => {  writer.end('Goodbye\n');});copy

One important caveat is that if theReadable stream emits an error duringprocessing, theWritable destinationis not closed automatically. If anerror occurs, it will be necessary tomanually close each stream in orderto prevent memory leaks.

Theprocess.stderr andprocess.stdoutWritable streams are neverclosed until the Node.js process exits, regardless of the specified options.

readable.read([size])#

• size<number> Optional argument to specify how much data to read.
• Returns:<string> |<Buffer> |<null> |<any>

Thereadable.read() method reads data out of the internal buffer andreturns it. If no data is available to be read,null is returned. By default,the data is returned as aBuffer object unless an encoding has beenspecified using thereadable.setEncoding() method or the stream is operatingin object mode.

The optionalsize argument specifies a specific number of bytes to read. Ifsize bytes are not available to be read,null will be returnedunlessthe stream has ended, in which case all of the data remaining in the internalbuffer will be returned.

If thesize argument is not specified, all of the data contained in theinternal buffer will be returned.

Thesize argument must be less than or equal to 1 GiB.

Thereadable.read() method should only be called onReadable streamsoperating in paused mode. In flowing mode,readable.read() is calledautomatically until the internal buffer is fully drained.

const readable =getReadableStreamSomehow();// 'readable' may be triggered multiple times as data is buffered inreadable.on('readable',() => {let chunk;console.log('Stream is readable (new data received in buffer)');// Use a loop to make sure we read all currently available datawhile (null !== (chunk = readable.read())) {console.log(`Read${chunk.length} bytes of data...`);  }});// 'end' will be triggered once when there is no more data availablereadable.on('end',() => {console.log('Reached end of stream.');});copy

Each call toreadable.read() returns a chunk of data ornull, signifyingthat there's no more data to read at that moment. These chunks aren't automaticallyconcatenated. Because a singleread() call does not return all the data, usinga while loop may be necessary to continuously read chunks until all data is retrieved.When reading a large file,.read() might returnnull temporarily, indicatingthat it has consumed all buffered content but there may be more data yet to bebuffered. In such cases, a new'readable' event is emitted once there's moredata in the buffer, and the'end' event signifies the end of data transmission.

Therefore to read a file's whole contents from areadable, it is necessaryto collect chunks across multiple'readable' events:

const chunks = [];readable.on('readable',() => {let chunk;while (null !== (chunk = readable.read())) {    chunks.push(chunk);  }});readable.on('end',() => {const content = chunks.join('');});copy

AReadable stream in object mode will always return a single item froma call toreadable.read(size), regardless of the value of thesize argument.

If thereadable.read() method returns a chunk of data, a'data' event willalso be emitted.

Callingstream.read([size]) after the'end' event hasbeen emitted will returnnull. No runtime error will be raised.

readable.readable#

• <boolean>

Istrue if it is safe to callreadable.read(), which meansthe stream has not been destroyed or emitted'error' or'end'.

readable.readableAborted#

• <boolean>

Returns whether the stream was destroyed or errored before emitting'end'.

readable.readableDidRead#

• <boolean>

Returns whether'data' has been emitted.

readable.readableEncoding#

• <null> |<string>

Getter for the propertyencoding of a givenReadable stream. Theencodingproperty can be set using thereadable.setEncoding() method.

readable.readableEnded#

• <boolean>

Becomestrue when'end' event is emitted.

readable.errored#

• <Error>

Returns error if the stream has been destroyed with an error.

readable.readableFlowing#

• <boolean>

This property reflects the current state of aReadable stream as describedin theThree states section.

readable.readableHighWaterMark#

• <number>

Returns the value ofhighWaterMark passed when creating thisReadable.

readable.readableLength#

• <number>

This property contains the number of bytes (or objects) in the queueready to be read. The value provides introspection data regardingthe status of thehighWaterMark.

readable.readableObjectMode#

• <boolean>

Getter for the propertyobjectMode of a givenReadable stream.

readable.resume()#

• Returns:<this>

Thereadable.resume() method causes an explicitly pausedReadable stream toresume emitting'data' events, switching the stream into flowing mode.

Thereadable.resume() method can be used to fully consume the data from astream without actually processing any of that data:

getReadableStreamSomehow()  .resume()  .on('end',() => {console.log('Reached the end, but did not read anything.');  });copy

Thereadable.resume() method has no effect if there is a'readable'event listener.

readable.setEncoding(encoding)#

• encoding<string> The encoding to use.
• Returns:<this>

Thereadable.setEncoding() method sets the character encoding fordata read from theReadable stream.

By default, no encoding is assigned and stream data will be returned asBuffer objects. Setting an encoding causes the stream datato be returned as strings of the specified encoding rather than asBufferobjects. For instance, callingreadable.setEncoding('utf8') will cause theoutput data to be interpreted as UTF-8 data, and passed as strings. Callingreadable.setEncoding('hex') will cause the data to be encoded in hexadecimalstring format.

TheReadable stream will properly handle multi-byte characters deliveredthrough the stream that would otherwise become improperly decoded if simplypulled from the stream asBuffer objects.

const readable =getReadableStreamSomehow();readable.setEncoding('utf8');readable.on('data',(chunk) => {  assert.equal(typeof chunk,'string');console.log('Got %d characters of string data:', chunk.length);});copy

readable.unpipe([destination])#

• destination<stream.Writable> Optional specific stream to unpipe
• Returns:<this>

Thereadable.unpipe() method detaches aWritable stream previously attachedusing thestream.pipe() method.

If thedestination is not specified, thenall pipes are detached.

If thedestination is specified, but no pipe is set up for it, thenthe method does nothing.

const fs =require('node:fs');const readable =getReadableStreamSomehow();const writable = fs.createWriteStream('file.txt');// All the data from readable goes into 'file.txt',// but only for the first second.readable.pipe(writable);setTimeout(() => {console.log('Stop writing to file.txt.');  readable.unpipe(writable);console.log('Manually close the file stream.');  writable.end();},1000);copy

readable.unshift(chunk[, encoding])#

• chunk<Buffer> |<TypedArray> |<DataView> |<string> |<null> |<any> Chunk of data to unshiftonto the read queue. For streams not operating in object mode,chunk mustbe a<string>,<Buffer>,<TypedArray>,<DataView> ornull.For object mode streams,chunk may be any JavaScript value.
• encoding<string> Encoding of string chunks. Must be a validBuffer encoding, such as'utf8' or'ascii'.

Passingchunk asnull signals the end of the stream (EOF) and behaves thesame asreadable.push(null), after which no more data can be written. The EOFsignal is put at the end of the buffer and any buffered data will still beflushed.

Thereadable.unshift() method pushes a chunk of data back into the internalbuffer. This is useful in certain situations where a stream is being consumed bycode that needs to "un-consume" some amount of data that it has optimisticallypulled out of the source, so that the data can be passed on to some other party.

Thestream.unshift(chunk) method cannot be called after the'end' eventhas been emitted or a runtime error will be thrown.

Developers usingstream.unshift() often should consider switching touse of aTransform stream instead. See theAPI for stream implementerssection for more information.

// Pull off a header delimited by \n\n.// Use unshift() if we get too much.// Call the callback with (error, header, stream).const {StringDecoder } =require('node:string_decoder');functionparseHeader(stream, callback) {  stream.on('error', callback);  stream.on('readable', onReadable);const decoder =newStringDecoder('utf8');let header ='';functiononReadable() {let chunk;while (null !== (chunk = stream.read())) {const str = decoder.write(chunk);if (str.includes('\n\n')) {// Found the header boundary.const split = str.split(/\n\n/);        header += split.shift();const remaining = split.join('\n\n');const buf =Buffer.from(remaining,'utf8');        stream.removeListener('error', callback);// Remove the 'readable' listener before unshifting.        stream.removeListener('readable', onReadable);if (buf.length)          stream.unshift(buf);// Now the body of the message can be read from the stream.callback(null, header, stream);return;      }// Still reading the header.      header += str;    }  }}copy

Unlikestream.push(chunk),stream.unshift(chunk) will notend the reading process by resetting the internal reading state of the stream.This can cause unexpected results ifreadable.unshift() is called during aread (i.e. from within astream._read() implementation on acustom stream). Following the call toreadable.unshift() with an immediatestream.push('') will reset the reading state appropriately,however it is best to simply avoid callingreadable.unshift() while in theprocess of performing a read.

readable.wrap(stream)#

• stream<Stream> An "old style" readable stream
• Returns:<this>

Prior to Node.js 0.10, streams did not implement the entirenode:streammodule API as it is currently defined. (SeeCompatibility for moreinformation.)

When using an older Node.js library that emits'data' events and has astream.pause() method that is advisory only, thereadable.wrap() method can be used to create aReadable stream that usesthe old stream as its data source.

It will rarely be necessary to usereadable.wrap() but the method has beenprovided as a convenience for interacting with older Node.js applications andlibraries.

const {OldReader } =require('./old-api-module.js');const {Readable } =require('node:stream');const oreader =newOldReader();const myReader =newReadable().wrap(oreader);myReader.on('readable',() => {  myReader.read();// etc.});copy

readable[Symbol.asyncIterator]()#

• Returns:<AsyncIterator> to fully consume the stream.

const fs =require('node:fs');asyncfunctionprint(readable) {  readable.setEncoding('utf8');let data ='';forawait (const chunkof readable) {    data += chunk;  }console.log(data);}print(fs.createReadStream('file')).catch(console.error);copy

If the loop terminates with abreak,return, or athrow, the stream willbe destroyed. In other terms, iterating over a stream will consume the streamfully. The stream will be read in chunks of size equal to thehighWaterMarkoption. In the code example above, data will be in a single chunk if the filehas less then 64 KiB of data because nohighWaterMark option is provided tofs.createReadStream().

readable[Symbol.asyncDispose]()#

Callsreadable.destroy() with anAbortError and returnsa promise that fulfills when the stream is finished.

readable.compose(stream[, options])#

• stream<Stream> |<Iterable> |<AsyncIterable> |<Function>
• options<Object>
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Duplex> a stream composed with the streamstream.

import {Readable }from'node:stream';asyncfunction*splitToWords(source) {forawait (const chunkof source) {const words =String(chunk).split(' ');for (const wordof words) {yield word;    }  }}const wordsStream =Readable.from(['this is','compose as operator']).compose(splitToWords);const words =await wordsStream.toArray();console.log(words);// prints ['this', 'is', 'compose', 'as', 'operator']copy

Seestream.compose for more information.

readable.iterator([options])#

• options<Object>
  • destroyOnReturn<boolean> When set tofalse, callingreturn on theasync iterator, or exiting afor await...of iteration using abreak,return, orthrow will not destroy the stream.Default:true.
• Returns:<AsyncIterator> to consume the stream.

The iterator created by this method gives users the option to cancel thedestruction of the stream if thefor await...of loop is exited byreturn,break, orthrow, or if the iterator should destroy the stream if the streamemitted an error during iteration.

const {Readable } =require('node:stream');asyncfunctionprintIterator(readable) {forawait (const chunkof readable.iterator({destroyOnReturn:false })) {console.log(chunk);// 1break;  }console.log(readable.destroyed);// falseforawait (const chunkof readable.iterator({destroyOnReturn:false })) {console.log(chunk);// Will print 2 and then 3  }console.log(readable.destroyed);// True, stream was totally consumed}asyncfunctionprintSymbolAsyncIterator(readable) {forawait (const chunkof readable) {console.log(chunk);// 1break;  }console.log(readable.destroyed);// true}asyncfunctionshowBoth() {awaitprintIterator(Readable.from([1,2,3]));awaitprintSymbolAsyncIterator(Readable.from([1,2,3]));}showBoth();copy

readable.map(fn[, options])#

• fn<Function> |<AsyncFunction> a function to map over every chunk in thestream.
  • data<any> a chunk of data from the stream.
  • options<Object>
    • signal<AbortSignal> aborted if the stream is destroyed allowing toabort thefn call early.
• options<Object>
  • concurrency<number> the maximum concurrent invocation offn to callon the stream at once.Default:1.
  • highWaterMark<number> how many items to buffer while waiting for userconsumption of the mapped items.Default:concurrency * 2 - 1.
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Readable> a stream mapped with the functionfn.

This method allows mapping over the stream. Thefn function will be calledfor every chunk in the stream. If thefn function returns a promise - thatpromise will beawaited before being passed to the result stream.

import {Readable }from'node:stream';import {Resolver }from'node:dns/promises';// With a synchronous mapper.forawait (const chunkofReadable.from([1,2,3,4]).map((x) => x *2)) {console.log(chunk);// 2, 4, 6, 8}// With an asynchronous mapper, making at most 2 queries at a time.const resolver =newResolver();const dnsResults =Readable.from(['nodejs.org','openjsf.org','www.linuxfoundation.org',]).map((domain) => resolver.resolve4(domain), {concurrency:2 });forawait (const resultof dnsResults) {console.log(result);// Logs the DNS result of resolver.resolve4.}copy

readable.filter(fn[, options])#

• fn<Function> |<AsyncFunction> a function to filter chunks from the stream.
  • data<any> a chunk of data from the stream.
  • options<Object>
    • signal<AbortSignal> aborted if the stream is destroyed allowing toabort thefn call early.
• options<Object>
  • concurrency<number> the maximum concurrent invocation offn to callon the stream at once.Default:1.
  • highWaterMark<number> how many items to buffer while waiting for userconsumption of the filtered items.Default:concurrency * 2 - 1.
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Readable> a stream filtered with the predicatefn.

This method allows filtering the stream. For each chunk in the stream thefnfunction will be called and if it returns a truthy value, the chunk will bepassed to the result stream. If thefn function returns a promise - thatpromise will beawaited.

import {Readable }from'node:stream';import {Resolver }from'node:dns/promises';// With a synchronous predicate.forawait (const chunkofReadable.from([1,2,3,4]).filter((x) => x >2)) {console.log(chunk);// 3, 4}// With an asynchronous predicate, making at most 2 queries at a time.const resolver =newResolver();const dnsResults =Readable.from(['nodejs.org','openjsf.org','www.linuxfoundation.org',]).filter(async (domain) => {const { address } =await resolver.resolve4(domain, {ttl:true });return address.ttl >60;}, {concurrency:2 });forawait (const resultof dnsResults) {// Logs domains with more than 60 seconds on the resolved dns record.console.log(result);}copy

readable.forEach(fn[, options])#

• fn<Function> |<AsyncFunction> a function to call on each chunk of the stream.
  • data<any> a chunk of data from the stream.
  • options<Object>
    • signal<AbortSignal> aborted if the stream is destroyed allowing toabort thefn call early.
• options<Object>
  • concurrency<number> the maximum concurrent invocation offn to callon the stream at once.Default:1.
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Promise> a promise for when the stream has finished.

This method allows iterating a stream. For each chunk in the stream thefn function will be called. If thefn function returns a promise - thatpromise will beawaited.

This method is different fromfor await...of loops in that it can optionallyprocess chunks concurrently. In addition, aforEach iteration can only bestopped by having passed asignal option and aborting the relatedAbortController whilefor await...of can be stopped withbreak orreturn. In either case the stream will be destroyed.

This method is different from listening to the'data' event in that ituses thereadable event in the underlying machinery and can limit thenumber of concurrentfn calls.

import {Readable }from'node:stream';import {Resolver }from'node:dns/promises';// With a synchronous predicate.forawait (const chunkofReadable.from([1,2,3,4]).filter((x) => x >2)) {console.log(chunk);// 3, 4}// With an asynchronous predicate, making at most 2 queries at a time.const resolver =newResolver();const dnsResults =Readable.from(['nodejs.org','openjsf.org','www.linuxfoundation.org',]).map(async (domain) => {const { address } =await resolver.resolve4(domain, {ttl:true });return address;}, {concurrency:2 });await dnsResults.forEach((result) => {// Logs result, similar to `for await (const result of dnsResults)`console.log(result);});console.log('done');// Stream has finishedcopy

readable.toArray([options])#

• options<Object>
  • signal<AbortSignal> allows cancelling the toArray operation if thesignal is aborted.
• Returns:<Promise> a promise containing an array with the contents of thestream.

This method allows easily obtaining the contents of a stream.

As this method reads the entire stream into memory, it negates the benefits ofstreams. It's intended for interoperability and convenience, not as the primaryway to consume streams.

import {Readable }from'node:stream';import {Resolver }from'node:dns/promises';awaitReadable.from([1,2,3,4]).toArray();// [1, 2, 3, 4]// Make dns queries concurrently using .map and collect// the results into an array using toArrayconst dnsResults =awaitReadable.from(['nodejs.org','openjsf.org','www.linuxfoundation.org',]).map(async (domain) => {const { address } =await resolver.resolve4(domain, {ttl:true });return address;}, {concurrency:2 }).toArray();copy

readable.some(fn[, options])#

• fn<Function> |<AsyncFunction> a function to call on each chunk of the stream.
  • data<any> a chunk of data from the stream.
  • options<Object>
    • signal<AbortSignal> aborted if the stream is destroyed allowing toabort thefn call early.
• options<Object>
  • concurrency<number> the maximum concurrent invocation offn to callon the stream at once.Default:1.
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Promise> a promise evaluating totrue iffn returned a truthyvalue for at least one of the chunks.

This method is similar toArray.prototype.some and callsfn on each chunkin the stream until the awaited return value istrue (or any truthy value).Once anfn call on a chunk awaited return value is truthy, the stream isdestroyed and the promise is fulfilled withtrue. If none of thefncalls on the chunks return a truthy value, the promise is fulfilled withfalse.

import {Readable }from'node:stream';import { stat }from'node:fs/promises';// With a synchronous predicate.awaitReadable.from([1,2,3,4]).some((x) => x >2);// trueawaitReadable.from([1,2,3,4]).some((x) => x <0);// false// With an asynchronous predicate, making at most 2 file checks at a time.const anyBigFile =awaitReadable.from(['file1','file2','file3',]).some(async (fileName) => {const stats =awaitstat(fileName);return stats.size >1024 *1024;}, {concurrency:2 });console.log(anyBigFile);// `true` if any file in the list is bigger than 1MBconsole.log('done');// Stream has finishedcopy

readable.find(fn[, options])#

• fn<Function> |<AsyncFunction> a function to call on each chunk of the stream.
  • data<any> a chunk of data from the stream.
  • options<Object>
    • signal<AbortSignal> aborted if the stream is destroyed allowing toabort thefn call early.
• options<Object>
  • concurrency<number> the maximum concurrent invocation offn to callon the stream at once.Default:1.
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Promise> a promise evaluating to the first chunk for whichfnevaluated with a truthy value, orundefined if no element was found.

This method is similar toArray.prototype.find and callsfn on each chunkin the stream to find a chunk with a truthy value forfn. Once anfn call'sawaited return value is truthy, the stream is destroyed and the promise isfulfilled with value for whichfn returned a truthy value. If all of thefn calls on the chunks return a falsy value, the promise is fulfilled withundefined.

import {Readable }from'node:stream';import { stat }from'node:fs/promises';// With a synchronous predicate.awaitReadable.from([1,2,3,4]).find((x) => x >2);// 3awaitReadable.from([1,2,3,4]).find((x) => x >0);// 1awaitReadable.from([1,2,3,4]).find((x) => x >10);// undefined// With an asynchronous predicate, making at most 2 file checks at a time.const foundBigFile =awaitReadable.from(['file1','file2','file3',]).find(async (fileName) => {const stats =awaitstat(fileName);return stats.size >1024 *1024;}, {concurrency:2 });console.log(foundBigFile);// File name of large file, if any file in the list is bigger than 1MBconsole.log('done');// Stream has finishedcopy

readable.every(fn[, options])#

• fn<Function> |<AsyncFunction> a function to call on each chunk of the stream.
  • data<any> a chunk of data from the stream.
  • options<Object>
    • signal<AbortSignal> aborted if the stream is destroyed allowing toabort thefn call early.
• options<Object>
  • concurrency<number> the maximum concurrent invocation offn to callon the stream at once.Default:1.
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Promise> a promise evaluating totrue iffn returned a truthyvalue for all of the chunks.

This method is similar toArray.prototype.every and callsfn on each chunkin the stream to check if all awaited return values are truthy value forfn.Once anfn call on a chunk awaited return value is falsy, the stream isdestroyed and the promise is fulfilled withfalse. If all of thefn callson the chunks return a truthy value, the promise is fulfilled withtrue.

import {Readable }from'node:stream';import { stat }from'node:fs/promises';// With a synchronous predicate.awaitReadable.from([1,2,3,4]).every((x) => x >2);// falseawaitReadable.from([1,2,3,4]).every((x) => x >0);// true// With an asynchronous predicate, making at most 2 file checks at a time.const allBigFiles =awaitReadable.from(['file1','file2','file3',]).every(async (fileName) => {const stats =awaitstat(fileName);return stats.size >1024 *1024;}, {concurrency:2 });// `true` if all files in the list are bigger than 1MiBconsole.log(allBigFiles);console.log('done');// Stream has finishedcopy

readable.flatMap(fn[, options])#

• fn<Function> |<AsyncGeneratorFunction> |<AsyncFunction> a function to map overevery chunk in the stream.
  • data<any> a chunk of data from the stream.
  • options<Object>
    • signal<AbortSignal> aborted if the stream is destroyed allowing toabort thefn call early.
• options<Object>
  • concurrency<number> the maximum concurrent invocation offn to callon the stream at once.Default:1.
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Readable> a stream flat-mapped with the functionfn.

This method returns a new stream by applying the given callback to eachchunk of the stream and then flattening the result.

It is possible to return a stream or another iterable or async iterable fromfn and the result streams will be merged (flattened) into the returnedstream.

import {Readable }from'node:stream';import { createReadStream }from'node:fs';// With a synchronous mapper.forawait (const chunkofReadable.from([1,2,3,4]).flatMap((x) => [x, x])) {console.log(chunk);// 1, 1, 2, 2, 3, 3, 4, 4}// With an asynchronous mapper, combine the contents of 4 filesconst concatResult =Readable.from(['./1.mjs','./2.mjs','./3.mjs','./4.mjs',]).flatMap((fileName) =>createReadStream(fileName));forawait (const resultof concatResult) {// This will contain the contents (all chunks) of all 4 filesconsole.log(result);}copy

readable.drop(limit[, options])#

• limit<number> the number of chunks to drop from the readable.
• options<Object>
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Readable> a stream withlimit chunks dropped.

This method returns a new stream with the firstlimit chunks dropped.

import {Readable }from'node:stream';awaitReadable.from([1,2,3,4]).drop(2).toArray();// [3, 4]copy

readable.take(limit[, options])#

• limit<number> the number of chunks to take from the readable.
• options<Object>
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Readable> a stream withlimit chunks taken.

This method returns a new stream with the firstlimit chunks.

import {Readable }from'node:stream';awaitReadable.from([1,2,3,4]).take(2).toArray();// [1, 2]copy

readable.reduce(fn[, initial[, options]])#

• fn<Function> |<AsyncFunction> a reducer function to call over every chunkin the stream.
  • previous<any> the value obtained from the last call tofn or theinitial value if specified or the first chunk of the stream otherwise.
  • data<any> a chunk of data from the stream.
  • options<Object>
    • signal<AbortSignal> aborted if the stream is destroyed allowing toabort thefn call early.
• initial<any> the initial value to use in the reduction.
• options<Object>
  • signal<AbortSignal> allows destroying the stream if the signal isaborted.
• Returns:<Promise> a promise for the final value of the reduction.

This method callsfn on each chunk of the stream in order, passing it theresult from the calculation on the previous element. It returns a promise forthe final value of the reduction.

If noinitial value is supplied the first chunk of the stream is used as theinitial value. If the stream is empty, the promise is rejected with aTypeError with theERR_INVALID_ARGS code property.

import {Readable }from'node:stream';import { readdir, stat }from'node:fs/promises';import { join }from'node:path';const directoryPath ='./src';const filesInDir =awaitreaddir(directoryPath);const folderSize =awaitReadable.from(filesInDir)  .reduce(async (totalSize, file) => {const { size } =awaitstat(join(directoryPath, file));return totalSize + size;  },0);console.log(folderSize);copy

The reducer function iterates the stream element-by-element which means thatthere is noconcurrency parameter or parallelism. To perform areduceconcurrently, you can extract the async function toreadable.map method.

import {Readable }from'node:stream';import { readdir, stat }from'node:fs/promises';import { join }from'node:path';const directoryPath ='./src';const filesInDir =awaitreaddir(directoryPath);const folderSize =awaitReadable.from(filesInDir)  .map((file) =>stat(join(directoryPath, file)), {concurrency:2 })  .reduce((totalSize, { size }) => totalSize + size,0);console.log(folderSize);copy

duplex.allowHalfOpen#

• <boolean>

Iffalse then the stream will automatically end the writable side when thereadable side ends. Set initially by theallowHalfOpen constructor option,which defaults totrue.

This can be changed manually to change the half-open behavior of an existingDuplex stream instance, but must be changed before the'end' event isemitted.

transform.destroy([error])#

• error<Error>
• Returns:<this>

Destroy the stream, and optionally emit an'error' event. After this call, thetransform stream would release any internal resources.Implementors should not override this method, but instead implementreadable._destroy().The default implementation of_destroy() forTransform also emit'close'unlessemitClose is set in false.

Oncedestroy() has been called, any further calls will be a no-op and nofurther errors except from_destroy() may be emitted as'error'.