new ReadableStream([underlyingSource [, strategy]])#

• underlyingSource<Object>
  • start<Function> A user-defined function that is invoked immediately whentheReadableStream is created.
    • controller<ReadableStreamDefaultController> |<ReadableByteStreamController>
    • Returns:undefined or a promise fulfilled withundefined.
  • pull<Function> A user-defined function that is called repeatedly when theReadableStream internal queue is not full. The operation may be sync orasync. If async, the function will not be called again until the previouslyreturned promise is fulfilled.
    • controller<ReadableStreamDefaultController> |<ReadableByteStreamController>
    • Returns: A promise fulfilled withundefined.
  • cancel<Function> A user-defined function that is called when theReadableStream is canceled.
    • reason<any>
    • Returns: A promise fulfilled withundefined.
  • type<string> Must be'bytes' orundefined.
  • autoAllocateChunkSize<number> Used only whentype is equal to'bytes'. When set to a non-zero value a view buffer is automaticallyallocated toReadableByteStreamController.byobRequest. When not setone must use stream's internal queues to transfer data via the defaultreaderReadableStreamDefaultReader.
• strategy<Object>
  • highWaterMark<number> The maximum internal queue size before backpressureis applied.
  • size<Function> A user-defined function used to identify the size of eachchunk of data.
    • chunk<any>
    • Returns:<number>

readableStream.locked#

• Type:<boolean> Set totrue if there is an active reader for this<ReadableStream>.

ThereadableStream.locked property isfalse by default, and isswitched totrue while there is an active reader consuming thestream's data.

readableStream.cancel([reason])#

• reason<any>
• Returns: A promise fulfilled withundefined once cancelation hasbeen completed.

readableStream.getReader([options])#

• options<Object>
  • mode<string>'byob' orundefined
• Returns:<ReadableStreamDefaultReader> |<ReadableStreamBYOBReader>

import {ReadableStream }from'node:stream/web';const stream =newReadableStream();const reader = stream.getReader();console.log(await reader.read());const {ReadableStream } =require('node:stream/web');const stream =newReadableStream();const reader = stream.getReader();reader.read().then(console.log);copy

Causes thereadableStream.locked to betrue.

readableStream.pipeThrough(transform[, options])#

• transform<Object>
  • readable<ReadableStream> TheReadableStream to whichtransform.writable will push the potentially modified datait receives from thisReadableStream.
  • writable<WritableStream> TheWritableStream to which thisReadableStream's data will be written.
• options<Object>
  • preventAbort<boolean> Whentrue, errors in thisReadableStreamwill not causetransform.writable to be aborted.
  • preventCancel<boolean> Whentrue, errors in the destinationtransform.writable do not cause thisReadableStream to becanceled.
  • preventClose<boolean> Whentrue, closing thisReadableStreamdoes not causetransform.writable to be closed.
  • signal<AbortSignal> Allows the transfer of data to be canceledusing an<AbortController>.
• Returns:<ReadableStream> Fromtransform.readable.

Connects this<ReadableStream> to the pair of<ReadableStream> and<WritableStream> provided in thetransform argument such that thedata from this<ReadableStream> is written in totransform.writable,possibly transformed, then pushed totransform.readable. Once thepipeline is configured,transform.readable is returned.

Causes thereadableStream.locked to betrue while the pipe operationis active.

import {ReadableStream,TransformStream,}from'node:stream/web';const stream =newReadableStream({start(controller) {    controller.enqueue('a');  },});const transform =newTransformStream({transform(chunk, controller) {    controller.enqueue(chunk.toUpperCase());  },});const transformedStream = stream.pipeThrough(transform);forawait (const chunkof transformedStream)console.log(chunk);// Prints: Aconst {ReadableStream,TransformStream,} =require('node:stream/web');const stream =newReadableStream({start(controller) {    controller.enqueue('a');  },});const transform =newTransformStream({transform(chunk, controller) {    controller.enqueue(chunk.toUpperCase());  },});const transformedStream = stream.pipeThrough(transform);(async () => {forawait (const chunkof transformedStream)console.log(chunk);// Prints: A})();copy

