Streams

1. Introduction

This section is non-normative.

Large swathes of the web platform are built on streaming data: that is, data that is created, processed, and consumed in an incremental fashion, without ever reading all of it into memory. The Streams Standard provides a common set of APIs for creating and interfacing with such streaming data, embodied in readable streams, writable streams, and transform streams.

This standard provides the base stream primitives which other parts of the web platform can use to expose their streaming data. For example, [FETCH] could expose request bodies as a writable stream, or response bodies as a readable stream. More generally, the platform is full of streaming abstractions waiting to be expressed as streams: multimedia streams, file streams, interprocess communication, and more benefit from being able to process data incrementally instead of buffering it all into memory and processing it in one go. By providing the foundation for these streams to be exposed to developers, the Streams Standard enables use cases like:

• Video effects: piping a readable video stream through a transform stream that applies effects in real time.
• Decompression: piping a file stream through a transform stream that selectively decompresses files from a .tgz archive, turning them into img elements as the user scrolls through an image gallery.
• Image decoding: piping an HTTP response stream through a transform stream that decodes bytes into bitmap data, and then through another transform that translates bitmaps into PNGs. If installed inside the fetch hook of a service worker [SERVICE-WORKERS], this would allow developers to transparently polyfill new image formats.

The APIs described here provide unifying abstraction for all such streams, encouraging an ecosystem to grow around these shared and composable interfaces. At the same time, they have been carefully designed to map efficiently to low-level I/O concerns, and to encapsulate the trickier issues (such as backpressure) that come along for the ride.

2. Model

A chunk#chunkReferenced in:2.1. Readable Streams (2)2.2. Writable Streams2.4. Pipe Chains and Backpressure2.5. Internal Queues and Queuing Strategies (2) (3)2.6. Locking3.1. Using Readable Streams3.2. Class ReadableStream3.2.4.3. getReader(options = {}) (2) (3)3.2.4.6. tee()3.5.2. Internal Slots3.5.4.3. read()3.6.2. Internal Slots3.8.2. Internal Slots (2) (3)3.8.4.2. close()3.8.4.3. enqueue(chunk)3.9.5. ReadableStreamDefaultControllerEnqueue ( controller, chunk )3.10.2. Internal Slots (2) (3)3.10.4.3. close()3.10.4.4. enqueue(chunk)4.1. Using Writable Streams4.2.2. Internal Slots4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {}) (2)4.2.4.6. write(chunk)6.1. Class ByteLengthQueuingStrategy (2) (3)6.2. Class CountQueuingStrategy (2)8.7. A writable stream with backpressure and success signals is a single piece of data that is written to or read from a stream. It can be of any type; streams can even contain chunks of different types. A chunk will often not be the most atomic unit of data for a given stream; for example a byte stream might contain chunks consisting of 16 KiB Uint8Arrays, instead of single bytes.

2.1. Readable Streams

A readable stream#readable-streamReferenced in:1. Introduction2.1. Readable Streams2.6. Locking3.2. Class ReadableStream3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})3.2.4.4. pipeThrough({ writable, readable }, options)3.2.4.5. pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})4.1. Using Writable Streams6.1. Class ByteLengthQueuingStrategy6.2. Class CountQueuingStrategy8.1. A readable stream with an underlying push source (no backpressure support)8.2. A readable stream with an underlying push source and backpressure support8.4. A readable stream with an underlying pull source represents a source of data, from which you can read. In other words, data comes out of a readable stream.

Although a readable stream can be created with arbitrary behavior, most readable streams wrap a lower-level I/O source, called the underlying source#underlying-sourceReferenced in:2.1. Readable Streams (2) (3)2.4. Pipe Chains and Backpressure (2)2.5. Internal Queues and Queuing Strategies2.6. Locking3.2. Class ReadableStream3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {}) (2) (3)3.2.4.6. tee()3.4. Readable Stream Abstract Operations Used by Controllers3.5.4.3. read()3.8. Class ReadableStreamDefaultController3.8.2. Internal Slots (2) (3) (4) (5)3.8.4.1. get desiredSize3.10.2. Internal Slots3.10.4.2. get desiredSize6.1. Class ByteLengthQueuingStrategy6.2. Class CountQueuingStrategy8.2. A readable stream with an underlying push source and backpressure support. There are two types of underlying source: push sources and pull sources.

Push sources#push-sourceReferenced in:3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})8.1. A readable stream with an underlying push source (no backpressure support) push data at you, whether or not you are listening for it. They may also provide a mechanism for pausing and resuming the flow of data. An example push source is a TCP socket, where data is constantly being pushed from the OS level, at a rate that can be controlled by changing the TCP window size.

Pull sources#pull-sourceReferenced in:3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})8.4. A readable stream with an underlying pull source require you to request data from them. The data may be available synchronously, e.g. if it is held by the operating system’s in-memory buffers, or asynchronously, e.g. if it has to be read from disk. An example pull source is a file handle, where you seek to specific locations and read specific amounts.

Readable streams are designed to wrap both types of sources behind a single, unified interface.

Chunks are enqueued into the stream by the stream’s underlying source. They can then be read one at a time via the stream’s public interface.

Code that reads from a readable stream using its public interface is known as a consumer#consumerReferenced in:2.6. Locking3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {}) (2)3.5.2. Internal Slots3.6.2. Internal Slots3.8.4.2. close()3.10.4.3. close().

Consumers also have the ability to cancel#cancel-a-readable-streamReferenced in:2.6. Locking (2)3.2.4.2. cancel(reason)3.2.4.3. getReader(options = {})3.2.4.6. tee() a readable stream. This indicates that the consumer has lost interest in the stream, and will immediately close the stream, throw away any queued chunks, and execute any cancellation mechanism of the underlying source.

Consumers can also tee#tee-a-readable-streamReferenced in:2.4. Pipe Chains and Backpressure3.1. Using Readable Streams3.2.4.6. tee()3.3.6. ReadableStreamTee ( stream, shouldClone ) a readable stream. This will lock the stream, making it no longer directly usable; however, it will create two new streams, called branches, which can be consumed independently.