readableStream.pipeTo(destination[, options])#

• destination<WritableStream> A<WritableStream> to which thisReadableStream's data will be written.
• options<Object>
  • preventAbort<boolean> Whentrue, errors in thisReadableStreamwill not causedestination to be aborted.
  • preventCancel<boolean> Whentrue, errors in thedestinationwill not cause thisReadableStream to be canceled.
  • preventClose<boolean> Whentrue, closing thisReadableStreamdoes not causedestination to be closed.
  • signal<AbortSignal> Allows the transfer of data to be canceledusing an<AbortController>.
• Returns: A promise fulfilled withundefined

Causes thereadableStream.locked to betrue while the pipe operationis active.

readableStream.tee()#

• Returns:<ReadableStream[]>

Returns a pair of new<ReadableStream> instances to which thisReadableStream's data will be forwarded. Each will receive thesame data.

Causes thereadableStream.locked to betrue.

readableStream.values([options])#

• options<Object>
  • preventCancel<boolean> Whentrue, prevents the<ReadableStream>from being closed when the async iterator abruptly terminates.Default:false.

Creates and returns an async iterator usable for consuming thisReadableStream's data.

Causes thereadableStream.locked to betrue while the async iteratoris active.

import {Buffer }from'node:buffer';const stream =newReadableStream(getSomeSource());forawait (const chunkof stream.values({preventCancel:true }))console.log(Buffer.from(chunk).toString());copy

Async Iteration#

The<ReadableStream> object supports the async iterator protocol usingfor await syntax.

import {Buffer }from'node:buffer';const stream =newReadableStream(getSomeSource());forawait (const chunkof stream)console.log(Buffer.from(chunk).toString());copy

The async iterator will consume the<ReadableStream> until it terminates.

By default, if the async iterator exits early (via either abreak,return, or athrow), the<ReadableStream> will be closed. To preventautomatic closing of the<ReadableStream>, use thereadableStream.values()method to acquire the async iterator and set thepreventCancel option totrue.

The<ReadableStream> must not be locked (that is, it must not have an existingactive reader). During the async iteration, the<ReadableStream> will be locked.

Transferring withpostMessage()#

A<ReadableStream> instance can be transferred using a<MessagePort>.

const stream =newReadableStream(getReadableSourceSomehow());const { port1, port2 } =newMessageChannel();port1.onmessage =({ data }) => {  data.getReader().read().then((chunk) => {console.log(chunk);  });};port2.postMessage(stream, [stream]);copy

new ReadableStreamDefaultReader(stream)#

• stream<ReadableStream>

Creates a new<ReadableStreamDefaultReader> that is locked to thegiven<ReadableStream>.

readableStreamDefaultReader.cancel([reason])#

• reason<any>
• Returns: A promise fulfilled withundefined.

Cancels the<ReadableStream> and returns a promise that is fulfilledwhen the underlying stream has been canceled.

readableStreamDefaultReader.closed#

• Type:<Promise> Fulfilled withundefined when the associated<ReadableStream> is closed or rejected if the stream errors or the reader'slock is released before the stream finishes closing.

readableStreamDefaultReader.read()#

• Returns: A promise fulfilled with an object:
  • value<any>
  • done<boolean>

Requests the next chunk of data from the underlying<ReadableStream>and returns a promise that is fulfilled with the data once it isavailable.

readableStreamDefaultReader.releaseLock()#

Releases this reader's lock on the underlying<ReadableStream>.

new ReadableStreamBYOBReader(stream)#

• stream<ReadableStream>

Creates a newReadableStreamBYOBReader that is locked to thegiven<ReadableStream>.

readableStreamBYOBReader.cancel([reason])#