For streams representing bytes, an extended version of the readable stream is provided to handle bytes efficiently, in particular by minimizing copies. The underlying source for such a readable stream is called a underlying byte source#underlying-byte-sourceReferenced in:3.4. Readable Stream Abstract Operations Used by Controllers3.6.4.3. read(view)3.10. Class ReadableByteStreamController3.10.2. Internal Slots (2) (3) (4) (5)8.5. A readable byte stream with an underlying pull source. A readable stream whose underlying source is an underlying byte source is sometimes called a readable byte stream#readable-byte-streamReferenced in:2.6. Locking3.1. Using Readable Streams3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {}) (2)3.2.4.3. getReader(options = {})3.4. Readable Stream Abstract Operations Used by Controllers3.8. Class ReadableStreamDefaultController8.3. A readable byte stream with an underlying push source and backpressure support8.5. A readable byte stream with an underlying pull source.

2.2. Writable Streams

A writable stream#writable-streamReferenced in:1. Introduction3.1. Using Readable Streams (2)3.2.4.5. pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})6.1. Class ByteLengthQueuingStrategy6.2. Class CountQueuingStrategy8.6. A writable stream with no backpressure or success signals8.7. A writable stream with backpressure and success signals represents a destination for data, into which you can write. In other words, data goes in to a writable stream.

Analogously to readable streams, most writable streams wrap a lower-level I/O sink, called the underlying sink#underlying-sinkReferenced in:2.2. Writable Streams (2)2.4. Pipe Chains and Backpressure2.5. Internal Queues and Queuing Strategies4.1. Using Writable Streams (2) (3)4.2.2. Internal Slots (2) (3) (4)4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {}) (2) (3) (4) (5)4.2.4.3. get state (2) (3)4.2.4.4. abort(reason)4.2.4.5. close()4.2.4.6. write(chunk)6.1. Class ByteLengthQueuingStrategy6.2. Class CountQueuingStrategy8.7. A writable stream with backpressure and success signals. Writable streams work to abstract away some of the complexity of the underlying sink, by queuing subsequent writes and only delivering them to the underlying sink one by one.

Chunks are written to the stream via its public interface, and are passed one at a time to the stream’s underlying sink.

Code that writes into a writable stream using its public interface is known as a producer#producerReferenced in:4.2.2. Internal Slots4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {})6.1. Class ByteLengthQueuingStrategy6.2. Class CountQueuingStrategy8.6. A writable stream with no backpressure or success signals8.7. A writable stream with backpressure and success signals.

Producers also have the ability to abort a writable stream. This indicates that the producer believes something has gone wrong, and that future writes should be discontinued. It puts the stream in an errored state, even without a signal from the underlying sink.

2.3. Transform Streams

A transform stream#transform-streamReferenced in:1. Introduction3.2.4.4. pipeThrough({ writable, readable }, options) consists of a pair of streams: a writable stream, and a readable stream. In a manner specific to the transform stream in question, writes to the writable side result in new data being made available for reading from the readable side.

Some examples of transform streams include:

• A GZIP compressor, to which uncompressed bytes are written and from which compressed bytes are read;
• A video decoder, to which encoded bytes are written and from which uncompressed video frames are read;
• A text decoder, to which bytes are written and from which strings are read;
• A CSV-to-JSON converter, to which strings representing lines of a CSV file are written and from which corresponding JavaScript objects are read.

2.4. Pipe Chains and Backpressure

Streams are primarily used by piping#pipingReferenced in:2.6. Locking3.1. Using Readable Streams (2) (3)3.2.4.4. pipeThrough({ writable, readable }, options)3.2.4.5. pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})4.1. Using Writable Streams them to each other. A readable stream can be piped directly to a writable stream, or it can be piped through one or more transform streams first.

A set of streams piped together in this way is referred to as a pipe chain#pipe-chainReferenced in:3.2.4.4. pipeThrough({ writable, readable }, options). In a pipe chain, the original source is the underlying source of the first readable stream in the chain; the ultimate sink is the underlying sink of the final writable stream in the chain.

Once a pipe chain is constructed, it can be used to propagate signals regarding how fast chunks should flow through it. If any step in the chain cannot yet accept chunks, it propagates a signal backwards through the pipe chain, until eventually the original source is told to stop producing chunks so fast. This process of normalizing flow from the original source according to how fast the chain can process chunks is called backpressure#backpressureReferenced in:1. Introduction2.4. Pipe Chains and Backpressure2.5. Internal Queues and Queuing Strategies3.1. Using Readable Streams (2)3.8.2. Internal Slots3.8.4.1. get desiredSize3.10.2. Internal Slots3.10.4.2. get desiredSize4.1. Using Writable Streams4.2.2. Internal Slots4.2.4.3. get state6.1. Class ByteLengthQueuingStrategy (2)6.2. Class CountQueuingStrategy (2)8.2. A readable stream with an underlying push source and backpressure support8.3. A readable byte stream with an underlying push source and backpressure support8.6. A writable stream with no backpressure or success signals8.7. A writable stream with backpressure and success signals.

When teeing a readable stream, the backpressure signals from its two branches will aggregate, such that if neither branch is read from, a backpressure signal will be sent to the underlying source of the original stream.

2.5. Internal Queues and Queuing Strategies

Both readable and writable streams maintain internal queues#internal-queuesReferenced in:2.5. Internal Queues and Queuing Strategies3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})3.4. Readable Stream Abstract Operations Used by Controllers3.8. Class ReadableStreamDefaultController3.10. Class ReadableByteStreamController8.3. A readable byte stream with an underlying push source and backpressure support, which they use for similar purposes. In the case of a readable stream, the internal queue contains chunks that have been enqueued by the underlying source, but not yet read by the consumer. In the case of a writable stream, the internal queue contains chunks which have been written to the stream by the producer, but not yet processed and acknowledged by the underlying sink.

A queuing strategy#queuing-strategyReferenced in:3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})3.8.2. Internal Slots (2) (3)3.10.2. Internal Slots (2) (3)4.2.2. Internal Slots (2)4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {})4.2.4.6. write(chunk)6.1. Class ByteLengthQueuingStrategy (2)6.2. Class CountQueuingStrategy is an object that determines how a stream should signal backpressure based on the state of its internal queue. The queuing strategy assigns a size to each chunk, and compares the total size of all chunks in the queue to a specified number, known as the high water mark#high-water-markReferenced in:3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {}) (2)4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {}). The resulting difference, high water mark minus total size, is used to determine the desired size to fill the stream’s queue#desired-size-to-fill-a-streams-internal-queueReferenced in:3.8.4.1. get desiredSize3.9.7. ReadableStreamDefaultControllerGetDesiredSize ( controller )3.10.4.2. get desiredSize.