• reason<any>
• Returns: A promise fulfilled withundefined.

Cancels the<ReadableStream> and returns a promise that is fulfilledwhen the underlying stream has been canceled.

readableStreamBYOBReader.closed#

• Type:<Promise> Fulfilled withundefined when the associated<ReadableStream> is closed or rejected if the stream errors or the reader'slock is released before the stream finishes closing.

readableStreamBYOBReader.read(view[, options])#

• view<Buffer> |<TypedArray> |<DataView>
• options<Object>
  • min<number> When set, the returned promise will only befulfilled as soon asmin number of elements are available.When not set, the promise fulfills when at least one elementis available.
• Returns: A promise fulfilled with an object:
  • value<TypedArray> |<DataView>
  • done<boolean>

Requests the next chunk of data from the underlying<ReadableStream>and returns a promise that is fulfilled with the data once it isavailable.

Do not pass a pooled<Buffer> object instance in to this method.PooledBuffer objects are created usingBuffer.allocUnsafe(),orBuffer.from(), or are often returned by variousnode:fs modulecallbacks. These types ofBuffers use a shared underlying<ArrayBuffer> object that contains all of the data from all ofthe pooledBuffer instances. When aBuffer,<TypedArray>,or<DataView> is passed in toreadableStreamBYOBReader.read(),the view's underlyingArrayBuffer isdetached, invalidatingall existing views that may exist on thatArrayBuffer. Thiscan have disastrous consequences for your application.

readableStreamBYOBReader.releaseLock()#

Releases this reader's lock on the underlying<ReadableStream>.

readableStreamDefaultController.close()#

Closes the<ReadableStream> to which this controller is associated.

readableStreamDefaultController.desiredSize#

• Type:<number>

Returns the amount of data remaining to fill the<ReadableStream>'squeue.

readableStreamDefaultController.enqueue([chunk])#

• chunk<any>

Appends a new chunk of data to the<ReadableStream>'s queue.

readableStreamDefaultController.error([error])#

• error<any>

Signals an error that causes the<ReadableStream> to error and close.

readableByteStreamController.byobRequest#

• Type:<ReadableStreamBYOBRequest>

readableByteStreamController.close()#

Closes the<ReadableStream> to which this controller is associated.

readableByteStreamController.desiredSize#

• Type:<number>

Returns the amount of data remaining to fill the<ReadableStream>'squeue.

readableByteStreamController.enqueue(chunk)#

• chunk<Buffer> |<TypedArray> |<DataView>

Appends a new chunk of data to the<ReadableStream>'s queue.

readableByteStreamController.error([error])#

• error<any>

Signals an error that causes the<ReadableStream> to error and close.

readableStreamBYOBRequest.respond(bytesWritten)#

• bytesWritten<number>

Signals that abytesWritten number of bytes have been writtentoreadableStreamBYOBRequest.view.

readableStreamBYOBRequest.respondWithNewView(view)#

• view<Buffer> |<TypedArray> |<DataView>

Signals that the request has been fulfilled with bytes writtento a newBuffer,TypedArray, orDataView.

readableStreamBYOBRequest.view#

• Type:<Buffer> |<TypedArray> |<DataView>

new WritableStream([underlyingSink[, strategy]])#

• underlyingSink<Object>
  • start<Function> A user-defined function that is invoked immediately whentheWritableStream is created.
    • controller<WritableStreamDefaultController>
    • Returns:undefined or a promise fulfilled withundefined.
  • write<Function> A user-defined function that is invoked when a chunk ofdata has been written to theWritableStream.
    • chunk<any>
    • controller<WritableStreamDefaultController>
    • Returns: A promise fulfilled withundefined.
  • close<Function> A user-defined function that is called when theWritableStream is closed.
    • Returns: A promise fulfilled withundefined.
  • abort<Function> A user-defined function that is called to abruptly closetheWritableStream.
    • reason<any>
    • Returns: A promise fulfilled withundefined.
  • type<any> Thetype option is reserved for future use andmust beundefined.