For readable streams, an underlying source can use this desired size as a backpressure signal, slowing down chunk generation so as to try to keep the desired size above or at zero. For writable streams, a producer can behave similarly, avoiding writes that would cause the desired size to go negative.

2.6. Locking

A readable stream reader#readable-stream-readerReferenced in:3.1. Using Readable Streams3.3.6. ReadableStreamTee ( stream, shouldClone ), or simply reader, is an object that allows direct reading of chunks from a readable stream. Without a reader, a consumer can only perform high-level operations on the readable stream: waiting for the stream to become closed or errored, canceling the stream, or piping the readable stream to a writable stream. Many of those high-level operations actually use a reader themselves.

A given readable stream only has at most one reader at a time. We say in this case the stream is locked to the reader#locked-to-a-readerReferenced in:2.1. Readable Streams3.2.2. Internal Slots3.2.4.1. get locked3.2.4.3. getReader(options = {})3.2.4.4. pipeThrough({ writable, readable }, options)3.2.4.5. pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})3.2.4.6. tee()3.3.5. IsReadableStreamLocked ( stream ), and that the reader is active#active-readerReferenced in:3.5.4.2. cancel(reason)3.5.4.4. releaseLock()3.6.4.2. cancel(reason)3.6.4.4. releaseLock().

A reader also has the capability to release its read lock#release-a-read-lockReferenced in:3.2.4.3. getReader(options = {})3.5.4.1. get closed3.5.4.4. releaseLock()3.6.4.1. get closed3.6.4.2. cancel(reason)3.6.4.4. releaseLock(), which makes it no longer active. At this point another reader can be acquired at will. If the stream becomes closed or errored as a result of the behavior of its underlying source or via cancellation, its reader (if one exists) will automatically release its lock.

A readable byte stream has the ability to vend two types of readers: default readers#default-readersReferenced in:3.1. Using Readable Streams3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})3.2.4.3. getReader(options = {})3.3.2. AcquireReadableStreamDefaultReader ( stream )3.5. Class ReadableStreamDefaultReader8.5. A readable byte stream with an underlying pull source (2) and BYOB readers#byob-readersReferenced in:3.1. Using Readable Streams3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {}) (2)3.2.4.3. getReader(options = {})3.3.1. AcquireReadableStreamBYOBReader ( stream )3.6. Class ReadableStreamBYOBReader8.3. A readable byte stream with an underlying push source and backpressure support (2)8.5. A readable byte stream with an underlying pull source. BYOB ("bring your own buffer") readers allow reading into a developer-supplied buffer, thus minimizing copies.

3. Readable Streams

3.1. Using Readable Streams

readableStream.pipeTo(writableStream)
  .then(() => console.log("All data successfully written!"))
  .catch(e => console.error("Something went wrong!", e));

readableStream.pipeTo(new WritableStream({
  write(chunk) {
    console.log("Chunk received", chunk);
  },
  close() {
    console.log("All data successfully read!");
  },
  abort(e) {
    console.error("Something went wrong!", e);
  }
}));

const reader = readableStream.getReader();

reader.read().then(
  ({ value, done }) => {
    if (done) {
      console.log("The stream was already closed!");
    } else {
      console.log(value);
    }
  },
  e => console.error("The stream became errored and cannot be read from!", e)
);

const reader = readableStream.getReader({ mode: "byob" });

let startingAB = new ArrayBuffer(1024);
readInto(startingAB)
  .then(buffer => console.log("The first 1024 bytes:", buffer))
  .catch(e => console.error("Something went wrong!", e));

function readInto(buffer, offset = 0) {
  if (offset === buffer.byteLength) {
    return Promise.resolve(buffer);
  }

  const view = new Uint8Array(buffer, offset, buffer.byteLength - offset);
  return reader.read(view).then(newView => {
    return readInto(newView.buffer, offset + newView.byteLength);
  });
}

3.2. Class ReadableStream#rs-classReferenced in:3.2. Class ReadableStream3.2.1. Class Definition3.2.2. Internal Slots3.2.4. Properties of the ReadableStream Prototype3.2.4.4. pipeThrough({ writable, readable }, options) (2)3.2.4.6. tee()3.3.6. ReadableStreamTee ( stream, shouldClone ) (2)3.4. Readable Stream Abstract Operations Used by Controllers (2) (3)3.5. Class ReadableStreamDefaultReader3.5.2. Internal Slots3.6. Class ReadableStreamBYOBReader3.6.2. Internal Slots3.8. Class ReadableStreamDefaultController (2)3.8.2. Internal Slots3.8.3. new ReadableStreamDefaultController(stream, underlyingSource, size, highWaterMark)3.10. Class ReadableByteStreamController (2)3.10.2. Internal Slots3.10.3. new ReadableByteStreamController(stream, underlyingByteSource, highWaterMark)7. Global Properties (2) (3)8. Examples of Creating Streams8.3. A readable byte stream with an underlying push source and backpressure support8.5. A readable byte stream with an underlying pull source

The ReadableStream class is a concrete instance of the general readable stream concept. It is adaptable to any chunk type, and maintains an internal queue to keep track of data supplied by the underlying source but not yet read by any consumer.

3.2.1. Class Definition

This section is non-normative.

If one were to write the ReadableStream class in something close to the syntax of [ECMASCRIPT], it would look like

class ReadableStream {
  constructor(underlyingSource = {}, { size, highWaterMark = 1 } = {})

  get locked()

  cancel(reason)
  getReader()
  pipeThrough({ writable, readable }, options)
  pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})
  tee()
}

3.2.2. Internal Slots

Instances of ReadableStream are created with the internal slots described in the following table:

3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})

3.2.4. Properties of the ReadableStream Prototype

3.2.4.1. get locked

3.2.4.2. cancel(reason)#rs-cancelReferenced in:3.4.4. ReadableStreamClose ( stream )3.9.5. ReadableStreamDefaultControllerEnqueue ( controller, chunk )

3.2.4.3. getReader(options = {})#rs-get-readerReferenced in:3.5.3. new ReadableStreamDefaultReader(stream)3.6.3. new ReadableStreamBYOBReader(stream)7. Global Properties

function readAllChunks(readableStream) {
  const reader = readableStream.getReader();
  const chunks = [];

  return pump();

  function pump() {
    return reader.read().then(({ value, done })=> {
      if (done) {
        return chunks;
      }

      chunks.push(value);
      return pump();
    });
  }
}

3.2.4.4. pipeThrough({ writable, readable }, options)#rs-pipe-throughReferenced in:3.2.4.4. pipeThrough({ writable, readable }, options)