• strategy<Object>
  • highWaterMark<number> The maximum internal queue size before backpressureis applied.
  • size<Function> A user-defined function used to identify the size of eachchunk of data.
    • chunk<any>
    • Returns:<number>

writableStream.abort([reason])#

• reason<any>
• Returns: A promise fulfilled withundefined.

Abruptly terminates theWritableStream. All queued writes will becanceled with their associated promises rejected.

writableStream.close()#

• Returns: A promise fulfilled withundefined.

Closes theWritableStream when no additional writes are expected.

writableStream.getWriter()#

• Returns:<WritableStreamDefaultWriter>

Creates and returns a new writer instance that can be used to writedata into theWritableStream.

writableStream.locked#

• Type:<boolean>

ThewritableStream.locked property isfalse by default, and isswitched totrue while there is an active writer attached to thisWritableStream.

Transferring with postMessage()#

A<WritableStream> instance can be transferred using a<MessagePort>.

const stream =newWritableStream(getWritableSinkSomehow());const { port1, port2 } =newMessageChannel();port1.onmessage =({ data }) => {  data.getWriter().write('hello');};port2.postMessage(stream, [stream]);copy

new WritableStreamDefaultWriter(stream)#

• stream<WritableStream>

Creates a newWritableStreamDefaultWriter that is locked to the givenWritableStream.

writableStreamDefaultWriter.abort([reason])#

• reason<any>
• Returns: A promise fulfilled withundefined.

Abruptly terminates theWritableStream. All queued writes will becanceled with their associated promises rejected.

writableStreamDefaultWriter.close()#

• Returns: A promise fulfilled withundefined.

Closes theWritableStream when no additional writes are expected.

writableStreamDefaultWriter.closed#

• Type:<Promise> Fulfilled withundefined when the associated<WritableStream> is closed or rejected if the stream errors or the writer'slock is released before the stream finishes closing.

writableStreamDefaultWriter.desiredSize#

• Type:<number>

The amount of data required to fill the<WritableStream>'s queue.

writableStreamDefaultWriter.ready#

• Type:<Promise> Fulfilled withundefined when the writer is readyto be used.

writableStreamDefaultWriter.releaseLock()#

Releases this writer's lock on the underlying<ReadableStream>.

writableStreamDefaultWriter.write([chunk])#

• chunk<any>
• Returns: A promise fulfilled withundefined.

Appends a new chunk of data to the<WritableStream>'s queue.

writableStreamDefaultController.error([error])#

• error<any>

Called by user-code to signal that an error has occurred while processingtheWritableStream data. When called, the<WritableStream> will be aborted,with currently pending writes canceled.

writableStreamDefaultController.signal#

• Type:<AbortSignal> AnAbortSignal that can be used to cancel pendingwrite or close operations when a<WritableStream> is aborted.

new TransformStream([transformer[, writableStrategy[, readableStrategy]]])#

• transformer<Object>
  • start<Function> A user-defined function that is invoked immediately whentheTransformStream is created.
    • controller<TransformStreamDefaultController>
    • Returns:undefined or a promise fulfilled withundefined
  • transform<Function> A user-defined function that receives, andpotentially modifies, a chunk of data written totransformStream.writable,before forwarding that on totransformStream.readable.
    • chunk<any>
    • controller<TransformStreamDefaultController>
    • Returns: A promise fulfilled withundefined.
  • flush<Function> A user-defined function that is called immediately beforethe writable side of theTransformStream is closed, signaling the end ofthe transformation process.
    • controller<TransformStreamDefaultController>
    • Returns: A promise fulfilled withundefined.
  • readableType<any> thereadableType option is reserved for future useandmust beundefined.
  • writableType<any> thewritableType option is reserved for future useandmust beundefined.