httpResponseBody
  .pipeThrough(decompressorTransform)
  .pipeThrough(ignoreNonImageFilesTransform)
  .pipeTo(mediaGallery);

3.2.4.5. pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})

The pipeTo method is still in some flux. Its design depends on the design of writable streams, which are still undergoing spec churn.

For now, the reference implementation and tests provide a guide to what this method is generally intended to do: reference-implementation/lib/readable-stream.js, look for the pipeTo method. In addition to changing as the writable stream design changes, one major aspect of pipeTo not captured by the reference implementation is that it will operate via unobservable abstract operation calls, instead of using the JavaScript-exposed readable and writable stream APIs. This will better allow optimization and specialization. See #407 and #97 for more information.

3.2.4.6. tee()

const [forLocal, forRemote] = readableStream.tee();

Promise.all([
  forLocal.pipeTo(cacheEntry),
  forRemote.pipeTo(httpRequestBody)
])
.then(() => console.log("Saved the stream to the cache and also uploaded it!"))
.catch(e => console.error("Either caching or uploading failed: ", e));

3.3. General Readable Stream Abstract Operations

The following abstract operations, unlike most in this specification, are meant to be generally useful by other specifications, instead of just being part of the implementation of this spec’s classes.

3.3.1. AcquireReadableStreamBYOBReader ( stream ) throws

This abstract operation is meant to be called from other specifications that may wish to acquire a BYOB reader for a given stream.

3.3.2. AcquireReadableStreamDefaultReader ( stream ) throws

This abstract operation is meant to be called from other specifications that may wish to acquire a default reader for a given stream.

3.3.3. IsReadableStream ( x ) nothrow

3.3.4. IsReadableStreamDisturbed ( stream ) nothrow

This abstract operation is meant to be called from other specifications that may wish to query whether or not a readable stream has ever been read from or canceled.

3.3.5. IsReadableStreamLocked ( stream ) nothrow

This abstract operation is meant to be called from other specifications that may wish to query whether or not a readable stream is locked to a reader.

3.3.6. ReadableStreamTee ( stream, shouldClone ) throws

This abstract operation is meant to be called from other specifications that may wish to tee a given readable stream. Its second argument governs whether or not the data from the original stream will be structured cloned before becoming visible in the returned branches. [HTML]

A ReadableStreamTee pull function#readablestreamtee-pull-functionReferenced in:3.3.6. ReadableStreamTee ( stream, shouldClone ) is an anonymous built-in function that pulls data from a given readable stream reader and enqueues it into two other streams ("branches" of the associated tee). Each ReadableStreamTee pull function has [[reader]], [[branch1]], [[branch2]], [[teeState]], and [[shouldClone]] internal slots. When a ReadableStreamTee pull function F is called, it performs the following steps:

A ReadableStreamTee branch 1 cancel function#readablestreamtee-branch-1-cancel-functionReferenced in:3.3.6. ReadableStreamTee ( stream, shouldClone ) is an anonymous built-in function that reacts to the cancellation of the first of the two branches of the associated tee. Each ReadableStreamTee branch 1 cancel function has [[stream]] and [[teeState]] internal slots. When a ReadableStreamTee branch 1 cancel function F is called with argument reason, it performs the following steps:

A ReadableStreamTee branch 2 cancel function#readablestreamtee-branch-2-cancel-functionReferenced in:3.3.6. ReadableStreamTee ( stream, shouldClone ) is an anonymous built-in function that reacts to the cancellation of the second of the two branches of the associated tee. Each ReadableStreamTee branch 2 cancel function has [[stream]] and [[teeState]] internal slots. When a ReadableStreamTee branch 2 cancel function F is called with argument reason, it performs the following steps:

3.4. Readable Stream Abstract Operations Used by Controllers

In terms of specification factoring, the way that the ReadableStream class encapsulates the behavior of both simple readable streams and readable byte streams into a single class is by centralizing most of the potentially-varying logic inside the two controller classes, ReadableStreamDefaultController and ReadableByteStreamController. Those classes define most of the stateful internal slots and abstract operations for how a stream’s internal queue is managed and how it interfaces with its underlying source or underlying byte source.

The abstract operations in this section are interfaces that are used by the controller implementations to affect their associated ReadableStream object, translating those internal state changes into developer-facing results visible through the ReadableStream's public API.

3.4.1. ReadableStreamAddReadIntoRequest ( stream ) nothrow

3.4.2. ReadableStreamAddReadRequest ( stream ) nothrow

3.4.3. ReadableStreamCancel ( stream, reason ) nothrow

3.4.4. ReadableStreamClose ( stream ) nothrow

This abstract operation can be called by other specifications that wish to close a readable stream, in the same way a developer-created stream would be closed by its associated controller object. Specifications should not do this to streams they did not create, and must ensure they have obeyed the preconditions (listed here as asserts).

3.4.5. ReadableStreamError ( stream, e ) nothrow

This abstract operation can be called by other specifications that wish to move a readable stream to an errored state, in the same way a developer would error a stream using its associated controller object. Specifications should not do this to streams they did not create, and must ensure they have obeyed the precondition (listed here as an assert).

3.4.6. ReadableStreamFulfillReadIntoRequest ( stream, chunk, done ) nothrow

3.4.7. ReadableStreamFulfillReadRequest ( stream, chunk, done ) nothrow

3.4.8. ReadableStreamGetNumReadIntoRequests ( stream ) nothrow

3.4.9. ReadableStreamGetNumReadRequests ( stream ) nothrow

3.4.10. ReadableStreamHasBYOBReader ( stream ) nothrow

3.4.11. ReadableStreamHasReader ( stream ) nothrow

3.5. Class ReadableStreamDefaultReader#default-reader-classReferenced in:3.2.2. Internal Slots3.2.4.3. getReader(options = {})3.3.2. AcquireReadableStreamDefaultReader ( stream )3.5. Class ReadableStreamDefaultReader3.5.1. Class Definition3.5.2. Internal Slots3.5.4. Properties of the ReadableStreamDefaultReader Prototype7. Global Properties

The ReadableStreamDefaultReader class represents a default reader designed to be vended by a ReadableStream instance.

3.5.1. Class Definition

This section is non-normative.

If one were to write the ReadableStreamDefaultReader class in something close to the syntax of [ECMASCRIPT], it would look like

class ReadableStreamDefaultReader {
  constructor(stream)

  get closed()

  cancel(reason)
  read()
  releaseLock()
}

3.5.2. Internal Slots

Instances of ReadableStreamDefaultReader are created with the internal slots described in the following table:

3.5.3. new ReadableStreamDefaultReader(stream)

3.5.4. Properties of the ReadableStreamDefaultReader Prototype

3.5.4.1. get closed#default-reader-closedReferenced in:3.5.2. Internal Slots

3.5.4.2. cancel(reason)

3.5.4.3. read()#default-reader-readReferenced in:3.2.4.3. getReader(options = {})3.5.2. Internal Slots3.5.4.4. releaseLock()

3.5.4.4. releaseLock()

3.6. Class ReadableStreamBYOBReader#byob-reader-classReferenced in:3.2.2. Internal Slots3.2.4.3. getReader(options = {})3.3.1. AcquireReadableStreamBYOBReader ( stream )3.6. Class ReadableStreamBYOBReader3.6.1. Class Definition3.6.2. Internal Slots3.6.4. Properties of the ReadableStreamBYOBReader Prototype7. Global Properties

The ReadableStreamBYOBReader class represents a BYOB reader designed to be vended by a ReadableStream instance.

3.6.1. Class Definition

This section is non-normative.

If one were to write the ReadableStreamBYOBReader class in something close to the syntax of [ECMASCRIPT], it would look like

class ReadableStreamBYOBReader {
  constructor(stream)

  get closed()

  cancel(reason)
  read(view)
  releaseLock()
}

3.6.2. Internal Slots

Instances of ReadableStreamBYOBReader are created with the internal slots described in the following table:

3.6.3. new ReadableStreamBYOBReader(stream)

3.6.4. Properties of the ReadableStreamBYOBReader Prototype

3.6.4.1. get closed#byob-reader-closedReferenced in:3.6.2. Internal Slots

3.6.4.2. cancel(reason)

3.6.4.3. read(view)#byob-reader-readReferenced in:3.2.4.3. getReader(options = {})3.6.2. Internal Slots3.6.4.4. releaseLock()

3.6.4.4. releaseLock()

3.7. Readable Stream Reader Abstract Operations

3.7.1. IsReadableStreamDefaultReader ( x ) nothrow

3.7.2. IsReadableStreamBYOBReader ( x ) nothrow

3.7.3. ReadableStreamReaderGenericCancel ( reader, reason ) throws

3.7.4. ReadableStreamReaderGenericInitialize ( reader, stream ) nothrow

3.7.5. ReadableStreamReaderGenericRelease ( reader ) throws

3.7.6. ReadableStreamBYOBReaderRead ( reader, view ) nothrow

3.7.7. ReadableStreamDefaultReaderRead ( reader ) nothrow

3.8. Class ReadableStreamDefaultController#rs-default-controller-classReferenced in:3.2.2. Internal Slots3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {}) (2)3.4. Readable Stream Abstract Operations Used by Controllers3.8. Class ReadableStreamDefaultController (2)3.8.1. Class Definition3.8.2. Internal Slots3.8.4. Properties of the ReadableStreamDefaultController Prototype3.8.5. Readable Stream Default Controller Internal Methods7. Global Properties

The ReadableStreamDefaultController class has methods that allow control of a ReadableStream's state and internal queue. When constructing a ReadableStream that is not a readable byte stream, the underlying source is given a corresponding ReadableStreamDefaultController instance to manipulate.

3.8.1. Class Definition

This section is non-normative.

If one were to write the ReadableStreamDefaultController class in something close to the syntax of [ECMASCRIPT], it would look like

class ReadableStreamDefaultController {
  constructor(stream, underlyingSource, size, highWaterMark)

  get desiredSize()

  close()
  enqueue(chunk)
  error(e)
}

3.8.2. Internal Slots

Instances of ReadableStreamDefaultController are created with the internal slots described in the following table:

3.8.3. new ReadableStreamDefaultController(stream, underlyingSource, size, highWaterMark)

3.8.4. Properties of the ReadableStreamDefaultController Prototype

3.8.4.1. get desiredSize#rs-default-controller-desired-sizeReferenced in:3.9.7. ReadableStreamDefaultControllerGetDesiredSize ( controller )

3.8.4.2. close()

3.8.4.3. enqueue(chunk)

3.8.4.4. error(e)

3.8.5. Readable Stream Default Controller Internal Methods

The following are additional internal methods implemented by each ReadableStreamDefaultController instance. They are similar to the supporting abstract operations in the following section, but are in method form to allow polymorphic dispatch from the readable stream implementation to either these or their counterparts for BYOB controllers.

3.8.5.1. [[Cancel]](reason)

3.8.5.2. [[Pull]]()

3.9. Readable Stream Default Controller Abstract Operations

3.9.1. IsReadableStreamDefaultController ( x ) nothrow

3.9.2. ReadableStreamDefaultControllerCallPullIfNeeded ( controller ) nothrow

3.9.3. ReadableStreamDefaultControllerShouldCallPull ( controller ) nothrow

3.9.4. ReadableStreamDefaultControllerClose ( controller ) nothrow

3.9.5. ReadableStreamDefaultControllerEnqueue ( controller, chunk ) nothrow

This abstract operation can be called by other specifications that wish to enqueue chunks in a readable stream, in the same way a developer would enqueue chunks using the stream’s associated controller object. Specifications should not do this to streams they did not create, and must ensure they have obeyed the preconditions (listed here as asserts).

3.9.6. ReadableStreamDefaultControllerError ( controller, e ) nothrow

3.9.7. ReadableStreamDefaultControllerGetDesiredSize ( controller ) nothrow

This abstract operation can be called by other specifications that wish to determine the desired size to fill this stream’s internal queue, similar to how a developer would consult the desiredSize property of the stream’s associated controller object. Specifications should not use this on streams they did not create.

3.10. Class ReadableByteStreamController#rbs-controller-classReferenced in:3.2.2. Internal Slots3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {}) (2)3.4. Readable Stream Abstract Operations Used by Controllers3.10. Class ReadableByteStreamController (2)3.10.1. Class Definition3.10.2. Internal Slots3.10.4. Properties of the ReadableByteStreamController Prototype3.10.5. Readable Stream BYOB Controller Internal Methods3.11. Class ReadableStreamBYOBRequest3.11.2. Internal Slots7. Global Properties

The ReadableByteStreamController class has methods that allow control of a ReadableStream's state and internal queue. When constructing a ReadableStream, the underlying byte source is given a corresponding ReadableByteStreamController instance to manipulate.