• writableStrategy<Object>
  • highWaterMark<number> The maximum internal queue size before backpressureis applied.
  • size<Function> A user-defined function used to identify the size of eachchunk of data.
    • chunk<any>
    • Returns:<number>
• readableStrategy<Object>
  • highWaterMark<number> The maximum internal queue size before backpressureis applied.
  • size<Function> A user-defined function used to identify the size of eachchunk of data.
    • chunk<any>
    • Returns:<number>

transformStream.readable#

• Type:<ReadableStream>

transformStream.writable#

• Type:<WritableStream>

Transferring with postMessage()#

A<TransformStream> instance can be transferred using a<MessagePort>.

const stream =newTransformStream();const { port1, port2 } =newMessageChannel();port1.onmessage =({ data }) => {const { writable, readable } = data;// ...};port2.postMessage(stream, [stream]);copy

transformStreamDefaultController.desiredSize#

• Type:<number>

The amount of data required to fill the readable side's queue.

transformStreamDefaultController.enqueue([chunk])#

• chunk<any>

Appends a chunk of data to the readable side's queue.

transformStreamDefaultController.error([reason])#

• reason<any>

Signals to both the readable and writable side that an error has occurredwhile processing the transform data, causing both sides to be abruptlyclosed.

transformStreamDefaultController.terminate()#

Closes the readable side of the transport and causes the writable sideto be abruptly closed with an error.

new ByteLengthQueuingStrategy(init)#

• init<Object>
  • highWaterMark<number>

byteLengthQueuingStrategy.highWaterMark#

• Type:<number>

byteLengthQueuingStrategy.size#

• Type:<Function>
  • chunk<any>
  • Returns:<number>

new CountQueuingStrategy(init)#

• init<Object>
  • highWaterMark<number>

countQueuingStrategy.highWaterMark#

• Type:<number>

countQueuingStrategy.size#

• Type:<Function>
  • chunk<any>
  • Returns:<number>

new TextEncoderStream()#

Creates a newTextEncoderStream instance.

textEncoderStream.encoding#

• Type:<string>

The encoding supported by theTextEncoderStream instance.

textEncoderStream.readable#

• Type:<ReadableStream>

textEncoderStream.writable#

• Type:<WritableStream>

new TextDecoderStream([encoding[, options]])#

• encoding<string> Identifies theencoding that thisTextDecoder instancesupports.Default:'utf-8'.
• options<Object>
  • fatal<boolean>true if decoding failures are fatal.
  • ignoreBOM<boolean> Whentrue, theTextDecoderStream will include thebyte order mark in the decoded result. Whenfalse, the byte order markwill be removed from the output. This option is only used whenencoding is'utf-8','utf-16be', or'utf-16le'.Default:false.

Creates a newTextDecoderStream instance.

textDecoderStream.encoding#

• Type:<string>

The encoding supported by theTextDecoderStream instance.

textDecoderStream.fatal#

• Type:<boolean>

The value will betrue if decoding errors result in aTypeError beingthrown.

textDecoderStream.ignoreBOM#

• Type:<boolean>

The value will betrue if the decoding result will include the byte ordermark.

textDecoderStream.readable#

• Type:<ReadableStream>

textDecoderStream.writable#

• Type:<WritableStream>

new CompressionStream(format)#

• format<string> One of'deflate','deflate-raw','gzip', or'brotli'.

compressionStream.readable#

• Type:<ReadableStream>

compressionStream.writable#

• Type:<WritableStream>

new DecompressionStream(format)#

• format<string> One of'deflate','deflate-raw','gzip', or'brotli'.

decompressionStream.readable#

• Type:<ReadableStream>

decompressionStream.writable#

• Type:<WritableStream>

streamConsumers.arrayBuffer(stream)#

• stream<ReadableStream> |<stream.Readable> |<AsyncIterator>
• Returns:<Promise> Fulfills with anArrayBuffer containing the fullcontents of the stream.

import { arrayBuffer }from'node:stream/consumers';import {Readable }from'node:stream';import {TextEncoder }from'node:util';const encoder =newTextEncoder();const dataArray = encoder.encode('hello world from consumers!');const readable =Readable.from(dataArray);const data =awaitarrayBuffer(readable);console.log(`from readable:${data.byteLength}`);// Prints: from readable: 76const { arrayBuffer } =require('node:stream/consumers');const {Readable } =require('node:stream');const {TextEncoder } =require('node:util');const encoder =newTextEncoder();const dataArray = encoder.encode('hello world from consumers!');const readable =Readable.from(dataArray);arrayBuffer(readable).then((data) => {console.log(`from readable:${data.byteLength}`);// Prints: from readable: 76});copy

streamConsumers.blob(stream)#

• stream<ReadableStream> |<stream.Readable> |<AsyncIterator>
• Returns:<Promise> Fulfills with a<Blob> containing the full contentsof the stream.

import { blob }from'node:stream/consumers';const dataBlob =newBlob(['hello world from consumers!']);const readable = dataBlob.stream();const data =awaitblob(readable);console.log(`from readable:${data.size}`);// Prints: from readable: 27const { blob } =require('node:stream/consumers');const dataBlob =newBlob(['hello world from consumers!']);const readable = dataBlob.stream();blob(readable).then((data) => {console.log(`from readable:${data.size}`);// Prints: from readable: 27});copy

streamConsumers.buffer(stream)#

• stream<ReadableStream> |<stream.Readable> |<AsyncIterator>
• Returns:<Promise> Fulfills with a<Buffer> containing the fullcontents of the stream.

import { buffer }from'node:stream/consumers';import {Readable }from'node:stream';import {Buffer }from'node:buffer';const dataBuffer =Buffer.from('hello world from consumers!');const readable =Readable.from(dataBuffer);const data =awaitbuffer(readable);console.log(`from readable:${data.length}`);// Prints: from readable: 27const { buffer } =require('node:stream/consumers');const {Readable } =require('node:stream');const {Buffer } =require('node:buffer');const dataBuffer =Buffer.from('hello world from consumers!');const readable =Readable.from(dataBuffer);buffer(readable).then((data) => {console.log(`from readable:${data.length}`);// Prints: from readable: 27});copy

streamConsumers.json(stream)#

• stream<ReadableStream> |<stream.Readable> |<AsyncIterator>
• Returns:<Promise> Fulfills with the contents of the stream parsed as aUTF-8 encoded string that is then passed throughJSON.parse().

import { json }from'node:stream/consumers';import {Readable }from'node:stream';const items =Array.from(  {length:100,  },() => ({message:'hello world from consumers!',  }),);const readable =Readable.from(JSON.stringify(items));const data =awaitjson(readable);console.log(`from readable:${data.length}`);// Prints: from readable: 100const { json } =require('node:stream/consumers');const {Readable } =require('node:stream');const items =Array.from(  {length:100,  },() => ({message:'hello world from consumers!',  }),);const readable =Readable.from(JSON.stringify(items));json(readable).then((data) => {console.log(`from readable:${data.length}`);// Prints: from readable: 100});copy

streamConsumers.text(stream)#

• stream<ReadableStream> |<stream.Readable> |<AsyncIterator>
• Returns:<Promise> Fulfills with the contents of the stream parsed as aUTF-8 encoded string.

import { text }from'node:stream/consumers';import {Readable }from'node:stream';const readable =Readable.from('Hello world from consumers!');const data =awaittext(readable);console.log(`from readable:${data.length}`);// Prints: from readable: 27const { text } =require('node:stream/consumers');const {Readable } =require('node:stream');const readable =Readable.from('Hello world from consumers!');text(readable).then((data) => {console.log(`from readable:${data.length}`);// Prints: from readable: 27});copy