3.10.1. Class Definition

This section is non-normative.

If one were to write the ReadableByteStreamController class in something close to the syntax of [ECMASCRIPT], it would look like

class ReadableByteStreamController {
  constructor(stream, underlyingByteSource, highWaterMark)

  get byobRequest()
  get desiredSize()

  close()
  enqueue(chunk)
  error(e)
}

3.10.2. Internal Slots

Instances of ReadableByteStreamController are created with the internal slots described in the following table:

3.10.3. new ReadableByteStreamController(stream, underlyingByteSource, highWaterMark)

3.10.4. Properties of the ReadableByteStreamController Prototype

3.10.4.1. get byobRequest

3.10.4.2. get desiredSize

3.10.4.3. close()

3.10.4.4. enqueue(chunk)

3.10.4.5. error(e)

3.10.5. Readable Stream BYOB Controller Internal Methods

The following are additional internal methods implemented by each ReadableByteStreamController instance. They are similar to the supporting abstract operations in the following section, but are in method form to allow polymorphic dispatch from the readable stream implementation to either these or their counterparts for default controllers.

3.10.5.1. [[Cancel]](reason)

3.10.5.2. [[Pull]]()

3.11. Class ReadableStreamBYOBRequest#rs-byob-request-classReferenced in:3.10.2. Internal Slots3.10.4.1. get byobRequest3.11. Class ReadableStreamBYOBRequest3.11.1. Class Definition3.11.2. Internal Slots3.11.4. Properties of the ReadableStreamBYOBRequest Prototype7. Global Properties

The ReadableStreamBYOBRequest class represents a pull into request in a ReadableByteStreamController.

3.11.1. Class Definition

This section is non-normative.

If one were to write the ReadableStreamBYOBRequest class in something close to the syntax of [ECMASCRIPT], it would look like

class ReadableStreamBYOBRequest {
  constructor(controller, view)

  get view()

  respond(bytesWritten)
  respondWithNewView(view)
}

3.11.2. Internal Slots

Instances of ReadableStreamBYOBRequest are created with the internal slots described in the following table:

3.11.3. new ReadableStreamBYOBRequest(controller, view)

3.11.4. Properties of the ReadableStreamBYOBRequest Prototype

3.11.4.1. get view

3.11.4.2. respond(bytesWritten)

3.11.4.3. respondWithNewView(view)

3.12. Readable Stream BYOB Controller Abstract Operations

3.12.1. IsReadableStreamBYOBRequest ( x ) nothrow

3.12.2. IsReadableByteStreamController ( x ) nothrow

3.12.3. ReadableByteStreamControllerCallPullIfNeeded ( controller ) nothrow

3.12.4. ReadableByteStreamControllerClearPendingPullIntos ( controller ) nothrow

3.12.5. ReadableByteStreamControllerClose ( controller ) nothrow

3.12.6. ReadableByteStreamControllerCommitPullIntoDescriptor ( stream, pullIntoDescriptor ) nothrow

3.12.7. ReadableByteStreamControllerConvertPullIntoDescriptor ( pullIntoDescriptor ) nothrow

3.12.8. ReadableByteStreamControllerEnqueue ( controller, chunk ) throws

3.12.9. ReadableByteStreamControllerEnqueueChunkToQueue ( controller, buffer, byteOffset, byteLength ) nothrow

3.12.10. ReadableByteStreamControllerError ( controller, e ) nothrow

3.12.11. ReadableByteStreamControllerFillHeadPullIntoDescriptor ( controller, size, pullIntoDescriptor ) nothrow

3.12.12. ReadableByteStreamControllerFillPullIntoDescriptorFromQueue ( controller, pullIntoDescriptor ) nothrow

3.12.13. ReadableByteStreamControllerGetDesiredSize ( controller ) nothrow

3.12.14. ReadableByteStreamControllerHandleQueueDrain ( controller ) nothrow

3.12.15. ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue ( controller ) nothrow

3.12.16. ReadableByteStreamControllerPullInto ( controller, view ) nothrow

3.12.17. ReadableByteStreamControllerRespond ( controller, bytesWritten ) throws

3.12.18. ReadableByteStreamControllerRespondInClosedState ( controller, firstDescriptor ) nothrow

3.12.19. ReadableByteStreamControllerRespondInReadableState ( controller, bytesWritten, pullIntoDescriptor ) nothrow

3.12.20. ReadableByteStreamControllerRespondInternal ( controller, bytesWritten ) throws

3.12.21. ReadableByteStreamControllerRespondWithNewView ( controller, view ) throws

3.12.22. ReadableByteStreamControllerShiftPendingPullInto ( controller ) nothrow

3.12.23. ReadableByteStreamControllerShouldCallPull ( controller ) nothrow

3.12.24. ReadableStreamBYOBRequestInvalidate ( request ) nothrow

4. Writable Streams

4.1. Using Writable Streams

readableStream.pipeTo(writableStream)
  .then(() => console.log("All data successfully written!"))
  .catch(e => console.error("Something went wrong!", e));

function writeArrayToStream(array, writableStream) {
  array.forEach(chunk => writableStream.write(chunk));

  return writableStream.close();
}

writeArrayToStream([1, 2, 3, 4, 5], writableStream)
  .then(() => console.log("All done!"))
  .catch(e => console.error("Error with the stream: " + e));

writableStream.write("i am a chunk of data")
  .then(() => console.log("chunk successfully written!"))
  .catch(e => console.error(e));

4.2. Class WritableStream#ws-classReferenced in:3.2.4.4. pipeThrough({ writable, readable }, options)4.2.1. Class Definition4.2.2. Internal Slots4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {}) (2) (3)4.2.4. Properties of the WritableStream Prototype7. Global Properties8. Examples of Creating Streams

4.2.1. Class Definition

This section is non-normative.

If one were to write the WritableStream class in something close to the syntax of [ECMASCRIPT], it would look like

class WritableStream {
  constructor(underlyingSink = {}, { size, highWaterMark = 0 } = {})

  get closed()
  get ready()
  get state()

  abort(reason)
  close()
  write(chunk)
}

4.2.2. Internal Slots

Instances of WritableStream are created with the internal slots described in the following table:

4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {})

A WritableStream error function#writablestream-error-functionReferenced in:4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {}) is an anonymous built-in function that is used to allow underlying sinks to error their associated writable stream. Each WritableStream error function has a [[stream]] internal slot. When a WritableStream error function F is called with argument e, it performs the following steps:

4.2.4. Properties of the WritableStream Prototype

4.2.4.1. get closed#ws-closedReferenced in:4.1. Using Writable Streams4.2.2. Internal Slots

4.2.4.2. get ready#ws-readyReferenced in:4.2.2. Internal Slots4.2.4.3. get state

4.2.4.3. get state#ws-stateReferenced in:4.2.2. Internal Slots4.2.4.6. write(chunk)

4.2.4.4. abort(reason)

4.2.4.5. close()#ws-closeReferenced in:4.1. Using Writable Streams (2)4.2.4.3. get state

4.2.4.6. write(chunk)#ws-writeReferenced in:4.1. Using Writable Streams (2)4.2.4.3. get state

4.3. Writable Stream Abstract Operations

4.3.1. CallOrScheduleWritableStreamAdvanceQueue ( stream ) nothrow

4.3.2. CloseWritableStream ( stream ) nothrow

4.3.3. ErrorWritableStream ( stream, e ) nothrow

4.3.4. IsWritableStream ( x ) nothrow

4.3.5. SyncWritableStreamStateWithQueue ( stream ) nothrow

4.3.6. WritableStreamAdvanceQueue ( stream ) nothrow

5. Transform Streams

Transform streams have been developed in the testable implementation, but not yet re-encoded in spec language. We are waiting to validate their design before doing so. In the meantime, see reference-implementation/lib/transform-stream.js.

6. Other Stream APIs and Operations

6.1. Class ByteLengthQueuingStrategy#blqs-classReferenced in:3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {})6.1.1. Class Definition (2)6.1.3. Properties of the ByteLengthQueuingStrategy Prototype7. Global Properties

A common queuing strategy when dealing with bytes is to wait until the accumulated byteLength properties of the incoming chunks reaches a specified high-water mark. As such, this is provided as a built-in queuing strategy that can be used when constructing streams.

const stream = new ReadableStream(
  { ... },
  new ByteLengthQueuingStrategy({ highWaterMark: 16 * 1024 })
);

const stream = new WritableStream(
  { ... },
  new ByteLengthQueuingStrategy({ highWaterMark: 32 * 1024 })
);

6.1.1. Class Definition

This section is non-normative.

If one were to write the ByteLengthQueuingStrategy class in something close to the syntax of [ECMASCRIPT], it would look like

class ByteLengthQueuingStrategy {
  constructor({ highWaterMark })
  size(chunk)
}

Each ByteLengthQueuingStrategy instance will additionally have an own data property highWaterMark set by its constructor.

6.1.2. new ByteLengthQueuingStrategy({ highWaterMark })

6.1.3. Properties of the ByteLengthQueuingStrategy Prototype

6.1.3.1. size(chunk)

6.2. Class CountQueuingStrategy#cqs-classReferenced in:3.2.3. new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {}) (2)4.2.3. new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {}) (2)6.2.1. Class Definition (2)6.2.3. Properties of the CountQueuingStrategy Prototype7. Global Properties

A common queuing strategy when dealing with streams of generic objects is to simply count the number of chunks that have been accumulated so far, waiting until this number reaches a specified high-water mark. As such, this strategy is also provided out of the box.

const stream = new ReadableStream(
  { ... },
  new CountQueuingStrategy({ highWaterMark: 10 })
);

const stream = new WritableStream(
  { ... },
  new CountQueuingStrategy({ highWaterMark: 5 })
);

6.2.1. Class Definition

This section is non-normative.

If one were to write the CountQueuingStrategy class in something close to the syntax of [ECMASCRIPT], it would look like

class CountQueuingStrategy {
  constructor({ highWaterMark })
  size()
}

Each CountQueuingStrategy instance will additionally have an own data property highWaterMark set by its constructor.

6.2.2. new CountQueuingStrategy({ highWaterMark })

6.2.3. Properties of the CountQueuingStrategy Prototype

6.2.3.1. size()

6.3. Queue-with-Sizes Operations

The streams in this specification use a "queue-with-sizes" data structure to store queued up values, along with their determined sizes. A queue-with-sizes is a List of Records with [[value]] and [[size]] fields (although in implementations it would of course be backed by a more efficient data structure).

A number of abstract operations are specified here to make working with queues-with-sizes more pleasant, and used throughout the rest of this standard.

6.3.1. DequeueValue ( queue ) nothrow

6.3.2. EnqueueValueWithSize ( queue, value, size ) throws

6.3.3. GetTotalQueueSize ( queue ) nothrow

6.3.4. PeekQueueValue ( queue ) nothrow

6.4. Miscellaneous Operations

A few abstract operations are used in this specification for utility purposes. We define them here.

6.4.1. IsFiniteNonNegativeNumber ( v ) nothrow

6.4.2. InvokeOrNoop ( O, P, args ) throws

6.4.3. PromiseInvokeOrFallbackOrNoop ( O, P1, args1, P2, args2 ) nothrow

6.4.4. PromiseInvokeOrNoop ( O, P, args ) nothrow

6.4.5. ValidateAndNormalizeHighWaterMark ( highWaterMark ) throws

6.4.6. ValidateAndNormalizeQueuingStrategy ( size, highWaterMark ) throws

7. Global Properties

The following constructors must be exposed on the global object as data properties of the same name:

• ReadableStream
• WritableStream
• ByteLengthQueuingStrategy
• CountQueuingStrategy

The attributes of these properties must be { [[Writable]]: true, [[Enumerable]]: false, [[Configurable]]: true }.

8. Examples of Creating Streams

This section, and all its subsections, are non-normative.

The previous examples throughout the standard have focused on how to use streams. Here we show how to create a stream, using the ReadableStream or WritableStream constructors.

8.1. A readable stream with an underlying push source (no backpressure support)

The following function creates readable streams that wrap WebSocket instances [HTML], which are push sources that do not support backpressure signals. It illustrates how, when adapting a push source, usually most of the work happens in the start function.

function makeReadableWebSocketStream(url, protocols) {
  const ws = new WebSocket(url, protocols);
  ws.binaryType = "arraybuffer";

  return new ReadableStream({
    start(controller) {
      ws.onmessage = event => controller.enqueue(event.data);
      ws.onend = () => controller.close();
      ws.onerror = () => controller.error(new Error("The WebSocket errored!"));
    },

    cancel() {
      ws.close();
    }
  });
}

We can then use this function to create readable streams for a web socket, and pipe that stream to an arbitrary writable stream:

const webSocketStream = makeReadableWebSocketStream("wss://example.com:443/", "protocol");

webSocketStream.pipeTo(writableStream)
  .then(() => console.log("All data successfully written!"))
  .catch(e => console.error("Something went wrong!", e));

8.2. A readable stream with an underlying push source and backpressure support

The following function returns readable streams that wrap "backpressure sockets," which are hypothetical objects that have the same API as web sockets, but also provide the ability to pause and resume the flow of data with their readStop and readStart methods. In doing so, this example shows how to apply backpressure to underlying sources that support it.

function makeReadableBackpressureSocketStream(host, port) {
  const socket = createBackpressureSocket(host, port);

  return new ReadableStream({
    start(controller) {
      socket.ondata = event => {
        controller.enqueue(event.data);

        if (controller.desiredSize <= 0) {
          // The internal queue is full, so propagate
          // the backpressure signal to the underlying source.
          socket.readStop();
        }
      };

      socket.onend = () => controller.close();
      socket.onerror = () => controller.error(new Error("The socket errored!"));
    },

    pull() {
      // This is called if the internal queue has been emptied, but the
      // stream’s consumer still wants more data. In that case, restart
      // the flow of data if we have previously paused it.
      socket.readStart();
    },

    cancel() {
      socket.close();
    }
  });
}

We can then use this function to create readable streams for such "backpressure sockets" in the same way we do for web sockets. This time, however, when we pipe to a destination that cannot accept data as fast as the socket is producing it, or if we leave the stream alone without reading from it for some time, a backpressure signal will be sent to the socket.

8.3. A readable byte stream with an underlying push source and backpressure support

The following function returns readable byte streams that allow efficient direct reading of TCP sockets, based on a hypothetical JavaScript translation of the POSIX socket API (the main innovation of which is translating select(2) into an EventTarget emitting a readable event). [DOM]

This setup allows zero-copy reading directly into developer-supplied buffers. Additionally, it ensures that when data is available from the socket but not yet requested by the developer, it is enqueued in the stream’s internal queue, to avoid overflowing the kernel-space queue. In this case, an additional copy will potentially be necessary when using a BYOB reader, to move the data from the stream’s internal queue to the developer-supplied buffer. If this occurs, backpressure will immediately be applied downstream on the socket, by adjusting the TCP window size.

const DEFAULT_WINDOW_SIZE = 65536;

function makeReadableBackpressureByteSocketStream(host, port) {
  const socket = createHypotheticalSelect2Socket(host, port);

  return new ReadableStream({
    type: "bytes",

    start(controller) {
      socket.setTCPWindowSize(Math.max(0, controller.desiredSize));

      socket.onreadable = () => {
        // Since onreadable can happen even when there’s no pending BYOB
        // requests, we need to handle both cases.
        if (controller.byobRequest) {
          const v = controller.byobRequest.view;
          const bytesRead = socket.readInto(v.buffer, v.byteOffset, v.byteLength);
          controller.byobRequest.respond(bytesRead);
        } else {
          const buffer = new ArrayBuffer(DEFAULT_WINDOW_SIZE);
          const bytesRead = socket.readInto(buffer, 0, DEFAULT_WINDOW_SIZE);
          controller.enqueue(new Uint8Array(buffer, 0, bytesRead));
        }

        // The internal queue size has changed, so propagate
        // the backpressure signal to the underlying source.
        socket.setTCPWindowSize(Math.max(0, controller.desiredSize));
      };

      socket.onend = () => controller.close();
      socket.onerror = () => controller.error(new Error("The socket errored!"));
    },

    pull(controller) {
      // This is called when the internal queue has been drained, and the
      // stream’s consumer can accept more data. Reflect the up-to-date
      // backpressure level by setting the TCP receive window of the socket
      // to desiredSize.
      socket.setTCPWindowSize(Math.max(0, controller.desiredSize));
    },

    cancel() {
      socket.close();
    }
  }, {
    highWaterMark: DEFAULT_WINDOW_SIZE
  });
}

ReadableStream instances returned from this function can now vend BYOB readers, with all of the aforementioned benefits and caveats.

8.4. A readable stream with an underlying pull source

The following function returns readable streams that wrap portions of the Node.js file system API (which themselves map fairly directly to C’s fopen, fread, and fclose trio). Files are a typical example of pull sources. Note how in contrast to the examples with push sources, most of the work here happens on-demand in the pull function, and not at startup time in the start function.

const fs = require("pr/fs"); // https://github.com/jden/pr
const CHUNK_SIZE = 1024;

function makeReadableFileStream(filename) {
  let fd;
  let position = 0;

  return new ReadableStream({
    start() {
      return fs.open(filename, "r").then(result => {
        fd = result;
      });
    },

    pull(controller) {
      const buffer = new ArrayBuffer(CHUNK_SIZE);

      return fs.read(fd, buffer, 0, CHUNK_SIZE, position).then(bytesRead => {
        if (bytesRead === 0) {
          return fs.close(fd).then(() => controller.close());
        } else {
          position += bytesRead;
          controller.enqueue(new Uint8Array(buffer, 0, bytesRead));
        }
      });
    },

    cancel() {
      return fs.close(fd);
    }
  });
}

We can then create and use readable streams for files just as we could before for sockets.

8.5. A readable byte stream with an underlying pull source

The following function returns readable byte streams that allow efficient zero-copy reading of files, again using the Node.js file system API. Instead of using a predetermined chunk size of 1024, it attempts to fill the developer-supplied buffer, allowing full control.

const fs = require("pr/fs"); // https://github.com/jden/pr
const DEFAULT_CHUNK_SIZE = 1024;

function makeReadableByteFileStream(filename) {
  let fd;
  let position = 0;

  return new ReadableStream({
    type: "bytes",

    start() {
      return fs.open(filename, "r").then(result => {
        fd = result;
      });
    },

    pull(controller) {
      // Even when the consumer is using the default reader, the auto-allocation
      // feature allocates a buffer and passes it to us via byobRequest.
      const v = controller.byobRequest.view;

      return fs.read(fd, v.buffer, v.byteOffset, v.byteLength, position).then(bytesRead => {
        if (bytesRead === 0) {
          return fs.close(fd).then(() => controller.close());
        } else {
          position += bytesRead;
          controller.byobRequest.respond(bytesRead);
        }
      });
    },

    cancel() {
      return fs.close(fd);
    },

    autoAllocateChunkSize: DEFAULT_CHUNK_SIZE
  });
}