index.bs

<pre class="metadata">
Title: Streams Standard
Group: WHATWG
H1: Streams
Shortname: streams
Status: LS
Editor: Domenic Denicola, Google https://www.google.com/, d@domenic.me, https://domenic.me/
Abstract: This specification provides APIs for creating, composing, and consuming streams of data.
Abstract: These streams are designed to map efficiently to low-level I/O primitives, and allow easy
Abstract: composition with built-in backpressure and queuing. On top of streams, the web platform can
Abstract: build higher-level abstractions, such as filesystem or socket APIs, while at the same time
Abstract: users can use the supplied tools to build their own streams which integrate well with those
Abstract: of the web platform.
Logo: https://resources.whatwg.org/logo-streams.svg
!Participate: <a href="https://github.com/whatwg/streams/issues/new">File an issue</a> (<a href="https://github.com/whatwg/streams/issues?state=open">open issues</a>)
!Participate: <a href="https://wiki.whatwg.org/wiki/IRC">IRC: #whatwg on Freenode</a>
!Version History: <a href="https://github.com/whatwg/streams/commits">https://github.com/whatwg/streams/commits</a>
!Version History: [SNAPSHOT-LINK]
!Version History: <a href="https://twitter.com/streamsstandard">@streamsstandard</a>

Link Defaults: html (dfn) structured clone
</pre>

<style>
  .note + .example, .note + .note { margin-top: 1em; }

  emu-val { font-weight: bold; }
  emu-alg > ol, emu-alg > ol ol ol ol { list-style-type: decimal; }
  emu-alg > ol ol, emu-alg > ol ol ol ol ol { list-style-type: lower-alpha; }
  emu-alg > ol ol ol, emu-alg > ol ol ol ol ol ol { list-style-type: lower-roman; }
  emu-alg li { margin: 0; }
</style>
<script src="https://resources.whatwg.org/file-issue.js" async></script>
<script src="https://resources.whatwg.org/commit-snapshot-shortcut-key.js" async></script>


<h2 id="intro">Introduction</h2>

<em>This section is non-normative.</em>

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 <a>readable streams</a>,
<a>writable streams</a>, and <a>transform streams</a>.

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:

<ul>
  <li> Video effects: piping a readable video stream through a transform stream that applies effects in real time.
  <li> Decompression: piping a file stream through a transform stream that selectively decompresses files from a
    <kbd>.tgz</kbd> archive, turning them into <code>img</code> elements as the user scrolls through an image gallery.
  <li> Image decoding: piping a 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 <code>fetch</code>
    hook of a service worker [[SERVICE-WORKERS]], this would allow developers to transparently polyfill new image
    formats.
</ul>

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 <a>backpressure</a>) that come along for the
ride.

<h2 id="model">Model</h2>

A <dfn>chunk</dfn> 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 <code>Uint8Array</code>s, instead of single
bytes.

<h3 id="rs-model">Readable Streams</h3>

A <dfn>readable stream</dfn> represents a source of data, from which you can read. In other words, data comes
<em>out</em> 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 <dfn>underlying source</dfn>. There are two types of underlying source: push sources and pull sources.

<dfn lt="push source">Push sources</dfn> 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.

<dfn lt="pull source">Pull sources</dfn> 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.

<a>Chunks</a> are enqueued into the stream by the stream's <a>underlying source</a>. 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 <dfn>consumer</dfn>.

Consumers also have the ability to <dfn lt="cancel a readable stream">cancel</dfn> a readable stream. This indicates
that the consumer has lost interest in the stream, and will immediately close the stream, throw away any queued
<a>chunks</a>, and execute any cancellation mechanism of the <a>underlying source</a>.

Consumers can also <dfn lt="tee a readable stream">tee</dfn> a readable stream. This will
<a lt="locked to a reader">lock</a> the stream, making it no longer directly usable; however, it will create two new
streams, called <dfn lt="branches of a readable stream tee">branches</dfn>, which can be consumed independently.

<h3 id="ws-model">Writable Streams</h3>

A <dfn>writable stream</dfn> represents a destination for data, into which you can write. In other words, data goes
<em>in</em> to a writable stream.

Analogously to readable streams, most writable streams wrap a lower-level I/O sink, called the
<dfn>underlying sink</dfn>. 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.

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

Code that writes into a writable stream using its public interface is known as a <dfn>producer</dfn>.

Producers also have the ability to <dfn lt="abort a writable stream">abort</dfn> 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 <a>underlying sink</a>.

<h3 id="ts-model">Transform Streams</h3>

A <dfn>transform stream</dfn> 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:

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

<h3 id="pipe-chains">Pipe Chains and Backpressure</h3>

Streams are primarily used by <dfn>piping</dfn> 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 <dfn>pipe chain</dfn>. In a pipe chain, the
<dfn>original source</dfn> is the <a>underlying source</a> of the first readable stream in the chain; the
<dfn>ultimate sink</dfn> is the <a>underlying sink</a> of the final writable stream in the chain.

Once a pipe chain is constructed, it can be used to propagate signals regarding how fast <a>chunks</a> 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 <dfn>backpressure</dfn>.

When <a lt="tee a readable stream">teeing</a> a readable stream, the <a>backpressure</a> signals from its two
<a href="branches of a readable stream tee">branches</a> will aggregate, such that if neither branch is read from, a
backpressure signal will be sent to the <a>underlying source</a> of the original stream.

<!-- TODO when we have writable stream writers
Piping a readable stream <a href="locked to a reader">locks</a> the readable stream, preventing it from being accessed
-->

<h3 id="queuing-strategies">Internal Queues and Queuing Strategies</h3>

Both readable and writable streams maintain <dfn>internal queues</dfn>, which they use for similar purposes. In the
case of a readable stream, the internal queue contains <a>chunks</a> that have been enqueued by the <a>underlying
source</a>, but not yet read by the consumer. In the case of a writable stream, the internal queue contains
<a>chunks</a> which have been written to the stream by the producer, but not yet processed and acknowledged by the
<a>underlying sink</a>.

A <dfn>queuing strategy</dfn> is an object that determines how a stream should signal <a>backpressure</a> based on
the state of its <a>internal queue</a>. The queuing strategy assigns a size to each <a>chunk</a>, and compares the
total size of all chunks in the queue to a specified number, known as the <dfn>high water mark</dfn>. The resulting
difference, high water mark minus total size, is used to determine the
<dfn lt="desired size to fill a stream's internal queue">desired size to fill the stream's queue</dfn>.

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.

<div class="example">
  A simple example of a queuing strategy would be one that assigns a size of one to each chunk, and has a high water
  mark of three. This would mean that up to three chunks could be enqueued in a readable stream, or three chunks
  written to a writable stream, before the streams are considered to be applying backpressure.
</div>

<h3 id="locking">Locking</h3>

<!-- TODO: writable streams too, probably -->

A <dfn>readable stream reader</dfn> (<dfn>readable byte stream reader</dfn> or
<dfn>readable byte stream byob reader</dfn> for <a>readable byte stream</a>) or simply reader is an object that allows
direct reading of <a>chunks</a> from a <a>readable stream</a>). Without a reader, a <a>consumer</a> can only perform
high-level operations on the readable stream: waiting for the stream to become closed or errored,
<a lt="cancel a readable stream">canceling</a> the stream, or <a>piping</a> 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
<dfn lt="locked to a reader">locked to the reader</dfn>, and that the reader is <dfn lt="active reader">active</dfn>.

A reader also has the capability to <dfn lt="release a read lock">release its read lock</dfn>, 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 <a>underlying source</a> or via <a lt="cancel a readable stream">cancellation</a>, its
reader (if one exists) will automatically release its lock.

<h3 id="byte-streams">Readable Byte Streams</h3>

For streams representing bytes, an extended version of the readable stream is provided to handle bytes efficiently.

A <dfn>readable byte stream</dfn> represents a source of bytes, from which you can read.

Although a readable byte stream can be created with arbitrary behavior, most readable byte streams wrap a lower-level
I/O source, called the <dfn>underlying byte source</dfn>.

<h2 id="rs">Readable Streams</h2>

<h3 id="rs-intro">Using Readable Streams</h3>

<div class="example">
  The simplest way to consume a readable stream is to simply <a lt="piping">pipe</a> it to a <a>writable stream</a>.
  This ensures that <a>backpressure</a> is respected, and any errors (either writing or reading) are propagated through
  the chain:

  <pre><code class="lang-javascript">
    readableStream.pipeTo(writableStream)
      .then(() => console.log("All data successfully written!"))
      .catch(e => console.error("Something went wrong!", e));
  </code></pre>
</div>

<div class="example">
  Although readable streams will usually be used by piping them to a writable stream, you can also read them directly
  by acquiring a <a lt="readable stream reader">reader</a> and using its <code>read()</code> method to get successive
  chunks. For example, this code logs the next <a>chunk</a> in the stream, if available:

  <pre><code class="lang-javascript">
    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)
    );
  </code></pre>
</div>

<h3 id="rs-class">Class <code>ReadableStream</code></h3>

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

<h4 id="rs-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

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

<pre><code class="lang-javascript">
  class ReadableStream {
    constructor(underlyingSource = {}, { size, highWaterMark = 1 } = {})

    cancel(reason)
    getReader()
    pipeThrough({ writable, readable }, options)
    pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})
    tee()
  }
</code></pre>

<h4 id="rs-internal-slots">Internal Slots</h4>

Instances of <code>ReadableStream</code> are created with the internal slots described in the following table:

<table>
  <thead>
    <tr>
      <th>Internal Slot</th>
      <th>Description (<em>non-normative</em>)</th>
    </tr>
  </thead>
  <tr>
    <td>\[[closeRequested]]
    <td>A boolean flag indicating whether the stream has been closed by its <a>underlying source</a>, but still has
      <a>chunks</a> in its internal queue that have not yet been read
  </tr>
  <tr>
    <td>\[[controller]]
    <td>A <a href="#rs-controller-class"><code>ReadableStreamController</code></a> created with the ability to control the
      state and queue of this stream
  </tr>
  <tr>
    <td>\[[pullAgain]]
    <td>A boolean flag set to <b>true</b> if the stream's mechanisms requested a call to the underlying source's
      <code>pull</code> method to pull more data, but the pull could not yet be done since a previous call is still
      executing
  </tr>
  <tr>
    <td>\[[pulling]]
    <td>A boolean flag set to <b>true</b> while the underlying source's <code>pull</code> method is executing and has
      not yet fulfilled, used to prevent reentrant calls
  </tr>
  <tr>
    <td>\[[queue]]
    <td>A List representing the stream's internal queue of <a>chunks</a>
  </tr>
  <tr>
    <td>\[[reader]]
    <td>A <a href="#reader-class"><code>ReadableStreamReader</code></a> instance, if the stream is <a>locked to a
      reader</a>, or <b>undefined</b> if it is not
  </tr>
  <tr>
    <td>\[[started]]
    <td>A boolean flag indicating whether the <a>underlying source</a> has finished starting
  </tr>
  <tr>
    <td>\[[state]]
    <td>A string containing the stream's current state, used internally; one of <code>"readable"</code>,
      <code>"closed"</code>, or <code>"errored"</code>.
  </tr>
  <tr>
    <td>\[[storedError]]
    <td>A value indicating how the stream failed, to be given as a failure reason or exception when trying to operate
      on an errored stream
  </tr>
  <tr>
    <td>\[[strategySize]]
    <td>A function supplied to the constructor as part of the stream's <a>queuing strategy</a>, designed to calculate
      the size of enqueued <a>chunks</a>; can be <b>undefined</b> for the default behavior.
  </tr>
  <tr>
    <td>\[[strategyHWM]]
    <td>A number supplied to the constructor as part of the stream's <a>queuing strategy</a>, indicating the point at
      which the stream will apply <a>backpressure</a> to its <a>underlying source</a>.
  </tr>
  <tr>
    <td>\[[underlyingSource]]
    <td>An object representation of the stream's <a>underlying source</a>, including its <a>queuing strategy</a>; also
      used for the <a href="#is-readable-stream">IsReadableStream</a> brand check
  </tr>
</table>

<h4 id="rs-constructor">new ReadableStream(underlyingSource = {}, { size, highWaterMark = 1 } = {})</h4>

<div class="note">
  The <var>underlyingSource</var> object passed to the constructor can implement any of the following methods to
  govern how the constructed stream instance behaves:

  <ul>
    <li> <code>start(controller)</code> is called immediately, and is typically used to adapt a <a>push
      source</a> by setting up relevant event listeners, or to acquire access to a <a>pull source</a>. If this process
      is asynchronous, it can return a promise to signal success or failure.
    <li> <code>pull(controller)</code> is called when the stream's internal queue of chunks is depleted, and the
      consumer has signaled that they wish to consume more data. If <code>pull</code> returns a promise, then
      <code>pull</code> will not be called again until that promise fulfills; if the promise rejects, the stream will
      become errored.
    <li> <code>cancel(reason)</code> is called when the consumer signals that they are no longer interested in the
      stream. It should perform any actions necessary to release access to the <a>underlying source</a>. If this
      process is asynchronous, it can return a promise to signal success or failure.
  </ul>

  Both <code>start</code> and <code>pull</code> are given the ability to manipulate the stream's internal queue and
  state via the passed <code>controller</code> object, which is an instance of
  <a href="#rs-controller-class"><code>ReadableStreamController</code></a>. This is an example of the
  <a href="https://blog.domenic.me/the-revealing-constructor-pattern/">revealing constructor pattern</a>.

  The constructor also accepts a second argument containing the <a>queuing strategy</a> object with
  two properties: a non-negative number <code>highWaterMark</code>, and a function <code>size(chunk)</code>. The
  supplied <code>strategy</code> could be an instance of the built-in <code>CountQueuingStrategy</code> or
  <code>ByteLengthQueuingStrategy</code> classes, or it could be custom. If no strategy is supplied, the default
  behavior will be the same as a <code>CountQueuingStrategy</code> with a <a>high water mark</a> of 1.
</div>

<pre is="emu-alg">
  1. Set *this*@[[underlyingSource]] to _underlyingSource_.
  1. Set *this*@[[queue]] to a new empty List.
  1. Set *this*@[[state]] to "readable".
  1. Set *this*@[[started]], *this*@[[closeRequested]], *this*@[[pullAgain]], and *this*@[[pulling]] to *false*.
  1. Set *this*@[[reader]] and *this*@[[storedError]] to *undefined*.
  1. Set *this*@[[controller]] to Construct(`ReadableStreamController`, «*this*»).
  1. Let _normalizedStrategy_ be ValidateAndNormalizeQueuingStrategy(_size_, _highWaterMark_).
  1. Set *this*@[[strategySize]] to _normalizedStrategy_.[[size]] and *this*@[[strategyHWM]] to _normalizedStrategy_.[[highWaterMark]].
  1. Let _startResult_ be InvokeOrNoop(_underlyingSource_, "start", «*this*@[[controller]]»).
  1. ReturnIfAbrupt(_startResult_).
  1. Resolve _startResult_ as a promise:
    1. Upon fulfillment,
      1. Set *this*@[[started]] to *true*.
      1. Return RequestReadableStreamPull(*this*).
    1. Upon rejection with reason _r_,
      1. If *this*@[[state]] is "readable", return ErrorReadableStream(*this*, _r_).
</pre>

<h4 id="rs-prototype">Properties of the <code>ReadableStream</code> Prototype</h4>

<h5 id="rs-cancel">cancel(reason)</h5>

<div class="note">
  The <code>cancel</code> method <a lt="cancel a readable stream">cancels</a> the stream, signaling a loss of interest
  in the stream by a consumer. The supplied <var>reason</var> argument will be given to the underlying source, which
  may or may not use it.
</div>

<pre is="emu-alg">
  1. If IsReadableStream(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. If IsReadableStreamLocked(*this*) is *true*, return a promise rejected with a *TypeError* exception.
  1. Return CancelReadableStream(*this*, _reason_).
</pre>

<h5 id="rs-get-reader">getReader()</h5>

<div class="note">
  The <code>getReader</code> method creates a <a>readable stream reader</a> and
  <a lt="locked to a reader">locks</a> the stream to the new reader. While the stream is locked, no other reader
  can be acquired until this one is <a lt="release a read lock">released</a>. The returned reader provides the ability
  to directly read individual <a>chunks</a> from the stream via the reader's <code>read</code> method.

  This functionality is especially useful for creating abstractions that desire the ability to consume a stream in its
  entirety. By getting a reader for the stream, you can ensure nobody else can interleave reads with yours or cancel
  the stream, which would interfere with your abstraction.

  Note that if a stream becomes closed or errored, any reader it is locked to is automatically released.
</div>

<pre is="emu-alg">
  1. If IsReadableStream(*this*) is *false*, throw a *TypeError* exception.
  1. Return AcquireReadableStreamReader(*this*).
</pre>

<div class="example">
  An example of an abstraction that might benefit from using a reader is a function like the following, which is
  designed to read an entire readable stream into memory as an array of <a>chunks</a>.

  <pre><code class="lang-javascript">
    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();
        });
      }
    }
  </code></pre>

  Note how the first thing it does is obtain a reader, and from then on it uses the reader exclusively. This ensures
  that no other consumer can interfere with the stream, either by reading chunks or by
  <a lt="cancel a readable stream">canceling</a> the stream.
</div>

<h5 id="rs-pipe-through">pipeThrough({ writable, readable }, options)</h5>

<div class="note">
  The <code>pipeThrough</code> method provides a convenient, chainable way of <a>piping</a> this <a>readable stream</a>
  through a <a>transform stream</a> (or any other <code>{ writable, readable }</code> pair). It simply pipes the stream
  into the writable side of the supplied pair, and returns the readable side for further use.

  Piping a stream will generally <a lt="locked to a reader">lock</a> it for the duration of the pipe, preventing any
  other consumer from acquiring a reader.

  This method is intentionally generic; it does not require that its <b>this</b> value be a <code>ReadableStream</code>
  object. It also does not require that its <code>writable</code> argument be a <code>WritableStream</code> instance,
  or that its <code>readable</code> argument be a <code>ReadableStream</code> instance.
</div>

<pre is="emu-alg">
  1. Call-with-rethrow Invoke(*this*, "pipeTo", «_writable_, _options_»).
  1. Return _readable_.
</pre>

<div class="example">
  A typical example of constructing <a>pipe chain</a> using <code>pipeThrough</code> would look like

  <pre><code class="lang-javascript">
    httpResponseBody
      .pipeThrough(decompressorTransform)
      .pipeThrough(ignoreNonImageFilesTransform)
      .pipeTo(mediaGallery);
  </code></pre>
</div>

<h5 id="rs-pipe-to">pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})</h5>

<div class="note">
  The <code>pipeTo</code> method <a lt="piping">pipes</a> this <a>readable stream</a> to a given <a>writable
  stream</a>. The way in which the piping process behaves under various error conditions can be customized with a
  number of passed options. It returns a promise that fulfills when the piping process completes successfully, or
  rejects if any errors were encountered.

  Piping a stream will generally <a lt="locked to a reader">lock</a> it for the duration of the pipe, preventing any
  other consumer from acquiring a reader.

  This method is intentionally generic; it does not require that its <b>this</b> value be a <code>ReadableStream</code>
  object.
</div>

The <code>pipeTo</code> method is one of the more complex methods, and is undergoing some revision and edge-case
bulletproofing before we write it up in prose.

For now, please consider the reference implementation normative:
<a href="https://github.com/whatwg/streams/blob/master/reference-implementation/lib/readable-stream.js">reference-implementation/lib/readable-stream.js</a>,
look for the <code>pipeTo</code> method.

<h5 id="rs-tee">tee()</h5>

<div class="note">
  The <code>tee</code> method <a lt="tee a readable stream">tees</a> this readable stream, returning a two-element
  array containing the two resulting branches as new <code>ReadableStream</code> instances.

  Teeing a stream will <a lt="locked to a reader">lock</a> it, preventing any other consumer from acquiring a reader.
  To <a lt="cancel a readable stream">cancel</a> the stream, cancel both of the resulting branches; a composite
  cancellation reason will then be propagated to the stream's <a>underlying source</a>.

  Note that the <a>chunks</a> seen in each branch will be the same object. If the chunks are not immutable, this could
  allow interference between the two branches. (<a href="https://github.com/whatwg/streams/issues/new">Let us know</a>
  if you think we should add an option to <code>tee</code> that creates <a>structured clones</a> of the chunks for each
  branch.)
</div>

<pre is="emu-alg">
  1. If IsReadableStream(*this*) is *false*, throw a *TypeError* exception.
  1. Let _branches_ be TeeReadableStream(*this*, *false*).
  1. ReturnIfAbrupt(_branches_).
  1. Return CreateArrayFromList(_branches_).
</pre>

<div class="example">
  Teeing a stream is most useful when you wish to let two independent consumers read from the stream in parallel,
  perhaps even at different speeds. For example, given a writable stream <code>cacheEntry</code> representing an
  on-disk file, and another writable stream <code>httpRequestBody</code> representing an upload to a remote server,
  you could pipe the same readable stream to both destinations at once:

  <pre><code class="lang-javascript">
    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));
  </code></pre>
</div>


<h3 id="rs-controller-class" lt="ReadableStreamController">Class <code>ReadableStreamController</code></h3>

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

<h4 id="rs-controller-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

If one were to write the <code>ReadableStreamController</code> class in something close to the syntax of [[!ECMASCRIPT]],
it would look like

<pre><code class="lang-javascript">
  class ReadableStreamController {
    constructor(stream)

    get desiredSize()

    close()
    enqueue(chunk)
    error(e)
  }
</code></pre>

<h4 id="rs-controller-internal-slots">Internal Slots</h4>

Instances of <code>ReadableStreamController</code> are created with the internal slots described in the following table:

<table>
  <thead>
    <tr>
      <th>Internal Slot</th>
      <th>Description (<em>non-normative</em>)</th>
    </tr>
  </thead>
  <tr>
    <td>\[[controlledReadableStream]]
    <td>The <code>ReadableStream</code> instance controlled
  </tr>
</table>

<h4 id="rs-controller-constructor">new ReadableStreamController(stream)</h4>

<div class="note">
  The <code>ReadableStreamController</code> constructor cannot be used directly; it only works on a
  <code>ReadableStream</code> that is in the middle of being constructed.
</div>

<pre is="emu-alg">
  1. If IsReadableStream(_stream_) is *false*, throw a *TypeError* exception.
  1. If _stream_@[[controller]] is not *undefined*, throw a *TypeError* exception.
  1. Set *this*@[[controlledReadableStream]] to _stream_.
</pre>

<h4 id="rs-controller-prototype">Properties of the <code>ReadableStreamController</code> Prototype</h4>

<h5 id="rs-controller-desired-size">get desiredSize</h5>

<div class="note">
  The <code>desiredSize</code> getter returns the <a lt="desired size to fill a stream's internal queue">desired size
  to fill the controlled stream's internal queue</a>. It can be negative, if the queue is over-full. An <a>underlying
  source</a> should use this information to determine when and how to apply <a>backpressure</a>.
</div>

<pre is="emu-alg">
  1. If IsReadableStreamController(*this*) is *false*, throw a *TypeError* exception.
  1. Return GetReadableStreamDesiredSize(*this*@[[controlledReadableStream]]).
</pre>

<h5 id="rs-controller-close">close()</h5>

<div class="note">
  The <code>close</code> method will close the controlled readable stream. <a>Consumers</a> will still be able to read
  any previously-enqueued <a>chunks</a> from the stream, but once those are read, the stream will become closed.
</div>

<pre is="emu-alg">
  1. If IsReadableStreamController(*this*) is *false*, throw a *TypeError* exception.
  1. Let _stream_ be *this*@[[controlledReadableStream]].
  1. If _stream_@[[closeRequested]] is *true*, throw a *TypeError* exception.
  1. If _stream_@[[state]] is "errored", throw a *TypeError* exception.
  1. Return CloseReadableStream(_stream_).
</pre>

<h5 id="rs-controller-enqueue">enqueue(chunk)</h5>

<div class="note">
  The <code>enqueue</code> method will enqueue a given <a>chunk</a> in the controlled readable stream.
</div>

<pre is="emu-alg">
  1. If IsReadableStreamController(*this*) is *false*, throw a *TypeError* exception.
  1. Let _stream_ be *this*@[[controlledReadableStream]].
  1. If _stream_@[[state]] is "errored", throw _stream_@[[storedError]].
  1. If _stream_@[[closeRequested]] is *true*, throw a *TypeError* exception.
  1. Return EnqueueInReadableStream(_stream_, _chunk_).
</pre>

<h5 id="rs-controller-error">error(e)</h5>

<div class="note">
  The <code>error</code> method will error the readable stream, making all future interactions with it fail with the
  given error <var>e</var>.
</div>

<pre is="emu-alg">
  1. If IsReadableStreamController(*this*) is *false*, throw a *TypeError* exception.
  1. If *this*@[[controlledReadableStream]]@[[state]] is not "readable", throw a *TypeError* exception.
  1. Return ErrorReadableStream(*this*@[[controlledReadableStream]]).
</pre>

<h3 id="reader-class">Class <code>ReadableStreamReader</code></h3>

The <code>ReadableStreamReader</code> class represents a <a>readable stream reader</a> designed to be vended by a
<code>ReadableStream</code> instance.

<h4 id="reader-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

If one were to write the <code>ReadableStreamReader</code> class in something close to the syntax of [[!ECMASCRIPT]],
it would look like

<pre><code class="lang-javascript">
  class ReadableStreamReader {
    constructor(stream)

    get closed()

    cancel(reason)
    read()
    releaseLock()
  }
</code></pre>

<h4 id="reader-internal-slots">Internal Slots</h4>

Instances of <code>ReadableStreamReader</code> are created with the internal slots described in the following table:

<table>
  <thead>
    <tr>
      <th>Internal Slot</th>
      <th>Description (<em>non-normative</em>)</th>
    </tr>
  </thead>
  <tr>
    <td>\[[closedPromise]]
    <td>A promise returned by the reader's <code>closed</code> getter
  </tr>
  <tr>
    <td>\[[ownerReadableStream]]
    <td>A <code>ReadableStream</code> instance that owns this reader; also used for the
      <a href="#is-readable-stream-reader">IsReadableStreamReader</a> brand check
  </tr>
  <tr>
    <td>\[[readRequests]]
    <td>A List of promises returned by calls to the reader's <code>read()</code> method that have not yet been resolved,
      due to the <a>consumer</a> requesting <a>chunks</a> sooner than they are available
  </tr>
  <tr>
    <td>\[[state]]
    <td>A string containing the reader's current state, used internally; one of <code>"readable"</code>,
      <code>"closed"</code>, or <code>"errored"</code>
  </tr>
  <tr>
    <td>\[[storedError]]
    <td>A value indicating how the reader's stream failed, to be given as a failure reason or exception when trying to
      operate on the reader; applicable only when \[[state]] is <code>"errored"</code>
  </tr>
</table>

<h4 id="reader-constructor">new ReadableStreamReader(stream)</h4>

<div class="note">
  The <code>ReadableStreamReader</code> constructor is generally not meant to be used directly; instead, a stream's
  <code>getReader()</code> method should be used. This allows different classes of readable streams to vend different
  classes of readers without the <a>consumer</a> needing to know which goes with which.
</div>

<pre is="emu-alg">
  1. If IsReadableStream(_stream_) is *false*, throw a *TypeError* exception.
  1. If IsReadableStreamLocked(_stream_) is *true*, throw a *TypeError* exception.
  1. Set _stream_@[[reader]] to *this*.
  1. Set *this*@[[ownerReadableStream]] to _stream_.
  1. Set *this*@[[state]] to "readable".
  1. Set *this*@[[storedError]] to *undefined*.
  1. Set *this*@[[readRequests]] to a new empty List.
  1. Set *this*@[[closedPromise]] to a new promise.
  1. If _stream_@[[state]] is "closed" or "errored", call-with-rethrow ReleaseReadableStreamReader(*this*).
</pre>

<h4 id="reader-prototype">Properties of the <code>ReadableStreamReader</code> Prototype</h4>

<h5 id="reader-closed">get closed</h5>

<div class="note">
  The <code>closed</code> getter returns a promise that will be fulfilled when the stream becomes closed or the
  reader's lock is <a lt="release a read lock">released</a>, or rejected if the stream ever errors.
</div>

<pre is="emu-alg">
  1. If IsReadableStreamReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. Return *this*@[[closedPromise]].
</pre>

<h5 id="reader-cancel">cancel(reason)</h5>

<div class="note">
  If the reader is <a lt="active reader">active</a>, the <code>cancel</code> method behaves the same as that for the
  associated stream. When done, it automatically <a lt="release a read lock">releases the lock</a>.
</div>

<pre is="emu-alg">
  1. If IsReadableStreamReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. If *this*@[[state]] is "closed", return a new promise resolved with *undefined*.
  1. If *this*@[[state]] is "errored", return a new promise rejected with *this*@[[storedError]].
  1. Assert: *this*@[[ownerReadableStream]] is not *undefined*.
  1. Assert: *this*@[[ownerReadableStream]]@[[state]] is "readable".
  1. Return CancelReadableStream(*this*@[[ownerReadableStream]], _reason_).
</pre>

<h5 id="reader-read">read()</h5>

<div class="note">
  The <code>read</code> method will return a promise that allows access to the next <a>chunk</a> from the stream's
  internal queue, if available.

  <ul>
    <li> If the chunk does become available, the promise will be fulfilled with an object of the form
      <code>{ value: theChunk, done: false }</code>.
    <li> If the stream becomes closed, the promise will be fulfilled with an object of the form
      <code>{ value: undefined, done: true }</code>.
    <li> If the stream becomes errored, the promise will be rejected with the relevant error.
  </ul>

  If reading a chunk causes the queue to become empty, more data will be pulled from the <a>underlying source</a>.
</div>

<pre is="emu-alg">
  1. If IsReadableStreamReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. Return ReadFromReadableStreamReader(*this*).
</pre>

<h5 id="reader-release-lock">releaseLock()</h5>

<div class="note">
  The <code>releaseLock</code> method <a lt="release a read lock">releases the reader's lock</a> on the corresponding
  stream. After the lock is released, the reader is no longer <a lt="active reader">active</a>. If the associated
  stream is errored when the lock is released, the reader will appear errored in the same way from now on; otherwise,
  the reader will appear closed.

  A reader's lock cannot be released while it still has a pending read request, i.e., if a promise returned by the
  reader's <code>read()</code> method has not yet been settled. Attempting to do so will throw a <b>TypeError</b>
  and leave the reader locked to the stream.
</div>

<pre is="emu-alg">
  1. If IsReadableStreamReader(*this*) is *false*, throw a *TypeError* exception.
  1. If *this*@[[ownerReadableStream]] is *undefined*, return *undefined*.
  1. If *this*@[[readRequests]] is not empty, throw a *TypeError* exception.
  1. Return ReleaseReadableStreamReader(*this*).
</pre>

<h3 id="rs-abstract-ops">Readable Stream Abstract Operations</h3>

<h4 id="acquire-readable-stream-reader" aoid="AcquireReadableStreamReader">AcquireReadableStreamReader ( stream )</h4>

This abstract operation is meant to be called from other specifications that may wish to acquire an <a>readable stream
reader</a> for a given stream.

<pre is="emu-alg">
  1. Return Construct(`ReadableStreamReader`, «‍_stream_»).
</pre>

<h4 id="cancel-readable-stream" aoid="CancelReadableStream">CancelReadableStream ( stream, reason )</h4>

<pre is="emu-alg">
  1. If _stream_@[[state]] is "closed", return a new promise resolved with *undefined*.
  1. If _stream_@[[state]] is "errored", return a new promise rejected with _stream_@[[storedError]].
  1. Set _stream_@[[queue]] to a new empty List.
  1. Call-with-rethrow FinishClosingReadableStream(_stream_).
  1. Let _sourceCancelPromise_ be PromiseInvokeOrNoop(_stream_@[[underlyingSource]], "cancel", «‍_reason_»).
  1. Return the result of transforming _sourceCancelPromise_ by a fulfillment handler that returns *undefined*.
</pre>

<h4 id="close-readable-stream" aoid="CloseReadableStream">CloseReadableStream ( stream )</h4>

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 <em>not</em> do
this to streams they did not create, and must ensure they have obeyed the preconditions (listed here as asserts).

<pre is="emu-alg">
  1. Assert: _stream_@[[closeRequested]] is *false*.
  1. Assert: _stream_@[[state]] is not "errored".
  1. If _stream_@[[state]] is "closed", return *undefined*.
  1. Set _stream_@[[closeRequested]] to *true*.
  1. If _stream_@[[queue]] is empty, return FinishClosingReadableStream(_stream_).
</pre>

<div class="note">
  The case where <var>stream</var>@\[[state]] is <code>"closed"</code>, but <var>stream</var>@\[[closeRequested]] is
  <emu-val>false</emu-val>, will happen if the stream was closed without its controller's close method ever being
  called: i.e., if the stream was closed by a call to <code>stream.cancel()</code>. In this case we allow the
  controller's <code>close</code> method to be called and silently do nothing, since the cancelation was outside the
  control of the underlying source.
</div>

<h4 id="finish-closing-readable-stream" aoid="FinishClosingReadableStream">FinishClosingReadableStream ( stream )</h4>

<pre is="emu-alg">
  1. Assert: _stream_@[[state]] is "readable".
  1. Set _stream_@[[state]] to "closed".
  1. If IsReadableStreamLocked(_stream_) is *true*, return ReleaseReadableStreamReader(_stream_@[[reader]]).
  1. Return *undefined*.
</pre>

<h4 id="enqueue-in-readable-stream" aoid="EnqueueInReadableStream">EnqueueInReadableStream ( stream, chunk )</h4>

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

<pre is="emu-alg">
  1. Assert: _stream_@[[closeRequested]] is *false*.
  1. Assert: _stream_@[[state]] is not "errored".
  1. If _stream_@[[state]] is "closed", return *undefined*.
  1. If IsReadableStreamLocked(_stream_) is *true* and _stream_@[[reader]]@[[readRequests]] is not empty,
    1. Let _readRequestPromise_ be the first element of _stream_@[[reader]]@[[readRequests]].
    1. Remove _readRequestPromise_ from _stream_@[[reader]]@[[readRequests]], shifting all other elements downward (so that the second becomes the first, and so on).
    1. Resolve _readRequestPromise_ with CreateIterResultObject(_chunk_, *false*).
  1. Otherwise,
    1. Let _chunkSize_ be *1*.
    1. If _stream_@[[strategySize]] is not *undefined*,
      1. Set _chunkSize_ to Call(_stream_@[[strategySize]], *undefined*, «‍_chunk_»).
      1. If _chunkSize_ is an abrupt completion,
        1. Call-with-rethrow ErrorReadableStream(_stream_, ‍_chunkSize_.[[value]]).
        1. Return _chunkSize_.
      1. Let _chunkSize_ be _chunkSize_.[[value]].
    1. Let _enqueueResult_ be EnqueueValueWithSize(_stream_@[[queue]], _chunk_, _chunkSize_).
    1. If _enqueueResult_ is an abrupt completion,
      1. Call-with-rethrow ErrorReadableStream(_stream_, ‍_enqueueResult_.[[value]]).
      1. Return _enqueueResult_.
  1. Call-with-rethrow RequestReadableStreamPull(_stream_).
  1. Return *undefined*.
</pre>

<div class="note">
  The case where <var>stream</var>@\[[state]] is <code>"closed"</code>, but <var>stream</var>@\[[closeRequested]] is
  <emu-val>false</emu-val>, will happen if the stream was closed without its controller's close method ever being
  called: i.e., if the stream was closed by a call to <code>stream.cancel()</code>. In this case we allow the
  controller's <code>enqueue</code> method to be called and silently do nothing, since the cancelation was outside the
  control of the underlying source.
</div>

<h4 id="error-readable-stream" aoid="ErrorReadableStream">ErrorReadableStream ( stream, e )</h4>

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
<em>not</em> do this to streams they did not create, and must ensure they have obeyed the precondition (listed here as
an assert).

<pre is="emu-alg">
  1. Assert: _stream_@[[state]] is "readable".
  1. Let _stream_@[[queue]] be a new empty List.
  1. Set _stream_@[[storedError]] to _e_.
  1. Set _stream_@[[state]] to "errored".
  1. If IsReadableStreamLocked(_stream_) is *true*, return ReleaseReadableStreamReader(_stream_@[[reader]]).
</pre>

<h4 id="get-readable-stream-desired-size" aoid="GetReadableStreamDesiredSize">GetReadableStreamDesiredSize ( stream )</h4>

<pre is="emu-alg">
  1. Let _queueSize_ be GetTotalQueueSize(_stream_@[[queue]]).
  1. ReturnIfAbrupt(_queueSize_).
  1. Return _stream_@[[strategyHWM]] − _queueSize_.
</pre>

<h4 id="is-readable-stream" aoid="IsReadableStream">IsReadableStream ( x )</h4>

<pre is="emu-alg">
  1. If Type(_x_) is not Object, return *false*.
  1. If _x_ does not have a [[underlyingSource]] internal slot, return *false*.
  1. Return *true*.
</pre>

<h4 id="is-readable-stream-controller" aoid="IsReadableStreamController">IsReadableStreamController ( x )</h4>

<pre is="emu-alg">
  1. If Type(_x_) is not Object, return *false*.
  1. If _x_ does not have a [[controlledReadableStream]] internal slot, return *false*.
  1. Return *true*.
</pre>

<h4 id="is-readable-stream-locked" aoid="IsReadableStreamLocked">IsReadableStreamLocked ( stream )</h4>

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

<pre is="emu-alg">
  1. Assert: IsReadableStream(_stream_) is *true*.
  1. If _stream_@[[reader]] is *undefined*, return *false*.
  1. Return *true*.
</pre>

<h4 id="is-readable-stream-reader" aoid="IsReadableStreamReader">IsReadableStreamReader ( x )</h4>

<pre is="emu-alg">
  1. If Type(_x_) is not Object, return *false*.
  1. If _x_ does not have an [[ownerReadableStream]] internal slot, return *false*.
  1. Return *true*.
</pre>

<h4 id="read-from-readable-stream-reader" aoid="ReadFromReadableStreamReader">ReadFromReadableStreamReader ( reader )</h4>

<pre is="emu-alg">
  1. If _reader_@[[state]] is "closed", return a new promise resolved with CreateIterResultObject(*undefined*, *true*).
  1. If _reader_@[[state]] is "errored", return a new promise rejected with _reader_@[[storedError]].
  1. Assert: _reader_@[[ownerReadableStream]] is not *undefined*.
  1. Assert: _reader_@[[ownerReadableStream]]@[[state]] is "readable".
  1. If _reader_@[[ownerReadableStream]]@[[queue]] is not empty,
    1. Let _chunk_ be DequeueValue(_reader_@[[ownerReadableStream]]@[[queue]]).
    1. If _reader_@[[ownerReadableStream]]@[[closeRequested]] is *true* and _reader_@[[ownerReadableStream]]@[[queue]] is now empty, call-with-rethrow FinishClosingReadableStream(_reader_@[[ownerReadableStream]]).
    1. Otherwise, call-with-rethrow RequestReadableStreamPull(_reader_@[[ownerReadableStream]]).
    1. Return a new promise resolved with CreateIterResultObject(_chunk_, *false*).
  1. Otherwise,
    1. Let _readRequestPromise_ be a new promise.
    1. Append _readRequestPromise_ as the last element of _reader_@[[readRequests]].
    1. Call-with-rethrow RequestReadableStreamPull(_reader_@[[ownerReadableStream]]).
    1. Return _readRequestPromise_.
</pre>

<h4 id="release-readable-stream-reader" aoid="ReleaseReadableStreamReader">ReleaseReadableStreamReader ( reader )</h4>

<pre is="emu-alg">
  1. Assert: _reader_@[[ownerReadableStream]] is not *undefined*.
  1. If _reader_@[[ownerReadableStream]]@[[state]] is "errored",
    1. Set _reader_@[[state]] to "errored".
    1. Let _e_ be _reader_@[[ownerReadableStream]]@[[storedError]].
    1. Set _reader_@[[storedError]] to _e_.
    1. Reject _reader_@[[closedPromise]] with _e_.
    1. Repeat for each _readRequestPromise_ that is an element of _reader_@[[readRequests]],
      1. Reject _readRequestPromise_ with _e_.
  1. Otherwise,
    1. Set _reader_@[[state]] to "closed".
    1. Resolve _reader_@[[closedPromise]] with *undefined*.
    1. Repeat for each _readRequestPromise_ that is an element of _reader_@[[readRequests]],
      1. Resolve _readRequestPromise_ with CreateIterResultObject(*undefined*, *true*).
  1. Set _reader_@[[readRequests]] to a new empty List.
  1. Set _reader_@[[ownerReadableStream]]@[[reader]] to *undefined*.
  1. Set _reader_@[[ownerReadableStream]] to *undefined*.
</pre>

<h4 id="request-readable-stream-pull" aoid="RequestReadableStreamPull">RequestReadableStreamPull ( stream )</h4>

<pre is="emu-alg">
  1. Let _shouldPull_ be ShouldReadableStreamPull(stream).
  1. ReturnIfAbrupt(_shouldPull_).
  1. If _shouldPull_ is *false*, return *undefined*.
  1. If _stream_@[[pulling]] is *true*,
    1. Set _stream_@[[pullAgain]] to *true*.
    1. Return *undefined*.
  1. Set _stream_@[[pulling]] to *true*.
  1. Let _pullPromise_ be PromiseInvokeOrNoop(_stream_@[[underlyingSource]], "pull", «‍_stream_@[[controller]]»).
  1. Upon fulfillment of _pullPromise_,
    1. Set _stream_@[[pulling]] to *false*.
    1. If _stream_@[[pullAgain]] is *true*,
      1. Set _stream_@[[pullAgain]] to *false*.
      1. Return RequestReadableStreamPull(_stream_).
  1. Upon rejection of _pullPromise_ with reason _e_,
    1. If _stream_@[[state]] is "readable", return ErrorReadableStream(_stream_, _e_).
  1. Return *undefined*.
</pre>

<h4 id="should-readable-stream-pull" aoid="ShouldReadableStreamPull">ShouldReadableStreamPull ( stream )</h4>

<pre is="emu-alg">
  1. If _stream_@[[state]] is "closed" or _stream_@[[state]] is "errored", return *false*.
  1. If _stream_@[[closeRequested]] is *true*, return *false*.
  1. If _stream_@[[started]] is *false*, return *false*.
  1. If IsReadableStreamLocked(_stream_) is *true* and _stream_@[[reader]]@[[readRequests]] is not empty, return *true*.
  1. Let _desiredSize_ be GetReadableStreamDesiredSize(_stream_).
  1. ReturnIfAbrupt(_desiredSize_).
  1. If _desiredSize_ > 0, return *true*.
  1. Return *false*.
</pre>

<h4 id="tee-readable-stream" aoid="TeeReadableStream">TeeReadableStream ( stream, shouldClone )</h4>

This abstract operation is meant to be called from other specifications that may wish to
<a lt="tee a readable stream">tee</a> a given readable stream. Its second argument governs whether or not the data from
the original stream will be <a lt="structured clone">structured cloned</a> before becoming visible in the returned
branches.

<pre is="emu-alg">
  1. Assert: IsReadableStream(_stream_) is *true*.
  1. Assert: Type(_shouldClone_) is Boolean.
  1. Let _reader_ be AcquireReadableStreamReader(_reader_).
  1. ReturnIfAbrupt(_reader_).
  1. Let _teeState_ be Record{[[closedOrErrored]]: *false*, [[canceled1]]: *false*, [[canceled2]]: *false*, [[reason1]]: *undefined*, [[reason2]]: *undefined*, [[promise]]: a new promise}.
  1. Let _pull_ be a new <a>TeeReadableStream pull function</a>.
  1. Set _pull_@[[reader]] to _reader_, _pull_@[[teeState]] to _teeState_, and _pull_@[[shouldClone]] to _shouldClone_.
  1. Let _cancel1_ be a new <a>TeeReadableStream branch 1 cancel function</a>.
  1. Set _cancel1_@[[stream]] to _stream_ and _cancel1_@[[teeState]] to _teeState_.
  1. Let _cancel2_ be a new <a>TeeReadableStream branch 2 cancel function</a>.
  1. Set _cancel2_@[[stream]] to _stream_ and _cancel2_@[[teeState]] to _teeState_.
  1. Let _underlyingSource1_ be ObjectCreate(%ObjectPrototype%).
  1. Perform CreateDataProperty(_underlyingSource1_, "pull", _pull_).
  1. Perform CreateDataProperty(_underlyingSource1_, "cancel", _cancel1_).
  1. Let _branch1_ be Construct(`ReadableStream`, _underlyingSource1_).
  1. Let _underlyingSource2_ be ObjectCreate(%ObjectPrototype%).
  1. Perform CreateDataProperty(_underlyingSource2_, "pull", _pull_).
  1. Perform CreateDataProperty(_underlyingSource2_, "cancel", _cancel2_).
  1. Let _branch2_ be Construct(`ReadableStream`, _underlyingSource2_).
  1. Set _pull_@[[branch1]] to _branch1_.
  1. Set _pull_@[[branch2]] to _branch2_.
  1. Upon rejection of _reader_@[[closedPromise]] with reason _r_,
    1. If _teeState_.[[closedOrErrored]] is *true*, return *undefined*.
    1. Call-with-rethrow ErrorReadableStream(_branch1_, _r_).
    1. Call-with-rethrow ErrorReadableStream(_branch2_, _r_).
    1. Set _teeState_.[[closedOrErrored]] to *true*.
  1. Return «branch1, branch2».
</pre>

<div class="note">
  The given algorithm creates two clones of each chunk, and discards the original, instead of creating one clone and
  giving the original to one branch and the clone to another. This is done to ensure symmetry between the chunks seen
  by each branch; for example, the clone of <code class="lang-javascript">const r = /?:/; r.expando = "!";</code> is
  distinguishable from the original since the clone will not have the expando property.

  However, in specific cases implementations may be able to do something more optimal, without observable consequences.
  For example if each chunk is created by the implementation, and cannot otherwise be modified by the developer, it may
  be possible to ensure the original and its clone are not distinguishable, in which case only one clone operation
  would be necessary. <a href="https://lists.w3.org/Archives/Public/public-webcrypto/2014Mar/0141.html">But, be
  careful!</a>
</div>

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

<pre is="emu-alg">
  1. Let _reader_ be _F_@[[reader]], _branch1_ be _F_@[[branch1]], _branch2_ be _F_@[[branch2]], _teeState_ be _F_@[[teeState]], and _shouldClone_ be _F_@[[shouldClone]].
  1. Return the result of transforming ReadFromReadableStreamReader(_reader_) by a fulfillment handler which takes the argument _result_ and performs the following steps:
    1. Assert: Type(_result_) is Object.
    1. Let _value_ be Get(_result_, "value").
    1. ReturnIfAbrupt(_value_).
    1. Let _done_ be Get(_result_, "done").
    1. ReturnIfAbrupt(_done_).
    1. Assert: Type(_done_) is Boolean.
    1. If _done_ is *true* and _teeState_.[[closedOrErrored]] is *false*,
      1. Call-with-rethrow CloseReadableStream(_branch1_).
      1. Call-with-rethrow CloseReadableStream(_branch2_).
      1. Set _teeState_.[[closedOrErrored]] to *true*.
    1. If _teeState_.[[closedOrErrored]] is *true*, return *undefined*.
    1. If _teeState_.[[canceled1]] is *false*,
      1. Let _value1_ be _value_.
      1. If _shouldClone_ is *true*, set _value1_ to a <a>structured clone</a> of _value_.
      1. Call-with-rethrow EnqueueInReadableStream(_branch1_, _value1_).
    1. If _teeState_.[[canceled2]] is *false*,
      1. Let _value2_ be _value_.
      1. If _shouldClone_ is *true*, set _value2_ to a <a>structured clone</a> of _value_.
      1. Call-with-rethrow EnqueueInReadableStream(_branch2_, _value2_).
</pre>

A <dfn>TeeReadableStream branch 1 cancel function</dfn> is an anonymous built-in function that reacts to the
cancellation of the first of the two branches of the associated tee. Each TeeReadableStream branch 1 cancel function
has \[[stream]] and \[[teeState]] internal slots. When a TeeReadableStream branch 1 cancel function <var>F</var> is
called with argument <var>r</var>, it performs the following steps:

<pre is="emu-alg">
  1. Let _stream_ be _F_@[[stream]] and _teeState_ be _F_@[[teeState]].
  1. Set _teeState_.[[canceled1]] to *true*.
  1. Set _teeState_.[[reason1]] to _r_.
  1. If _teeState_.[[canceled2]] is *true*,
    1. Let _compositeReason_ be CreateArrayFromList(«_teeState_.[[reason1]], _teeState_.[[reason2]]»).
    1. Let _cancelResult_ be CancelReadableStream(_stream_, _compositeReason_).
    1. ReturnIfAbrupt(_cancelResult_).
    1. Resolve _teeState_.[[promise]] with _cancelResult_.
  1. Return _teeState_.[[promise]].
</pre>

A <dfn>TeeReadableStream branch 2 cancel function</dfn> is an anonymous built-in function that reacts to the
cancellation of the second of the two branches of the associated tee. Each TeeReadableStream branch 2 cancel function
has \[[stream]] and \[[teeState]] internal slots. When a TeeReadableStream branch 2 cancel function <var>F</var> is
called with argument <var>r</var>, it performs the following steps:

<pre is="emu-alg">
  1. Let _stream_ be _F_@[[stream]] and _teeState_ be _F_@[[teeState]].
  1. Set _teeState_.[[canceled2]] to *true*.
  1. Set _teeState_.[[reason2]] to _r_.
  1. If _teeState_.[[canceled1]] is *true*,
    1. Let _compositeReason_ be CreateArrayFromList(«_teeState_.[[reason1]], _teeState_.[[reason2]]»).
    1. Let _cancelResult_ be CancelReadableStream(_stream_, _compositeReason_).
    1. ReturnIfAbrupt(_cancelResult_).
    1. Resolve _teeState_.[[promise]] with _cancelResult_.
  1. Return _teeState_.[[promise]].
</pre>

<div class="note">
  The algorithm given here is written such that three new function objects are created for each call to to
  TeeReadableStream. This is just a simplification, and is not actually necessary, since it is unobservable to
  developer code. For example, a self-hosted implementation could optimize by creating a class whose prototype contains
  methods for these functions, with the state stored as instance variables.
</div>

<h2 id="ws">Writable Streams</h2>

<div class="warning">
  Although readable streams have been significantly evolved recently due to implementation progress providing feedback,
  writable streams have not yet caught up to all the discoveries in that space. As such, while the following spec will
  be the basis for a final API, it is expected to change in several important ways before being ready to ship. Please
  follow along on the <a href="https://github.com/whatwg/streams/labels/writable%20streams">writable streams issues
  label</a> for details.
</div>

<h3 id="ws-intro">Using Writable Streams</h3>

<div class="example">
  The usual way to write to a writable stream is to simply <a lt="piping">pipe</a> a <a>readable stream</a> to it.
  This ensures that <a>backpressure</a> is respected, so that if the writable stream's <a>underlying sink</a> is not
  able to accept data as fast as the readable stream can produce it, the readable stream is informed of this and has a
  chance to slow down its data production.

  <pre><code class="lang-javascript">
    readableStream.pipeTo(writableStream)
      .then(() => console.log("All data successfully written!"))
      .catch(e => console.error("Something went wrong!", e));
  </code></pre>
</div>

<div class="example">
  You can also write directly to writable streams using their <code>write()</code> and <code>close()</code> methods.
  Since writable streams queue any incoming writes, and take care internally to forward them to the <a>underlying
  sink</a> in sequence, you can indiscriminately write to a writable stream without much ceremony:

  <pre><code class="lang-javascript">
    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));
  </code></pre>
</div>

<div class="example">
  In the previous example we only paid attention to the success or failure of the entire stream, by looking at the
  promise returned by its <code>close()</code> method. That promise (which can also be accessed using the
  <code>closed</code> getter) will reject if anything goes wrong with the stream—initializing it, writing to it, or
  closing it. And it will fulfill once the stream is successfully closed. Often this is all you care about.

  However, if you care about the success of writing a specific <a>chunk</a>, you can use the promise returned by the
  stream's <code>write()</code> method:

  <pre><code class="lang-javascript">
    writableStream.write("i am a chunk of data")
      .then(() => console.log("chunk successfully written!"))
      .catch(e => console.error(e));
  </code></pre>

  What "success" means is up to a given stream instance (or more precisely, its <a>underlying sink</a>) to decide. For
  example, for a file stream it could simply mean that the OS has accepted the write, and not necessarily that the
  chunk has been flushed to disk.
</div>

<h3 id="ws-class">Class <code>WritableStream</code></h3>

<h4 id="ws-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

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


<pre><code class="lang-javascript">
  class WritableStream {
    constructor(underlyingSink = {}, { size, highWaterMark = 0 } = {})

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

    abort(reason)
    close()
    write(chunk)
  }
</code></pre>

<h4 id="ws-internal-slots">Internal Slots</h4>

Instances of <code>WritableStream</code> are created with the internal slots described in the following table:

<table>
  <thead>
    <tr>
      <th>Internal Slot</th>
      <th>Description (<em>non-normative</em>)</th>
    </tr>
  </thead>
  <tr>
    <td>\[[closedPromise]]
    <td>A promise that becomes fulfilled when the stream becomes <code>"closed"</code>; returned by the
      <code>closed</code> getter
  </tr>
  <tr>
    <td>\[[queue]]
    <td>A List representing the stream's internal queue of pending writes
.  </tr>
  <tr>
    <td>\[[started]]
    <td>A boolean flag indicating whether the <a>underlying sink</a> has finished starting
  </tr>
  <tr>
    <td>\[[startedPromise]]
    <td>A promise storing the result of starting the <a>underlying sink</a>, used to delay actions until that is
      complete
  </tr>
  <tr>
    <td>\[[state]]
    <td>A string containing the stream's current state; returned by the <code>state</code> getter
  </tr>
  <tr>
    <td>\[[storedError]]
    <td>A value indicating how the stream failed, to be given as a failure reason or exception when trying to operate
      on the stream while in the <code>"errored"</code> state
  </tr>
  <tr>
    <td>\[[strategySize]]
    <td>A function supplied to the constructor as part of the stream's <a>queuing strategy</a>, designed to calculate
      the size of <a>chunks</a> written; can be <b>undefined</b> for the default behavior.
  </tr>
  <tr>
    <td>\[[strategyHWM]]
    <td>A number supplied to the constructor as part of the stream's <a>queuing strategy</a>, indicating the point at
      which the stream will apply <a>backpressure</a> to any <a>producers</a>.
  </tr>
  <tr>
    <td>\[[readyPromise]]
    <td>A promise returned by the <code>ready</code> getter
  </tr>
  <tr>
    <td>\[[underlyingSink]]
    <td>An object representation of the stream's <a>underlying sink</a>; also used for the
      <a href="#is-writable-stream">IsWritableStream</a> brand check
  </tr>
  <tr>
    <td>\[[writing]]
    <td>A boolean flag indicating whether the stream is currently writing to the <a>underlying sink</a>, used to
      prevent concurrent such writes
  </tr>
</table>

<h4 id="ws-constructor">new WritableStream(underlyingSink = {}, { size, highWaterMark = 0 } = {})</h4>

<div class="note">
  The <var>underlyingSink</var> object passed to the constructor can implement any of the following methods to govern
  how the constructed stream instance behaves:

  <ul>
    <li> <code>start(error)</code> is called immediately, and should perform any actions necessary to acquire
      access to the <a>underlying sink</a>. If this process is asynchronous, it can return a promise to signal success
      or failure.
    <li> <code>write(chunk)</code> is called when a new <a>chunk</a> of data is ready to be written to the
      <a>underlying sink</a>. It can return a promise to signal success or failure of the write operation. The stream
      implementation guarantees that this method will be called only after previous writes have succeeded, and never
      after <code>close</code> or <code>abort</code> is called.
    <li> <code>close()</code> is called after the producer signals that they are done writing chunks to the stream, and
      all queued-up writes successfully complete. It should perform any actions necessary to finalize writes to the
      <a>underlying sink</a>, and release access to it. If this process is asynchronous, it can return a promise to
      signal success or failure. The stream implementation guarantees that this method will be called only after all
      queued-up writes have succeeded.
    <li> <code>abort(reason)</code> is called when the producer signals they wish to abruptly close the stream
      and put it in an <code>"errored"</code> state. It should clean up any held resources, much like
      <code>close</code>, but perhaps with some custom handling. Unlike <code>close</code>, <var>abort</var> will be
      called even if writes are queued up; those <a>chunks</a> will be thrown away. If this process is asynchronous, it
      can return a promise to signal success or failure. If no abort method is passed, by default the
      <code>close</code> method will be called instead.
  </ul>

  The constructor also accepts a second argument containing the <a>queuing strategy</a> object with
  two properties: a non-negative number <code>highWaterMark</code>, and a function <code>size(chunk)</code>. The
  supplied <code>strategy</code> could be an instance of the built-in <code>CountQueuingStrategy</code> or
  <code>ByteLengthQueuingStrategy</code> classes, or it could be custom. If no strategy is supplied, the default
  behavior will be the same as a <code>CountQueuingStrategy</code> with a <a>high water mark</a> of 0.
</div>

<div class="note">
  Due to the way writable streams asynchronously close, it is possible for both <code>close</code> and
  <code>abort</code> to be called, in cases where the <a>producer</a> aborts the stream while it is in the
  <code>"closing"</code> state. Notably, since a stream always spends at least one turn in the <code>"closing"</code>
  state, code like <code>ws.close(); ws.abort(...);</code> will cause both to be called, even if the <code>close</code>
  method itself has no asynchronous behavior. A well-designed <a>underlying sink</a> object should be able to deal with
  this.
</div>

<pre is="emu-alg">
  1. Set *this*@[[underlyingSink]] to _underlyingSink_.
  1. Set *this*@[[closedPromise]] to a new promise.
  1. Set *this*@[[readyPromise]] to a new promise resolved with *undefined*.
  1. Set *this*@[[queue]] to a new empty List.
  1. Set *this*@[[state]] to "writable".
  1. Set *this*@[[started]] and *this*@[[writing]] to *false*.
  1. Let _normalizedStrategy_ be ValidateAndNormalizeQueuingStrategy(_size_, _highWaterMark_).
  1. Set *this*@[[strategySize]] to _normalizedStrategy_.[[size]] and *this*@[[strategyHWM]] to _normalizedStrategy_.[[highWaterMark]].
  1. Call-with-rethrow SyncWritableStreamStateWithQueue(*this*).
  1. Let _error_ be a new <a><code>WritableStream</code> error function</a>.
  1. Set _error_@[[stream]] to *this*.
  1. Let _startResult_ be InvokeOrNoop(_underlyingSink_, "start", «_error_»).
  1. ReturnIfAbrupt(_startResult_).
  1. Set *this*@[[startedPromise]] to the result of resolving _startResult_ as a promise.
    1. Upon fulfillment,
      1. Set *this*@[[started]] to *true*.
      1. Set *this*@[[startedPromise]] to *undefined*.
    1. Upon rejection with reason _r_, return ErrorWritableStream(*this*, _r_).
</pre>

A <dfn><code>WritableStream</code> error function</dfn> is an anonymous built-in function that is used to allow
<a>underlying sinks</a> to error their associated writable stream. Each <code>WritableStream</code> error function has
a \[[stream]] internal slot. When a <code>WritableStream</code> error function <var>F</var> is called with argument
<var>e</var>, it performs the following steps:

<pre is="emu-alg">
  1. Let _stream_ be _F_@[[stream]].
  1. Return ErrorWritableStream(_stream_, _e_).
</pre>

<h4 id="ws-prototype">Properties of the <code>WritableStream</code> Prototype</h4>

<h5 id="ws-closed">get closed</h5>

<div class="note">
  The <code>closed</code> getter returns a promise that will be fulfilled when the stream becomes closed, or rejected
  if it ever errors.
</div>

<pre is="emu-alg">
  1. If IsWritableStream(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. Return *this*@[[closedPromise]].
</pre>

<h5 id="ws-ready">get ready</h5>

<div class="note">
  The <code>ready</code> getter returns a promise that will be fulfilled when the stream transitions away from the
  <code>"waiting"</code> state to any other state. Once the stream transitions back to <code>"waiting"</code>, the
  getter will return a new promise that stays pending until the next state transition.

  In essence, this promise gives a signal as to when any backpressure has let up (or that the stream has been closed
  or errored).
</div>

<pre is="emu-alg">
  1. If IsWritableStream(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. Return *this*@[[readyPromise]].
</pre>

<h5 id="ws-state">get state</h5>

<div class="note">
  The <code>state</code> getter returns the state of the stream, which will be one of the following:

  <dl>
    <dt><code>"waiting"</code>
    <dd>The stream's internal queue is full; that is, the stream is
      exerting <a>backpressure</a>. Use <code>.ready</code> to be notified of when the pressure subsides.

    <dt><code>"writable"</code>
    <dd>The stream's internal queue is not full; call <code>.write()</code> until backpressure is exerted.

    <dt><code>"closing"</code>
    <dd>The stream's <code>.close()</code> method has been called, and a command to close is in the queue or
      being processed by the <a>underlying sink</a>; attempts to write will now fail.

    <dt><code>"closed"</code>
    <dd>The <a>underlying sink</a> has been closed; writing is no longer possible.

    <dt><code>"errored"</code>
    <dd>An error occurred interacting with the <a>underlying sink</a> or the stream has been aborted, so the stream is
      now dead.
  </dl>
</div>

<pre is="emu-alg">
  1. If IsWritableStream(*this*) is *false*, throw a *TypeError* exception.
  1. Return *this*@[[state]].
</pre>

<h5 id="ws-abort">abort(reason)</h5>

<div class="note">
  The <code>abort</code> method signals that the producer can no longer successfully write to the stream and it should
  be immediately moved to an <code>"errored"</code> state, with any queued-up writes discarded. This will also execute
  any abort mechanism of the <a>underlying sink</a>.
</div>

<pre is="emu-alg">
  1. If IsWritableStream(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. If *this*@[[state]] is "closed", return a new promise resolved with *undefined*.
  1. If *this*@[[state]] is "errored", return a new promise rejected with *this*@[[storedError]].
  1. Call-with-rethrow ErrorWritableStream(*this*, _reason_).
  1. Let _sinkAbortPromise_ be PromiseInvokeOrFallbackOrNoop(*this*@[[underlyingSink]], "abort", «_reason_», "close", «»).
  1. Return the result of transforming _sinkAbortPromise_ by a fulfillment handler that returns *undefined*.
</pre>

<h5 id="ws-close">close()</h5>

<div class="note">
  The <code>close</code> method signals that the producer is done writing chunks to the stream and wishes to move the
  stream to a <code>"closed"</code> state. This queues an action to close the stream, such that once any currently
  queued-up writes complete, the close mechanism of the <a>underlying sink</a> will execute, releasing any held
  resources. In the meantime, the stream will be in a <code>"closing"</code> state.
</div>

<pre is="emu-alg">
  1. If IsWritableStream(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. If *this*@[[state]] is "closing" or "closed", return a promise rejected with a *TypeError* exception.
  1. If *this*@[[state]] is "errored", return a promise rejected with *this*@[[storedError]].
  1. If *this*@[[state]] is "waiting", resolve *this*@[[readyPromise]] with *undefined*.
  1. Set *this*@[[state]] to "closing".
  1. Call-with-rethrow EnqueueValueWithSize(*this*@[[queue]], "close", *0*).
  1. Call-with-rethrow CallOrScheduleWritableStreamAdvanceQueue(*this*).
  1. Return *this*@[[closedPromise]].
</pre>

<h5 id="ws-write">write(chunk)</h5>

<div class="note">
  The <code>write</code> method adds a write to the stream's internal queue, instructing the stream to
  write the given <a>chunk</a> of data to the <a>underlying sink</a> once all other pending writes have finished
  successfully. It returns a promise that will be fulfilled or rejected depending on the success or failure of writing
  the chunk to the underlying sink.

  The impact of enqueuing this chunk will be immediately reflected in the stream's <code>state</code> property; in
  particular, if the internal queue is now full according to the stream's <a>queuing strategy</a>, the stream will
  exert backpressure by changing its state to <code>"waiting"</code>.
</div>

<pre is="emu-alg">
  1. If IsWritableStream(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. If *this*@[[state]] is "closing" or "closed", return a promise rejected with a *TypeError* exception.
  1. If *this*@[[state]] is "errored", return a promise rejected with *this*@[[storedError]].
  1. Assert: *this*@[[state]] is either "waiting" or "writable".
  1. Let _chunkSize_ be *1*.
  1. If *this*@[[strategySize]] is not *undefined*, then
    1. Set _chunkSize_ to Call(*this*@[[strategySize]], *undefined*, «‍_chunk_»).
    1. If _chunkSize_ is an abrupt completion,
      1. Call-with-rethrow ErrorWritableStream(*this*, _chunkSize_.[[value]]).
      1. Return a new promise rejected with _chunkSize_.[[value]].
    1. Set _chunkSize_ to _chunkSize_.[[value]].
  1. Let _promise_ be a new promise.
  1. Let _writeRecord_ be Record{[[promise]]: _promise_, [[chunk]]: _chunk_}.
  1. Let _enqueueResult_ be EnqueueValueWithSize(*this*@[[queue]], _writeRecord_, _chunkSize_).
  1. If _enqueueResult_ is an abrupt completion,
    1. Call-with-rethrow ErrorWritableStream(*this*, _enqueueResult_.[[value]]).
    1. Return a new promise rejected with _enqueueResult_.[[value]].
  1. Let _syncResult_ be SyncWritableStreamStateWithQueue(*this*).
  1. If _syncResult_ is an abrupt completion,
    1. Call-with-rethrow ErrorWritableStream(*this*, _syncResult_.[[value]]).
    1. Return _promise_.
  1. Call-with-rethrow CallOrScheduleWritableStreamAdvanceQueue(*this*).
  1. Return _promise_.
</pre>

<h3 id="ws-abstract-ops">Writable Stream Abstract Operations</h3>

<h4 id="call-or-schedule-writable-stream-advance-queue" aoid="CallOrScheduleWritableStreamAdvanceQueue">CallOrScheduleWritableStreamAdvanceQueue ( stream )</h4>

<pre is="emu-alg">
  1. If _stream_@[[started]] is *false*, then
    1. Upon fulfillment of _stream_@[[startedPromise]], call-with-rethrow WritableStreamAdvanceQueue(_stream_).
    1. Return *undefined*.
  1. Otherwise, return WritableStreamAdvanceQueue(_stream_).
</pre>

<h4 id="close-writable-stream" aoid="CloseWritableStream">CloseWritableStream ( stream )</h4>

<pre is="emu-alg">
  1. Assert: _stream_@[[state]] is "closing".
  1. Let _sinkClosePromise_ be PromiseInvokeOrNoop(_stream_@[[underlyingSink]], "close").
    1. Upon fulfillment,
      1. If _stream_@[[state]] is "errored", return.
      1. Assert: _stream_@[[state]] is "closing".
      1. Resolve _stream_@[[closedPromise]] with *undefined*.
      1. Set _stream_@[[state]] to "closed".
    1. Upon rejection with reason _r_, return ErrorWritableStream(_stream_, _r_).
  1. Return *undefined*.
</pre>

<h4 id="error-writable-stream" aoid="ErrorWritableStream">ErrorWritableStream ( stream, e )</h4>

<pre is="emu-alg">
  1. If _stream_@[[state]] is "closed" or "errored", return *undefined*.
  1. Repeat while _stream_@[[queue]] is not empty:
    1. Let _writeRecord_ be DequeueValue(_stream_@[[queue]]).
    1. If _writeRecord_ is not "close", reject _writeRecord_.[[promise]] with _e_.
  1. Set _stream_@[[storedError]] to _e_.
  1. If _stream_@[[state]] is "waiting", resolve _stream_@[[readyPromise]] with *undefined*.
  1. Reject _stream_@[[closedPromise]] with _e_.
  1. Set _stream_@[[state]] to "errored".
  1. Return *undefined*.
</pre>

<h4 id="is-writable-stream" aoid="IsWritableStream">IsWritableStream ( x )</h4>

<pre is="emu-alg">
  1. If Type(_x_) is not Object, return *false*.
  1. If _x_ does not have a [[underlyingSink]] internal slot, return *false*.
  1. Return *true*.
</pre>

<h4 id="sync-writable-stream-state-with-queue" aoid="SyncWritableStreamStateWithQueue">SyncWritableStreamStateWithQueue ( stream )</h4>

<pre is="emu-alg">
  1. If _stream_@[[state]] is "closing", return *undefined*.
  1. Assert: _stream_@[[state]] is either "writable" or "waiting".
  1. Let _queueSize_ be GetTotalQueueSize(_stream_@[[queue]]).
  1. ReturnIfAbrupt(_queueSize_).
  1. Let _shouldApplyBackpressure_ be *true* if _queueSize_ > _stream_@[[strategyHWM]], and *false* otherwise.
  1. If _shouldApplyBackpressure_ is *true* and _stream_@[[state]] is "writable", then
    1. Set _stream_@[[state]] to "waiting".
    1. Set _stream_@[[readyPromise]] to a new promise.
  1. If _shouldApplyBackpressure_ is *false* and _stream_@[[state]] is "waiting", then
    1. Set _stream_@[[state]] to "writable".
    1. Resolve _stream_@[[readyPromise]] with *undefined*.
  1. Return *undefined*.
</pre>

<h4 id="writable-stream-advance-queue" aoid="WritableStreamAdvanceQueue">WritableStreamAdvanceQueue ( stream )</h4>

<pre is="emu-alg">
  1. If _stream_@[[queue]] is empty, or _stream_@[[writing]] is *true*, return *undefined*.
  1. Let _writeRecord_ be PeekQueueValue(_stream_@[[queue]]).
  1. If _writeRecord_ is "close", then
    1. Assert: _stream_@[[state]] is "closing".
    1. DequeueValue(_stream_@[[queue]]).
    1. Assert: _stream_@[[queue]] is now empty.
    1. Return CloseWritableStream(_stream_).
  1. Set _stream_@[[writing]] to *true*.
  1. Let _writeResult_ be PromiseInvokeOrNoop(_stream_@[[underlyingSink]], "write", «_writeRecord_.[[chunk]]»).
  1. Upon fulfillment of _writeResult_,
    1. If _stream_@[[state]] is "errored", return.
    1. Set _stream_@[[writing]] to *false*.
    1. Resolve _writeRecord_.[[promise]] with *undefined*.
    1. DequeueValue(_stream_@[[queue]]).
    1. Let _syncResult_ be SyncWritableStreamStateWithQueue(_stream_).
    1. If _syncResult_ is an abrupt completion, then return ErrorWritableStream(_stream_, ‍_syncResult_.[[value]]).
    1. Otherwise, return WritableStreamAdvanceQueue(_stream_).
  1. Upon rejection of _writeResult_ with reason _r_, return ErrorWritableStream(_stream_, _r_).
</pre>

<h2 id="ts">Transform Streams</h2>

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
<a href="https://github.com/whatwg/streams/blob/master/reference-implementation/lib/transform-stream.js">reference-implementation/lib/transform-stream.js</a>.


<h2 id="other-stuff">Other Stream APIs and Operations</h2>

<h3 id="blqs-class">Class <code>ByteLengthQueuingStrategy</code></h3>

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

<div class="example">
  When creating a <a>readable stream</a> or <a>writable stream</a>, you can supply a byte-length queuing strategy
  directly:

  <pre><code class="lang-javascript">
    const stream = new ReadableStream(
      { ... },
      new ByteLengthQueuingStrategy({ highWaterMark: 16 * 1024 })
    );
  </code></pre>

  In this case, 16 KiB worth of <a>chunks</a> can be enqueued by the readable stream's <a>underlying source</a> before
  the readable stream implementation starts sending <a>backpressure</a> signals to the underlying source.

  <pre><code class="lang-javascript">
    const stream = new WritableStream(
      { ... },
      new ByteLengthQueuingStrategy({ highWaterMark: 32 * 1024 })
    );
  </code></pre>

  In this case, 32 KiB worth of <a>chunks</a> can be accumulated in the writable stream's internal queue, waiting for
  previous writes to the <a>underlying sink</a> to finish, before the writable stream starts sending
  <a>backpressure</a> signals to any <a>producers</a>.
</div>

<h4 id="blqs-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

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

<pre><code class="lang-javascript">
  class ByteLengthQueuingStrategy {
    constructor({ highWaterMark })
    size(chunk)
  }
</code></pre>

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

<h4 id="blqs-constructor">new ByteLengthQueuingStrategy({ highWaterMark })</h4>

<div class="note">
  The constructor takes a nonnegative number for the high-water mark, and stores it on the object as a property.
</div>

<pre is="emu-alg">
  1. CreateDataProperty(*this*, "highWaterMark", _highWaterMark_).
</pre>

<h4 id="blqs-prototype">Properties of the <code>ByteLengthQueuingStrategy</code> Prototype</h4>

<h5 id="blqs-size">size(chunk)</h5>

<div class="note">
  The <code>size</code> method returns the given chunk's <code>byteLength</code> property. (If the chunk doesn't have
  one, it will return <b>undefined</b>, causing the stream using this strategy to error.)

  This method is intentionally generic; it does not require that its <b>this</b> value be a
  <code>ByteLengthQueuingStrategy</code> object.
</div>

<pre is="emu-alg">
  1. Return Get(_chunk_, "byteLength").
</pre>

<h3 id="cqs-class">Class <code>CountQueuingStrategy</code></h3>

A common <a>queuing strategy</a> 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.

<div class="example">
  When creating a <a>readable stream</a> or <a>writable stream</a>, you can supply a count queuing strategy directly:

  <pre><code class="lang-javascript">
    const stream = new ReadableStream(
      { ... },
      new CountQueuingStrategy({ highWaterMark: 10 })
    );
  </code></pre>

  In this case, 10 <a>chunks</a> (of any kind) can be enqueued by the readable stream's <a>underlying source</a> before
  the readable stream implementation starts sending <a>backpressure</a> signals to the underlying source.

  <pre><code class="lang-javascript">
    const stream = new WritableStream(
      { ... },
      new CountQueuingStrategy({ highWaterMark: 5 })
    );
  </code></pre>

  In this case, five <a>chunks</a> (of any kind) can be accumulated in the writable stream's internal queue, waiting
  for previous writes to the <a>underlying sink</a> to finish, before the writable stream starts sending
  <a>backpressure</a> signals to any <a>producers</a>.
</div>

<h4 id="cqs-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

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

<pre><code class="lang-javascript">
  class CountQueuingStrategy {
    constructor({ highWaterMark })
    size()
  }
</code></pre>

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

<h4 id="cqs-constructor">new CountQueuingStrategy({ highWaterMark })</h4>

<div class="note">
  The constructor takes a nonnegative number for the high-water mark, and stores it on the object as a property.
</div>

<pre is="emu-alg">
  1. CreateDataProperty(*this*, "highWaterMark", _highWaterMark_).
</pre>

<h4 id="cqs-prototype">Properties of the <code>CountQueuingStrategy</code> Prototype</h4>

<h5 id="cqs-size">size()</h5>

<div class="note">
  The <code>size</code> method returns one always, so that the total queue size is a count of the number of chunks in
  the queue.

  This method is intentionally generic; it does not require that its <b>this</b> value be a
  <code>CountQueuingStrategy</code> object.
</div>

<pre is="emu-alg">
  1. Return *1*.
</pre>

<h3 id="queue-with-sizes">Queue-with-Sizes Operations</h3>

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.

<h4 id="dequeue-value" aoid="DequeueValue">DequeueValue ( queue )</h4>

<pre is="emu-alg">
  1. Assert: _queue_ is not empty.
  1. Let _pair_ be the first element of queue.
  1. Remove _pair_ from _queue_, shifting all other elements downward (so that the second becomes the first, and so on).
  1. Return _pair_.[[value]].
</pre>

<h4 id="enqueue-value-with-size" aoid="EnqueueValueWithSize">EnqueueValueWithSize ( queue, value, size )</h4>

<pre is="emu-alg">
  1. Let _size_ be ToNumber(_size_).
  1. ReturnIfAbrupt(_size_).
  1. If _size_ is *NaN*, *+∞*, or *−∞*, throw a *RangeError* exception.
  1. Append Record{[[value]]: _value_, [[size]]: _size_} as the last element of _queue_.
</pre>

<h4 id="get-total-queue-size" aoid="GetTotalQueueSize">GetTotalQueueSize ( queue )</h4>

<pre is="emu-alg">
  1. Let _totalSize_ be *0*.
  1. Repeat for each Record{[[value]], [[size]]} _pair_ that is an element of _queue_,
    1. Assert: _pair_.[[size]] is a finite, non-*NaN* number.
    1. Add _pair_.[[size]] to _totalSize_.
  1. Return _totalSize_.
</pre>

<h4 id="peek-queue-value" aoid="PeekQueueValue">PeekQueueValue ( queue )</h4>

<pre is="emu-alg">
  1. Assert: _queue_ is not empty.
  1. Let _pair_ be the first element of _queue_.
  1. Return _pair_.[[value]].
</pre>

<h3 id="misc-abstract-ops">Miscellaneous Operations</h3>

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

<h4 id="get-viewed-array-buffer" aoid="GetViewedArrayBuffer">GetViewedArrayBuffer ( O )</h4>

<pre is="emu-alg">
  1. Return the result of calling the get accessor function of %TypedArray%.prototype.buffer on _O_ passing no arguments.
</pre>

<h4 id="get-byte-length" aoid="GetByteLength">GetByteLength ( O )</h4>

<pre is="emu-alg">
  1. Return the result of calling the get accessor function of %TypedArray%.prototype.byteLength on _O_ passing no arguments.
</pre>

<h4 id="get-byte-offset" aoid="GetByteOffset">GetByteOffset ( O )</h4>

<pre is="emu-alg">
  1. Return the result of calling the get accessor function of %TypedArray%.prototype.byteOffset on _O_ passing no arguments.
</pre>

<h4 id="invoke-or-noop" aoid="InvokeOrNoop">InvokeOrNoop ( O, P, args )</h4>

<div class="note">
  InvokeOrNoop is a slight modification of the [[!ECMASCRIPT]]
  <a href="https://people.mozilla.org/~jorendorff/es6-draft.html#sec-invoke">Invoke</a> abstract operation to return
  <b>undefined</b> when the method is not present.
</div>

<pre is="emu-alg">
  1. Assert: _P_ is a valid property key.
  1. If _args_ was not passed, let _args_ be a new empty List.
  1. Let _method_ be GetV(_O_, _P_).
  1. ReturnIfAbrupt(_method_).
  1. If _method_ is *undefined*, return *undefined*.
  1. Return Call(_method_, _O_, _args_).
</pre>

<h4 id="promise-invoke-or-fallback-or-noop" aoid="PromiseInvokeOrFallbackOrNoop">PromiseInvokeOrFallbackOrNoop ( O, P1, args1, P2, args2 )</h4>

<div class="note">
  PromiseInvokeOrFallbackOrNoop is a specialized version of
  <a href="http://www.w3.org/2001/tag/doc/promises-guide#promise-calling">promise-calling</a> that works on methods,
  calls a fallback method if the first method is not present, and returns a promise for <b>undefined</b> when neither
  method is not present.
</div>

<pre is="emu-alg">
  1. Assert: _P1_ is a valid property key.
  1. Assert: _P2_ is a valid property key.
  1. Let _method_ be GetV(_O_, _P1_).
  1. If _method_ is an abrupt completion, return a new promise rejected with _method_.[[value]].
  1. Let _method_ be _method_.[[value]].
  1. If _method_ is *undefined*, return PromiseInvokeOrNoop(_O_, _P2_, _args2_).
  1. Let _returnValue_ be Call(_method_, _O_, _args1_).
  1. If _returnValue_ is an abrupt completion, return a new promise rejected with _returnValue_.[[value]].
  1. Otherwise, return a new promise resolved with _returnValue_.[[value]].
</pre>

<h4 id="promise-invoke-or-noop" aoid="PromiseInvokeOrNoop">PromiseInvokeOrNoop ( O, P, args )</h4>

<div class="note">
  PromiseInvokeOrNoop is a specialized version of
  <a href="http://www.w3.org/2001/tag/doc/promises-guide#promise-calling">promise-calling</a> that both works on
  methods and returns a promise for <b>undefined</b> when the method is not present.
</div>

<pre is="emu-alg">
  1. Assert: _P_ is a valid property key.
  1. If _args_ was not passed, let _args_ be a new empty List.
  1. Let _method_ be GetV(_O_, _P_).
  1. If _method_ is an abrupt completion, return a new promise rejected with _method_.[[value]].
  1. Let _method_ be _method_.[[value]].
  1. If _method_ is *undefined*, return a new promise resolved with *undefined*.
  1. Let _returnValue_ be Call(_method_, _O_, _args_).
  1. If _returnValue_ is an abrupt completion, return a new promise rejected with _returnValue_.[[value]].
  1. Otherwise, return a new promise resolved with _returnValue_.[[value]].
</pre>

<h4 id="validate-and-normalize-queuing-strategy" aoid="ValidateAndNormalizeQueuingStrategy">ValidateAndNormalizeQueuingStrategy ( size, highWaterMark )</h4>

<pre is="emu-alg">
  1. If _size_ is not *undefined* and IsCallable(_size_) is *false*, throw a *TypeError* exception.
  1. Set _highWaterMark_ to ToNumber(_highWaterMark_).
  1. ReturnIfAbrupt(_highWaterMark_).
  1. If _highWaterMark_ is *NaN*, throw a *TypeError* exception.
  1. If _highWaterMark_ < *0*, throw a *RangeError* exception.
  1. Return Record{[[size]]: size, [[highWaterMark]]: highWaterMark}.
</pre>

<h2 id="rbs">Readable Byte Streams</h2>

<div class="warning">
  Readable byte streams are still a work in progress. We are currently trying to ensure that their public API is stable,
  as well as any operations needed to interface with them from other specifications. However, we're much less sure
  about the details of their internal algorithms, as well as the APIs necessary to construct readable byte streams
  (viz. the underlying byte source API and the controller API). Please follow along in
  <a href="https://github.com/whatwg/streams/issues/300">the tracking issue</a>, as well as other issues under the
  <a href="https://github.com/whatwg/streams/labels/byte%20streams">byte streams issues label</a>.
</div>

<h3 id="rbs-intro">Using Readable Byte Streams</h3>

<div class="note">
  Unlike the readable stream, the readable byte stream provides multiple different kinds of readers. One is the
  <code>ReadableByteStreamReader</code>. It has the same methods as the <code>ReadableStreamReader</code>, and can be
  read in the same manner as the <code>ReadableStreamReader</code>. Other readers dedicated for the readable byte stream
  are to be introduced into this spec.
</div>

<h3 id="rbs-class">Class <code>ReadableByteStream</code></h3>

<h4 id="rbs-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

If one were to write the <code>ReadableByteStream</code> class in something close to the syntax of [[!ECMASCRIPT]], it
would look like

<pre><code class="lang-javascript">
  class ReadableByteStream {
    constructor(underlyingByteSource = {})

    cancel(reason)
    getByobReader()
    getReader()
    pipeThrough({ writable, readable }, options)
    pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})
    tee()
  }
</code></pre>

<h4 id="rbs-internal-slots">Internal Slots</h4>

Instances of <code>ReadableByteStream</code> are created with the internal slots described in the following table:

<table>
  <thead>
    <tr>
      <th>Internal Slot</th>
      <th>Description (<em>non-normative</em>)</th>
    </tr>
  </thead>
  <tr>
    <td>\[[controller]]
    <td>A <a href="#rbs-controller-class"><code>ReadableByteStreamController</code></a> created with the ability to
      control the state of this stream; also used for the
      <a href="#is-readable-byte-stream">IsReadableByteStream</a> brand check
  </tr>
  <tr>
    <td>\[[reader]]
    <td>A <a href="#byte-reader-class"><code>ReadableByteStreamReader</code></a> instance, if the stream is <a>locked to
      a reader</a>, or <b>undefined</b> if it is not
  </tr>
  <tr>
    <td>\[[state]]
    <td>A string containing the stream's current state, used internally; one of <code>"readable"</code>,
      <code>"closed"</code>, or <code>"errored"</code>
  </tr>
  <tr>
    <td>\[[storedError]]
    <td>A value indicating how the stream failed, to be given as a failure reason or exception when trying to operate
      on an errored stream
  </tr>
</table>

<h4 id="rbs-constructor">new ReadableByteStream(underlyingByteSource = {})</h4>

<pre is="emu-alg">
  1. Set *this*@[[state]] to "readable".
  1. Set *this*@[[reader]] and *this*@[[storedError]] to *undefined*.
  1. Set *this*@[[controller]] to Construct(`ReadableByteStreamController`, «*this*, _underlyingByteSource_»).
</pre>

<h4 id="rbs-prototype">Properties of the <code>ReadableByteStream</code> Prototype</h4>

<h5 id="rbs-cancel">cancel(reason)</h5>

<pre is="emu-alg">
  1. If IsReadableByteStream(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. If IsReadableByteStreamLocked(*this*) is *true*, return a promise rejected with a *TypeError* exception.
  1. Return CancelReadableByteStream(*this*, _reason_).
</pre>

<h5 id="rbs-get-byob-reader">getByobReader()</h5>

<pre is="emu-alg">
  1. If IsReadableByteStream(*this*) is *false*, throw a *TypeError* exception.
  1. Return AcquireReadableByteStreamByobReader(*this*).
</pre>

<h5 id="rbs-get-reader">getReader()</h5>

<pre is="emu-alg">
  1. If IsReadableByteStream(*this*) is *false*, throw a *TypeError* exception.
  1. Return AcquireReadableByteStreamReader(*this*).
</pre>

<h5 id="rbs-pipe-through">pipeThrough({ writable, readable }, options)</h5>

<pre is="emu-alg">
  1. Call-with-rethrow Invoke(*this*, "pipeTo", «_writable_, _options_»).
  1. Return _readable_.
</pre>

<h5 id="rbs-pipe-to">pipeTo(dest, { preventClose, preventAbort, preventCancel } = {})</h5>

The <code>pipeTo</code> method is one of the more complex methods, and is undergoing some revision and edge-case
bulletproofing before we write it up in prose.

For now, please consider the reference implementation normative:
<a href="https://github.com/whatwg/streams/blob/master/reference-implementation/lib/readable-stream.js">reference-implementation/lib/readable-stream.js</a>,
look for the <code>pipeTo</code> method.

<h5 id="rbs-tee">tee()</h5>

<pre is="emu-alg">
  1. If IsReadableByteStream(*this*) is *false*, throw a *TypeError* exception.
  1. Let _branches_ be TeeReadableByteStream(*this*, *false*).
  1. ReturnIfAbrupt(_branches_).
  1. Return CreateArrayFromList(_branches_).
</pre>

<h3 id="rbs-controller-class" lt="ReadableByteStreamController">Class <code>ReadableByteStreamController</code></h3>

The <code>ReadableByteStreamController</code> class has methods that allow control of a <code>ReadableStream</code>'s
state. When constructing a <code>ReadableByteStream</code>, the <a>underlying byte source</a> is given a
corresponding <code>ReadableByteStreamController</code> instance to manipulate.

<h4 id="rbs-controller-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

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

<pre><code class="lang-javascript">
  class ReadableStreamController {
    constructor(controlledReadableByteStream, underlyingByteSource)

    close()
    enqueue(chunk)
    error(e)
    respond(chunk)
  }
</code></pre>

<h4 id="rbs-controller-internal-slots">Internal Slots</h4>

Instances of <code>ReadableByteStreamController</code> are created with the internal slots described in the following
table:

<table>
  <thead>
    <tr>
      <th>Internal Slot</th>
      <th>Description (<em>non-normative</em>)</th>
    </tr>
  </thead>
  <tr>
    <td>\[[consumedBytesOfQueueHead]]
    <td>The number of bytes of the first element in \[[queue]] which are already consumed
  </tr>
  <tr>
    <td>\[[filledBytesOfPendingViewsHead]]
    <td>The number of bytes of the first element in \[[pendingView]] which are already filled
  </tr>
  <tr>
    <td>\[[controlledReadableByteStream]]
    <td>The <code>ReadableByteStream</code> instance controlled
  </tr>
  <tr>
    <td>\[[pendingViews]]
    <td>A List of objects implementing the ArrayBufferView type provided by the consumer via
      <code>ReadableByteStreamByobReader</code>; used only when the provided underlying byte source doesn't have the
      "pullInto" method.
  </tr>
  <tr>
    <td>\[[queue]]
    <td>A List representing the stream's internal queue of pending writes
  </tr>
  <tr>
    <td>\[[underlyingByteSource]]
    <td>An object representation of the stream's <a>underlying byte source</a>
  </tr>
</table>

<h4 id="rbs-controller-constructor">new ReadableByteStreamController(controlledReadableByteStream, underlyingByteSource)</h4>

<pre is="emu-alg">
  1. If IsReadableByteStream(_controlledReadableByteStream_) is *false*, throw a *TypeError* exception.
  1. If _controlledReadableByteStream_@[[controller]] is not *undefined*, throw a *TypeError* exception.
  1. Let _pullFunction_ be GetV(_underlyingByteSource_, "pull").
  1. ReturnIfAbrupt(_pullFunction_).
  1. If _pullFunction_ is not *undefined*,
    1. If IsCallable(_pullFunction_) is *false*, throw a *TypeError* exception.
  1. Let _pullIntoFunction_ be GetV(_underlyingByteSource_, "pullInto").
  1. ReturnIfAbrupt(_pullIntoFunction_).
  1. If _pullIntoFunction_ is not *undefined*,
    1. If IsCallable(_pullIntoFunction_) is *false*, throw a *TypeError* exception.
  1. Set *this*@[[consumedBytesOfQueueHead]] to *0*;
  1. Set *this*@[[filledBytesOfPendingViewsHead]] to *0*;
  1. Set *this*@[[controlledReadableByteStream]] to _controlledReadableByteStream_.
  1. Set *this*@[[pendingViews]] be a new empty List.
  1. Set *this*@[[queue]] be a new empty List.
  1. Set *this*@[[underlyingByteSource]] to _underlyingByteSource_.
  1. Let _startResult_ be InvokeOrNoop(_underlyingByteSource_, "start", «*this*»).
  1. ReturnIfAbrupt(startResult).
</pre>

<h4 id="rbs-controller-prototype">Properties of the <code>ReadableByteStreamController</code> Prototype</h4>

<h5 id="rbs-controller-close">close()</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamController(*this*) is *false*, throw a *TypeError* exception.
  1. Let _stream_ be *this*@[[controlledReadableByteStream]].
  1. If *this*@[[closeRequested]] is *true*, throw a *TypeError* exception.
  1. If _stream_@[[state]] is not "readable", throw a *TypeError* exception.
  1. If *this*@[[queue]] is empty,
    1. If *this*@[[filledBytesOfPendingViewsHead]] is not equal to *0*, throw a *TypeError* exception.
    1. CloseReadableByteStream(_stream_).
    1. Return *undefined*.
  1. Set *this*@[[closeRequested]] to *true*.
  1. Return *undefined*.
</pre>

<h5 id="rbs-controller-enqueue">enqueue(chunk)</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamController(*this*) is *false*, throw a *TypeError* exception.
  1. Let _stream_ be *this*@[[controlledReadableByteStream]].
  1. If *this*@[[closeRequested]] is *true*, throw a *TypeError* exception.
  1. If _stream_@[[state]] is not "readable", throw a *TypeError* exception.
  1. Let _reader_ be _stream_@[[reader]].
  1. If _reader_ is *undefined*,
    1. EnqueueInReadableByteStreamController(*this*, _chunk_).
  1. Otherwise,
    1. If _reader_ has a [[readRequests]] internal slot,
      1. If _reader_@[[readRequests]] is empty,
        1. EnqueueInReadableByteStreamController(*this*, _chunk_).
      1. Otherwise,
        1. Assert: *this*@[[queue]] is empty.
        1. RespondToReadableByteStreamReaderReadRequest(_stream_, _chunk_).
    1. Otherwise, _reader_ must have a [[readIntoRequests]] internal slot so,
      1. If _reader_@[[readIntoRequests]] is empty,
        1. EnqueueInReadableByteStreamController(*this*, _chunk_).
      1. Otherwise, throw a *TypeError* exception.
  1. Return *undefined*.
</pre>

<h5 id="rbs-controller-error">error(e)</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamController(*this*) is *false*, throw a *TypeError* exception.
  1. Let _stream_ be *this*@[[controlledReadableByteStream]].
  1. If _stream_@[[state]] is not "readable", throw a *TypeError* exception.
  1. Let *this*@[[queue]] be a new empty List.
  1. ErrorReadableByteStream(_stream_, e).
  1. Return *undefined*.
</pre>

<h5 id="rbs-controller-respond">respond(chunk)</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamController(*this*) is *false*, throw a *TypeError* exception.
  1. Let _stream_ be *this*@[[controlledReadableByteStream]].
  1. Let _reader_ be _stream_@[[reader]].
  1. If _reader_ is *undefined*, throw a *TypeError* exception.
  1. If _reader_ has a [[readRequests]] internal slot, throw a *TypeError* exception.
  1. Assert: _reader_ has a [[readIntoRequests]] internal slot.
  1. If _reader_@[[readIntoRequests]] is empty, throw a *TypeError* exception.
  1. RespondToReadableByteStreamByobReaderReadIntoRequest(_stream_, _chunk_).
  1. Return *undefined*.
</pre>

<h3 id="byte-reader-class">Class <code>ReadableByteStreamReader</code></h3>

<h4 id="byte-reader-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

If one were to write the <code>ReadableByteStreamReader</code> class in something close to the syntax of [[!ECMASCRIPT]],
it would look like

<pre><code class="lang-javascript">
  class ReadableByteStreamReader {
    constructor(stream)

    get closed()

    cancel(reason)
    read()
    releaseLock()
  }
</code></pre>

<h4 id="byte-reader-internal-slots">Internal Slots</h4>

Instances of <code>ReadableByteStreamReader</code> are created with the internal slots described in the following table:

<table>
  <thead>
    <tr>
      <th>Internal Slot</th>
      <th>Description (<em>non-normative</em>)</th>
    </tr>
  </thead>
  <tr>
    <td>\[[closedPromise]]
    <td>A promise returned by the reader's <code>closed</code> getter
  </tr>
  <tr>
    <td>\[[ownerReadableByteStream]]
    <td>A <code>ReadableByteStream</code> instance that owns this reader
  </tr>
  <tr>
    <td>\[[readRequests]]
    <td>A List of promises returned by calls to the reader's <code>read()</code> method that have not yet been resolved,
      due to the <a>consumer</a> requesting <a>chunks</a> sooner than they are available; also used for the
      <a href="#is-readable-byte-stream-reader">IsReadableByteStreamReader</a> brand check
  </tr>
  <tr>
    <td>\[[state]]
    <td>A string containing the reader's current state, used internally; one of <code>"readable"</code>,
      <code>"closed"</code>, or <code>"errored"</code>
  </tr>
  <tr>
    <td>\[[storedError]]
    <td>A value indicating how the reader's stream failed, to be given as a failure reason or exception when trying to
      operate on the reader; applicable only when \[[state]] is <code>"errored"</code>
  </tr>
</table>

<h4 id="byte-reader-constructor">new ReadableByteStreamReader(stream)</h4>

<div class="note">
  The <code>ReadableByteStreamReader</code> constructor is generally not meant to be used directly; instead, a stream's
  <code>getReader()</code> method should be used. This allows different classes of readable streams to vend different
  classes of readers without the <a>consumer</a> needing to know which goes with which.
</div>

<pre is="emu-alg">
  1. If IsReadableByteStream(_stream_) is *false*, throw a *TypeError* exception.
  1. If IsReadableByteStreamLocked(_stream_) is *true*, throw a *TypeError* exception.
  1. Set _stream_@[[reader]] to *this*.
  1. Set *this*@[[ownerReadableByteStream]] to _stream_.
  1. Set *this*@[[state]] to "readable".
  1. Set *this*@[[storedError]] to *undefined*.
  1. Set *this*@[[readRequests]] to a new empty List.
  1. Set *this*@[[closedPromise]] to a new promise.
  1. If _stream_@[[state]] is "closed" or "errored", ReleaseReadableByteStreamReader(*this*).
</pre>

<h4 id="byte-reader-prototype">Properties of the <code>ReadableByteStreamReader</code> Prototype</h4>

<h5 id="byte-reader-closed">get closed</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. Return *this*@[[closedPromise]].
</pre>

<h5 id="byte-reader-cancel">cancel(reason)</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. If *this*@[[state]] is "closed", return a new promise resolved with *undefined*.
  1. If *this*@[[state]] is "errored", return a new promise rejected with *this*@[[storedError]].
  1. Assert: *this*@[[ownerReadableByteStream]] is not *undefined*.
  1. Return CancelReadableByteStream(*this*@[[ownerReadableByteStream]], _reason_).
</pre>

<h5 id="byte-reader-read">read()</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. Return ReadFromReadableByteStreamReader(*this*).
</pre>

<h5 id="byte-reader-release-lock">releaseLock()</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamReader(*this*) is *false*, throw a *TypeError* exception.
  1. If *this*@[[ownerReadableByteStream]] is *undefined*, return *undefined*.
  1. If *this*@[[readRequests]] is not empty, throw a *TypeError* exception.
  1. SyncReadableByteStreamReaderStateWithOwner(*this*@[[ownerReadableByteStream]]).
  1. DetachReadableByteStreamReader(*this*).
  1. Return *undefined*.
</pre>

<h3 id="byte-byob-reader-class">Class <code>ReadableByteStreamByobReader</code></h3>

<h4 id="byte-byob-reader-class-definition">Class Definition</h4>

<em>This section is non-normative.</em>

If one were to write the <code>ReadableByteStreamByobReader</code> class in something close to the syntax of
[[!ECMASCRIPT]], it would look like

<pre><code class="lang-javascript">
  class ReadableByteStreamByobReader {
    constructor(stream)

    get closed()

    cancel(reason)
    read(view)
    releaseLock()
  }
</code></pre>

<h4 id="byte-byob-reader-internal-slots">Internal Slots</h4>

Instances of <code>ReadableByteStreamByobReader</code> are created with the internal slots described in the following
table:

<table>
  <thead>
    <tr>
      <th>Internal Slot</th>
      <th>Description (<em>non-normative</em>)</th>
    </tr>
  </thead>
  <tr>
    <td>\[[closedPromise]]
    <td>A promise returned by the reader's <code>closed</code> getter
  </tr>
  <tr>
    <td>\[[ownerReadableByteStream]]
    <td>A <code>ReadableByteStream</code> instance that owns this reader
  </tr>
  <tr>
    <td>\[[readIntoRequests]]
    <td>A List of promises returned by calls to the reader's <code>read()</code> method that have not yet been
      resolved, due to the <a>consumer</a> requesting <a>chunks</a> sooner than they are available; also used for the
      <a href="#is-readable-byte-stream-byob-reader">IsReadableByteStreamByobReader</a> brand check
  </tr>
  <tr>
    <td>\[[state]]
    <td>A string containing the reader's current state, used internally; one of <code>"readable"</code>,
      <code>"closed"</code>, or <code>"errored"</code>
  </tr>
  <tr>
    <td>\[[storedError]]
    <td>A value indicating how the reader's stream failed, to be given as a failure reason or exception when trying to
      operate on the reader; applicable only when \[[state]] is <code>"errored"</code>
  </tr>
</table>

<h4 id="byte-byob-reader-constructor">new ReadableByteStreamByobReader(stream)</h4>

<div class="note">
  The <code>ReadableByteStreamByobReader</code> constructor is generally not meant to be used directly; instead, a
  stream's <code>getByobReader()</code> method should be used. This allows different classes of readable streams to
  vend different classes of readers without the <a>consumer</a> needing to know which goes with which.
</div>

<pre is="emu-alg">
  1. If IsReadableByteStream(_stream_) is *false*, throw a *TypeError* exception.
  1. If IsReadableByteStreamLocked(_stream_) is *true*, throw a *TypeError* exception.
  1. Set _stream_@[[reader]] to *this*.
  1. Set *this*@[[ownerReadableByteStream]] to _stream_.
  1. Set *this*@[[state]] to "readable".
  1. Set *this*@[[storedError]] to *undefined*.
  1. Set *this*@[[readIntoRequests]] to a new empty List.
  1. Set *this*@[[closedPromise]] to a new promise.
  1. If _stream_@[[state]] is "closed" or "errored", ReleaseReadableByteStreamReader(*this*).
</pre>

<h4 id="byte-byob-reader-prototype">Properties of the <code>ReadableByteStreamByobReader</code> Prototype</h4>

<h5 id="byte-byob-reader-closed">get closed</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamByobReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. Return *this*@[[closedPromise]].
</pre>

<h5 id="byte-byob-reader-cancel">cancel(reason)</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamByobReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. If *this*@[[state]] is "closed", return a new promise resolved with *undefined*.
  1. If *this*@[[state]] is "errored", return a new promise rejected with *this*@[[storedError]].
  1. Assert: *this*@[[ownerReadableByteStream]] is not *undefined*.
  1. Return CancelReadableByteStream(*this*@[[ownerReadableByteStream]], _reason_).
</pre>

<h5 id="byte-byob-reader-read">read(view)</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamByobReader(*this*) is *false*, return a promise rejected with a *TypeError* exception.
  1. Return ReadFromReadableByteStreamByobReader(*this*, _view_).
</pre>

<h5 id="byte-byob-reader-release-lock">releaseLock()</h5>

<pre is="emu-alg">
  1. If IsReadableByteStreamByobReader(*this*) is *false*, throw a *TypeError* exception.
  1. If *this*@[[ownerReadableByteStream]] is *undefined*, return *undefined*.
  1. If *this*@[[readIntoRequests]] is not empty, throw a *TypeError* exception.
  1. SyncReadableByteStreamReaderStateWithOwner(*this*@[[ownerReadableByteStream]]).
  1. DetachReadableByteStreamReader(*this*).
  1. Return *undefined*.
</pre>

<h3 id="byte-rs-abstract-ops">Readable Byte Stream Abstract Operations</h3>

<h4 id="acquire-readable-byte-stream-byob-reader" aoid="AcquireReadableByteStreamByobReader">AcquireReadableByteStreamByobReader ( stream )</h4>

This abstract operation is meant to be called from other specifications that may wish to acquire an <a>readable byte
stream byob reader</a> for a given stream.

<pre is="emu-alg">
  1. Return Construct(`ReadableByteStreamByobReader`, «_stream_»).
</pre>

<h4 id="acquire-readable-byte-stream-reader" aoid="AcquireReadableByteStreamReader">AcquireReadableByteStreamReader ( stream )</h4>

This abstract operation is meant to be called from other specifications that may wish to acquire an <a>readable byte
stream reader</a> for a given stream.

<pre is="emu-alg">
  1. Return Construct(`ReadableByteStreamReader`, «_stream_»).
</pre>

<h4 id="cancel-readable-byte-stream" aoid="CancelReadableByteStream">CancelReadableByteStream ( stream, reason )</h4>

<pre is="emu-alg">
  1. If _stream_@[[state]] is "closed", return a new promise resolved with *undefined*.
  1. If _stream_@[[state]] is "errored", return a new promise rejected with _stream_@[[storedError]].
  1. CloseReadableByteStream(_stream_).
  1. Set _stream_@[[controller]]@[[queue]] to a new empty List.
  1. Let _sourceCancelPromise_ be PromiseInvokeOrNoop(_stream_@[[controller]]@[[underlyingByteSource]], "cancel", «‍_reason_»).
  1. Return the result of transforming _sourceCancelPromise_ by a fulfillment handler that returns *undefined*.
</pre>

<h4 id="close-readable-byte-stream" aoid="CloseReadableByteStream">CloseReadableByteStream ( stream )</h4>

<pre is="emu-alg">
  1. Assert: IsReadableByteStream(_stream_) is *true*.
  1. Assert: _stream_@[[state]] is "readable".
  1. Set _stream_@[[state]] to "closed".
  1. ReleaseReadableByteStreamReader(_stream_).
  1. Return *undefined*.
</pre>

<h4 id="detach-readable-byte-stream-reader" aoid="DetachReadableByteStreamReader">DetachReadableByteStreamReader ( stream )</h4>

<pre is="emu-alg">
  1. Assert: _stream_@[[reader]] is not *undefined*.
  1. Set _stream_@[[reader]]@[[ownerReadableByteStream]] to *undefined*.
  1. Set _stream_@[[reader]] to *undefined*.
</pre>

<h4 id="enqueue-in-readable-byte-stream-controller" aoid="EnqueueInReadableByteStreamController">EnqueueInReadableByteStreamController ( controller, chunk )</h4>

<pre is="emu-alg">
  1. Let _chunkSize_ be *1*.
  1. If _controller_@[[strategySize]] is not *undefined*,
    1. Set _chunkSize_ to Call(_controller_@[[strategySize]], *undefined*, «_chunk_»).
    1. If _chunkSize_ is an abrupt completion,
      1. Let _controller_@[[queue]] be a new empty List.
      1. ErrorReadableByteStream(_stream_, ‍_chunkSize_.[[value]]).
      1. Return _chunkSize_.
    1. Let _chunkSize_ be _chunkSize_.[[value]].
  1. Let _enqueueResult_ be EnqueueValueWithSize(_controller_@[[queue]], _chunk_, _chunkSize_).
  1. If _enqueueResult_ is an abrupt completion,
    1. Let _controller_@[[queue]] be a new empty List.
    1. ErrorReadableByteStream(_stream_, ‍_enqueueResult_.[[value]]).
    1. Return _enqueueResult_.
  1. Return *undefined*.
</pre>

<h4 id="error-readable-byte-stream" aoid="ErrorReadableByteStream">ErrorReadableByteStream ( stream, e )</h4>

<pre is="emu-alg">
  1. Assert: IsReadableByteStream(_stream_) is *true*.
  1. Assert: _stream_@[[state]] is "readable".
  1. Set _stream_@[[storedError]] to _e_.
  1. Set _stream_@[[state]] to "errored".
  1. ReleaseReadableByteStreamReader(_stream_).
  1. Return *undefined*.
</pre>

<h4 id="flush-readable-byte-stream-reader-read-requests" aoid="FlushReadableByteStreamReaderReadRequests">FlushReadableByteStreamReaderReadRequests ( stream )</h4>

<pre is="emu-alg">
  1. Assert: _stream_@[[reader]] is not *undefined*.
  1. Let _reader_ be _stream_@[[reader]].
  1. If _stream_@[[state]] is "errored",
    1. Repeat for each _readRequestPromise_ that is an element of _reader_@[[readRequests]],
      1. Reject _readRequestPromise_ with _stream_@[[storedError]].
  1. Otherwise,
    1. Repeat for each _readRequestPromise_ that is an element of _reader_@[[readRequests]],
      1. Resolve _readRequestPromise_ with CreateIterResultObject(*undefined*, *true*).
  1. Set _reader_@[[readRequests]] to a new empty List.
</pre>

<h4 id="get-readable-byte-stream-desired-size" aoid="GetReadableByteStreamDesiredSize">GetReadableByteStreamDesiredSize ( stream )</h4>

<pre is="emu-alg">
  1. Let _queueSize_ be GetTotalQueueSize(_stream_@[[controller]]@[[queue]]) - _stream_@[[controller]]@[[consumedBytesOfQueueHead]].
  1. ReturnIfAbrupt(_queueSize_).
  1. Return _stream_@[[controller]]@[[strategyHWM]] − _queueSize_.
</pre>

<h4 id="is-readable-byte-stream" aoid="IsReadableByteStream">IsReadableByteStream ( x )</h4>

<pre is="emu-alg">
  1. If Type(_x_) is not Object, return *false*.
  1. If _x_ does not have a [[underlyingByteSource]] internal slot, return *false*.
  1. Return *true*.
</pre>

<h4 id="is-readable-byte-stream-controller" aoid="IsReadableByteStreamController">IsReadableByteStreamController ( x )</h4>

<pre is="emu-alg">
  1. If Type(_x_) is not Object, return *false*.
  1. If _x_ does not have a [[controlledReadableByteStream]] internal slot, return *false*.
  1. Return *true*.
</pre>

<h4 id="is-readable-byte-stream-locked" aoid="IsReadableByteStreamLocked">IsReadableByteStreamLocked ( stream )</h4>

<pre is="emu-alg">
  1. Assert: IsReadableByteStream(_stream_) is *true*.
  1. If _stream_@[[reader]] is *undefined*, return *false*.
  1. Return *true*.
</pre>

<h4 id="is-readable-byte-stream-reader" aoid="IsReadableByteStreamReader">IsReadableByteStreamReader ( x )</h4>

<pre is="emu-alg">
  1. If Type(_x_) is not Object, return *false*.
  1. If _x_ does not have a [[readRequests]] internal slot, return *false*.
  1. Return *true*.
</pre>

<h4 id="is-readable-byte-stream-byob-reader" aoid="IsReadableByteStreamByobReader">IsReadableByteStreamByobReader ( x )</h4>

<pre is="emu-alg">
  1. If Type(_x_) is not Object, return *false*.
  1. If _x_ does not have a [[readIntoRequests]] internal slot, return *false*.
  1. Return *true*.
</pre>

<h4 id="maybe-respond-to-read-into-request-from-queue" aoid="MaybeRespondToReadIntoRequestFromQueue">MaybeRespondToReadIntoRequestFromQueue ( stream )</h4>

<pre is="emu-alg">
  1. Let _controller_ be _stream_@[[controller]].
  1. Let _queue_ be _controller_@[[queue]].
  1. Assert: _queue_ is not empty.
  1. Let _dest_ be the first element of _stream_@[[controller]]@[[pendingViews]].
  1. Let _destBuffer_ be GetViewedArrayBuffer(_dest_).
  1. ReturnIfAbrupt(_destBuffer_).
  1. Let _destByteLength_ be GetByteLength(_dest_).
  1. ReturnIfAbrupt(_destByteLength_).
  1. Let _destByteOffset_ be GetByteOffset(_dest_).
  1. ReturnIfAbrupt(_destByteOffset_).
  1. Let _destBytesAlreadyFilled_ be _controller_@[[filledBytesOfPendingViewsHead]].
  1. Let _chunk_ be PeakQueueValue(_queue_).
  1. Let _chunkBuffer_ be GetViewedArrayBuffer(_chunk_).
  1. ReturnIfAbrupt(_chunkBuffer_).
  1. Let _chunkByteLength_ be GetByteLength(_chunk_).
  1. ReturnIfAbrupt(_chunkByteLength_).
  1. Let _chunkByteOffset_ be GetByteOffset(_chunk_).
  1. ReturnIfAbrupt(_chunkByteOffset_).
  1. Let _chunkBytesAlreadyConsumed_ be _controller_@[[consumedBytesOfQueueHead]].
  1. Let _bytesToCopy_ be _chunkByteLength_ - _chunkBytesAlreadyConsumed_.
  1. Let _destRemaining_ be _destByteLength_ - _destBytesAlreadyFilled_.
  1. If _bytesToCopy_ > _destRemaining_, let _bytesToCopy_ be _destRemaining_.
  1. Let _validationResult_ be ValidateTypedArray(_dest_).
  1. Let _elementSize_ be *1*.
  1. If _validationResult_ is not an abrupt completion,
    1. Let _constructorName_ be the string value of _dest_'s [[TypedArrayName]] internal slot.
    1. Let _elementSize_ be the Number value of the Element Size value for _constructorName_.
    1. If _bytesToCopy_ + _destBytesAlreadyFilled_ > _elementSize_,
      1. Let _remainderBytes_ be (_bytesToCopy_ + _destBytesAlreadyFilled_) modulo _elementSize_.
      1. Let _bytesToCopy_ be _bytesToCopy_ - _remainderBytes_.
  1. Let _destBeginPosition_ be _destByteOffset_ + _destBytesAlreadyFilled_.
  1. Let _destEndPosition_ be _destByteOffset_ + _destBytesAlreadyFilled_ + _bytesToCopy_.
  1. Let _chunkBeginPosition_ be _chunkByteOffset_ + _chunkBytesAlreadyConsumed_.
  1. Let _chunkEndPosition_ be _chunkByteOffset_ + _chunkBytesAlreadyConsumed_ + _bytesToCopy_.
  1. Set the contents of _destBuffer_ from _destBeginPosition_, inclusive, up to _destEndPosition_, exclusive, to the bytes in _chunkBuffer_ from _chunkBeginPosition_, inclusive, up to _chunkEndPosition_, exclusive.
  1. Let _destBytesAlreadyFilled_ be _destBytesAlreadyFilled_ + _bytesToCopy_.
  1. If _destBytesAlreadyFilled_ ≥ _elementSize_,
    1. Remove _dest_ from _controller_@[[pendingViews]], shifting all other elements downward (so that the second becomes the first, and so on).
    1. If _destByteLength_ is not equal to _destBytesAlreadyFilled_,
      1. If _validationResult_ is not an abrupt completion,
        1. Let _newLength_ be _destBytesAlreadyFilled_ / _elementSize_.
        1. Let _dest_ be the result of calling %TypedArray%.prototype.subarray on _dest_ passing *0* and _newLength_ as the arguments.
      1. Otherwise,
        1. If _dest_ does not have a [[DataView]] internal slot, throw a *TypeError* exception.
        1. Let _dest_ be Construct(`DataView`, «_destBuffer_, _destByteOffset, _destBytesAlreadyFilled_»).
    1. RespondToReadableByteStreamByobReaderReadIntoRequest(_stream_, _dest_).
    1. Set _controller_@[[filledBytesOfPendingViewsHead]] to *0*.
  1. Let _chunkBytesAlreadyConsumed_ be _chunkBytesAlreadyConsumed_ + _bytesToCopy_.
  1. If _chunkBytesAlreadyConsumed_ equals _chunkByteLength_,
    1. DequeueValue(_queue_).
    1. Set _controller_@[[consumedBytesOfQueueHead]] to *0*.
    1. If _queue_ is empty and _controller_@[[closeRequested]] is *true*,
      1. If _controller_@[[filledBytesOfPendingViewsHead]] is not equal to *0*, throw a *TypeError* exception.
      1. CloseReadableByteStream(_stream_).
  1. Otherwise, set _controller_@[[consumedBytesOfQueueHead]] to _chunkBytesAlreadyConsumed_.
</pre>

<h4 id="pop-bytes-from-queue" aoid="PopBytesFromQueue">PopBytesFromQueue ( controller )</h4>

<pre is="emu-alg">
  1. Let _queue_ be _controller_@[[queue]].
  1. Let _chunk_ be PeekQueueValue(_queue_).
  1. DequeueValue(_queue_).
  1. Let _bytesAlreadyConsumed_ be _controller_@[[consumedBytesOfQueueHead]].
  1. If _bytesAlreadyConsumed_ is not *0*,
    1. Let _buffer_ be GetViewedArrayBuffer(_chunk_).
    1. ReturnIfAbrupt(_buffer_).
    1. Let _byteLength_ be GetByteLength(_chunk_).
    1. ReturnIfAbrtup(_byteLength_).
    1. Let _byteOffset_ be GetByteOffset(_chunk_).
    1. ReturnIfAbrtup(_byteOffset_).
    1. Let _chunk_ be Construct(`Uint8Array`, «_buffer_, _byteOffset_ + _bytesAlreadyConsumed_, _byteLength_ - _bytesAlreadyConsumed_»).
    1. Set _controller_@[[consumedBytesOfQueueHead]] to *0*.
  1. Return _chunk_.
</pre>

<h4 id="pull-from-readable-byte-stream" aoid="PullFromReadableByteStream">PullFromReadableByteStream ( stream )</h4>

<pre is="emu-alg">
  1. Let _controller_ be _stream_@[[controller]].
  1. Let _queue_ be _controller_@[[queue]].
  1. If _queue_ is empty,
    1. Let _pullResult_ be Invoke(_stream_@[[underlyingByteSource]], "pull").
    1. If _pullResult_ is an abrupt completion,
      1. If _stream_@[[state]] is "readable", ErrorReadableByteStream(_stream_, _pullResult_.[[value]]).
    1. Return *undefined*.
  1. Let _chunk_ be PopBytesFromQueue(_controller_).
  1. If _chunk_ is an abrupt completion,
    1. ErrorReadableByteStream(_stream_, _chunk_.[[value]]).
    1. Return *undefined*.
  1. RespondToReadableByteStreamReaderReadRequest(_stream_, _chunk_).
  1. If _queue_ is empty and _controller_@[[closeRequested]] is *true*, CloseReadableByteStream(_stream_).
  1. Return *undefined*.
</pre>

<h4 id="pull-from-readable-byte-stream-into" aoid="PullFromReadableByteStreamInto">PullFromReadableByteStreamInto ( stream, view )</h4>

<pre is="emu-alg">
  1. Let _controller_ be _stream_@[[controller]].
  1. Let _queue_ be _controller_@[[queue]].
  1. If _queue_ is empty,
    1. Let _pullIntoResult_ be Invoke(_controller_@[[underlyingByteSource]], "pullInto", «_view_»).
    1. If _pullIntoResult_ is an abrupt completion,
      1. If _stream_@[[state]] is "readable", ErrorReadableByteStream(_stream_, _pullIntoResult_.[[value]]).
    1. Return *undefined*.
  1. Append _view_ as the last element of _controller_@[[pendingViews]].
  1. Let _respondResult_ be MaybeRespondToReadIntoRequestFromQueue(_stream_).
  1. If _respondResult_ is an abrupt completion,
    1. If _stream_@[[state]] is "readable", ErrorReadableByteStream(_stream_, _respondResult_.[[value]]).
    1. Return *undefined*.
  1. Return *undefined*.
</pre>

<h4 id="read-from-readable-byte-stream-byob-reader" aoid="ReadFromReadableByteStreamByobReader">ReadFromReadableByteStreamByobReader ( reader, view )</h4>

<pre is="emu-alg">
  1. If _reader_@[[ownerReadableByteStream]] is *undefined*,
    1. If _reader_@[[state]] is "closed", return a new promise resolved with CreateIterResultObject(_view_, *true*).
    1. Assert: _reader_@[[state]] is "errored".
    1. Return a new promise rejected with _reader_@[[storedError]].
  1. Let _readRequestPromise_ be a new promise.
  1. Append _readRequestPromise_ as the last element of _reader_@[[readIntoRequests]].
  1. PullFromReadableByteStreamInto(_reader_@[[ownerReadableByteStream]], _view_).
  1. Return _readRequestPromise_.
</pre>

<h4 id="read-from-readable-byte-stream-reader" aoid="ReadFromReadableByteStreamReader">ReadFromReadableByteStreamReader ( reader )</h4>

<pre is="emu-alg">
  1. If _reader_@[[state]] is "closed", return a new promise resolved with CreateIterResultObject(*undefined*, *true*).
  1. If _reader_@[[state]] is "errored", return a new promise rejected with _reader_@[[storedError]].
  1. Assert: _reader_@[[ownerReadableByteStream]] is not *undefined*.
  1. Assert: _reader_@[[ownerReadableByteStream]]@[[state]] is "readable".
  1. Let _readRequestPromise_ be a new promise.
  1. Append _readRequestPromise_ as the last element of _reader_@[[readRequests]].
  1. PullFromReadableByteStream(_reader_@[[ownerReadableByteStream]]).
  1. Return _readRequestPromise_.
</pre>

<h4 id="release-readable-byte-stream-reader" aoid="ReleaseReadableByteStreamReader">ReleaseReadableByteStreamReader ( stream )</h4>

<pre is="emu-alg">
  1. Let _reader_ be _stream_@[[reader]].
  1. If _reader_ is *undefined*, return *undefined*.
  1. If _reader_ has a [[readRequests]] internal slot,
    1. SyncReadableByteStreamReaderStateWithOwner(_stream_).
    1. FlushReadableByteStreamReaderReadRequests(_stream_).
    1. DetachReadableByteStreamReader(_reader_).
    1. Return *undefined*.
  1. Assert: _reader_ has a [[readIntoRequests]] internal slot.
  1. SyncReadableByteStreamReaderStateWithOwner(_stream_).
  1. If _reader_@[[readIntoRequests]] is empty, DetachReadableByteStreamReader(_stream_).
  1. Return *undefined*.
</pre>

<h4 id="respond-to-readable-byte-stream-reader-read-request" aoid="RespondToReadableByteStreamReaderReadRequest">RespondToReadableByteStreamReaderReadRequest ( stream, chunk )</h4>

<pre is="emu-alg">
  1. Assert: _stream_@[[state]] is "readable".
  1. Assert: _stream_@[[reader]]@[[readRequests]] is not empty,
  1. Let _readRequestPromise_ be the first element of _stream_@[[reader]]@[[readRequests]].
  1. Remove _readRequestPromise_ from _stream_@[[reader]]@[[readRequests]], shifting all other elements downward (so that the second becomes the first, and so on).
  1. Resolve _readRequestPromise_ with CreateIterResultObject(_chunk_, *false*).
  1. Return *undefined*.
</pre>

<h4 id="respond-to-readable-byte-stream-byob-reader-read-into-request" aoid="RespondToReadableByteStreamByobReaderReadIntoRequest">RespondToReadableByteStreamByobReaderReadIntoRequest ( stream, chunk )</h4>

<pre is="emu-alg">
  1. Assert: _stream_@[[reader]] is not *undefined*.
  1. Assert: _stream_@[[reader]]@[[readIntoRequests]] is not empty,
  1. Let _reader_ be _stream_@[[reader]].
  1. Let _readRequestPromise_ be the first element of _reader_@[[readIntoRequests]].
  1. Remove _readRequestPromise_ from _reader_@[[readIntoRequests]], shifting all other elements downward (so that the second becomes the first, and so on).
  1. If _stream_@[[state]] is "readable",
    1. Resolve _readRequestPromise_ with CreateIterResultObject(_chunk_, *false*).
    1. Return *undefined*.
  1. If _stream_@[[state]] is "closed",
    1. Resolve _readRequestPromise_ with CreateIterResultObject(_chunk_, *true*).
  1. Otherwise, _reader_@[[state]] must be "errored" so,
    1. Reject _readRequestPromise_ with CreateIterResultObject(_stream_@[[storedError]], _chunk_).
  1. If _reader_@[[readIntoRequests]] is empty, DetachReadableByteStreamReader(_stream_).
  1. Return *undefined*.
</pre>

<h4 id="sync-readable-byte-stream-reader-state-with-owner" aoid="SyncReadableByteStreamReaderStateWithOwner">SyncReadableByteStreamReaderStateWithOwner ( stream )</h4>

<pre is="emu-alg">
  1. Assert: _stream_@[[reader]] is not *undefined*.
  1. Let _reader_ be _stream_@[[reader]].
  1. Let _e_ be _stream_@[[storedError]].
  1. If _stream_@[[state]] is "errored",
    1. Set _reader_@[[state]] to "errored".
    1. Set _reader_@[[storedError]] to _e_.
    1. Reject _reader_@[[closedPromise]] with _e_.
  1. Otherwise,
    1. Set _reader_@[[state]] to "closed".
    1. Resolve _reader_@[[closedPromise]] with *undefined*.
</pre>

<h4 id="tee-readable-byte-stream" aoid="TeeReadableByteStream">TeeReadableByteStream ( stream, shouldClone )</h4>

This abstract operation is meant to be called from other specifications that may wish to tee a given readable byte
stream. Its second argument governs whether or not the data from the original stream will be
<a lt="structured clone">structured cloned</a> before becoming visible in the returned branches.

<pre is="emu-alg">
  1. Assert: IsReadableByteStream(_stream_) is *true*.
  1. Assert: Type(_shouldClone_) is Boolean.
  1. Let _reader_ be AcquireReadableByteStreamReader(_reader_).
  1. ReturnIfAbrupt(_reader_).
  1. Let _pull1_ be a new <a>TeeReadableByteStream branch 1 pull function</a>.
  1. Let _pull2_ be a new <a>TeeReadableByteStream branch 2 pull function</a>.
  1. Let _cancel1_ be a new <a>TeeReadableByteStream branch 1 cancel function</a>.
  1. Let _cancel2_ be a new <a>TeeReadableByteStream branch 2 cancel function</a>.
  1. Let _underlyingByteSource1_ be ObjectCreate(%ObjectPrototype%).
  1. Perform CreateDataProperty(_underlyingByteSource1_, "pull", _pull1_).
  1. Perform CreateDataProperty(_underlyingByteSource1_, "cancel", _cancel1_).
  1. Let _branch1_ be Construct(`ReadableByteStream`, _underlyingByteSource1_).
  1. Let _underlyingByteSource2_ be ObjectCreate(%ObjectPrototype%).
  1. Perform CreateDataProperty(_underlyingByteSource2_, "pull", _pull2_).
  1. Perform CreateDataProperty(_underlyingByteSource2_, "cancel", _cancel2_).
  1. Let _branch2_ be Construct(`ReadableByteStream`, _underlyingByteSource2_).
  1. Let _teeState_ be Record{[[reader]]: _reader_, [[closedOrError]]: *false*, [[branch1]]: _branch1_, [[branch2]]: _branch2_, [[pullCount1]]: 0, [[pullCount2]]: 0, [[queue1]]: an empty List, [[queue2]]: an empty List, [[originalClosed]]: *false*, [[canceled1]]: *false*, [[canceled2]]: *false*, [[reason1]]: *undefined*, [[reason2]]: *undefined*, [[promise]]: a new promise, [[shouldClone]]: _shouldClone_}.
  1. Set _pull1_@[[teeState]] to _teeState_.
  1. Set _pull2_@[[teeState]] to _teeState_.
  1. Set _cancel1_@[[teeState]] to _teeState_.
  1. Set _cancel2_@[[teeState]] to _teeState_.
  1. Upon fulfillment of _reader_@[[closedPromise]],
    1. Set _teeState_.[[originalClosed]] to *true*.
    1. If _teeState_.[[canceled1]] is *false* and _teeState_.[[queue1]] is empty, CloseReadableByteStream(_branch1_).
    1. If _teeState_.[[canceled2]] is *false* and _teeState_.[[queue2]] is empty, CloseReadableByteStream(_branch2_).
  1. Upon rejection of _reader_@[[closedPromise]] with reason _r_,
    1. If _teeState_.[[canceled1]] is *false*, ErrorReadableByteStream(_branch1_, _r_).
    1. If _teeState_.[[canceled2]] is *false*, ErrorReadableByteStream(_branch2_, _r_).
  1. Return «branch1, branch2».
</pre>

A <dfn>TeeReadableByteStream read handler function</dfn> is an anonymous built-in function that reads data from a given
<a>readable byte stream reader</a> and fulfills read requests of two other streams ("branches" of the associated tee).
Each TeeReadableByteStream read handler function has \[[teeState]] internal slot. When a TeeReadableByteStream read
handler function <var>F</var> is called with the argument _result_, it performs the following steps:

<pre is="emu-alg">
  1. Let _teeState_ be _F_@[[teeState]].
  1. Assert: Type(_result_) is Object.
  1. Let _value_ be Get(_result_, "value").
  1. ReturnIfAbrupt(_value_).
  1. Let _done_ be Get(_result_, "done").
  1. ReturnIfAbrupt(_done_).
  1. Assert: Type(_done_) is Boolean.
  1. If _done_ is *true*, return *undefined*.
  1. If _teeState_.[[canceled1]] is *false*,
    1. Let _value1_ be _value_.
    1. If _shouldClone_ is *true*, set _value1_ to a <a>structured clone</a> of _value_.
    1. If _teeState_.[[pullCount1]] > 0,
      1. RespondReadableByteStream(_branch1_, _value1_).
      1. Set _teeState_.[[pullCount1]] to _teeState_.[[pullCount1]] - 1.
      1. Return *undefined*.
    1. Append _value1_ as the last element of _teeState_.[[queue1]].
  1. If _teeState_.[[canceled2]] is *false*,
    1. Let _value2_ be _value_.
    1. If _shouldClone_ is *true*, set _value2_ to a <a>structured clone</a> of _value_.
    1. If _teeState_.[[pullCount2]] > 0,
      1. RespondReadableByteStream(_branch2_, _value2_).
      1. Set _teeState_.[[pullCount2]] to _teeState_.[[pullCount2]] - 1.
      1. Return *undefined*.
    1. Append _value2_ as the last element of _teeState_.[[queue2]].
</pre>

A <dfn>TeeReadableByteStream branch 1 pull function</dfn> is an anonymous built-in function that pulls data from a
given <a>readable byte stream reader</a>. Each TeeReadableByteStream branch 1 pull function has \[[teeState]] internal
slot. When a TeeReadableByteStream branch 1 pull function <var>F</var> is called, it performs the following steps:

<pre is="emu-alg">
  1. Let _teeState_ be _F_@[[teeState]].
  1. If _teeState_.[[queue1]] is not empty,
    1. Remove _chunk_ from _teeState_@[[queue1]], shifting all other elements downward (so that the second becomes the first, and so on).
    1. RespondReadableByteStream(_branch1_, _value1_).
    1. If _teeState_.[[originalClosed]], CloseReadableByteStream(_branch1_).
    1. Return *undefined*.
  1. Set _teeState_.[[pullCount1]] to _teeState_.[[pullCount1]] + 1.
  1. If _teeState_.[[pullCount1]] > _teeState_.[[pullCount2]],
    1. Let _readFunction_ be a new <a>TeeReadableByteStream read handler function</a>.
    1. Set _readFunction_@[[teeState][] to _teeState_.
    1. Return the result of transforming ReadFromReadableStreamReader(_reader_) by a fulfillment handler _readFunction_.
</pre>

A <dfn>TeeReadableByteStream branch 2 pull function</dfn> is an anonymous built-in function that pulls data from a
given <a>readable byte stream reader</a>. Each TeeReadableByteStream branch 2 pull function has \[[teeState]] internal
slot. When a TeeReadableByteStream branch 2 pull function <var>F</var> is called, it performs the following steps:

<pre is="emu-alg">
  1. Let _teeState_ be _F_@[[teeState]].
  1. If _teeState_.[[queue2]] is not empty,
    1. Remove _chunk_ from _teeState_@[[queue2]], shifting all other elements downward (so that the second becomes the first, and so on).
    1. RespondReadableByteStream(_branch2_, _value1_).
    1. If _teeState_.[[originalClosed]], CloseReadableByteStream(_branch2_).
    1. Return *undefined*.
  1. Set _teeState_.[[pullCount2]] to _teeState_.[[pullCount2]] + 1.
  1. If _teeState_.[[pullCount2]] > _teeState_.[[pullCount1]],
    1. Let _readFunction_ be a new <a>TeeReadableByteStream read handler function</a>.
    1. Set _readFunction_@[[teeState][] to _teeState_.
    1. Return the result of transforming ReadFromReadableStreamReader(_reader_) by a fulfillment handler _readFunction_.
</pre>

A <dfn>TeeReadableByteStream branch 1 cancel function</dfn> is an anonymous built-in function that reacts to the
cancellation of the first of the two branches of the associated tee. Each TeeReadableByteStream branch 1 cancel
function has \[[teeState]] internal slot. When a TeeReadableByteStream branch 1 cancel function <var>F</var> is called
with argument <var>r</var>, it performs the following steps:

<pre is="emu-alg">
  1. Let _teeState_ be _F_@[[teeState]].
  1. Set _teeState_.[[queue1]] to *undefined*.
  1. Set _teeState_.[[canceled1]] to *true*.
  1. Set _teeState_.[[reason1]] to _r_.
  1. If _teeState_.[[canceled2]] is *true*,
    1. Let _compositeReason_ be CreateArrayFromList(«_teeState_.[[reason1]], _teeState_.[[reason2]]»).
    1. Let _cancelResult_ be CancelReadableStream(_teeState_.[[reader]], _compositeReason_).
    1. Resolve _teeState_.[[promise]] with _cancelResult_.
  1. Return _teeState_.[[promise]].
</pre>

A <dfn>TeeReadableByteStream branch 2 cancel function</dfn> is an anonymous built-in function that reacts to the
cancellation of the second of the two branches of the associated tee. Each TeeReadableByteStream branch 2 cancel
function has \[[teeState]] internal slot. When a TeeReadableStream branch 2 cancel function <var>F</var> is called with
argument <var>r</var>, it performs the following steps:

<pre is="emu-alg">
  1. Let _teeState_ be _F_@[[teeState]].
  1. Set _teeState_.[[queue2]] to *undefined*.
  1. Set _teeState_.[[canceled2]] to *true*.
  1. Set _teeState_.[[reason2]] to _r_.
  1. If _teeState_.[[canceled1]] is *true*,
    1. Let _compositeReason_ be CreateArrayFromList(«_teeState_.[[reason1]], _teeState_.[[reason2]]»).
    1. Let _cancelResult_ be CancelReadableStream(_teeState_.[[reader]], _compositeReason_).
    1. Resolve _teeState_.[[promise]] with _cancelResult_.
  1. Return _teeState_.[[promise]].
</pre>

<h2 id="globals">Global Properties</h2>

Ideally, this standard would add no new properties to the global object, instead relying on a standard ECMAScript
module. However, given that the specification for and implementation of modules in a browser environment is still in
flux, in the meantime the following properties must be exposed on the global object:

<ul>
  <li> <code>ReadableStream</code>
  <li> <code>ReadableByteStream</code>
  <li> <code>WritableStream</code>
  <li> <code>ByteLengthQueuingStrategy</code>
  <li> <code>CountQueuingStrategy</code>
</ul>

In all cases the property must be a data property, with its value being the corresponding constructor defined in this
standard, and other attributes { \[[Writable]]: <b>true</b>, \[[Enumerable]]: <b>false</b>, \[[Configurable]]:
<b>true</b> }.

<div class="note">
  The <code>ReadableStreamReader</code> class is specifically <em>not</em> exposed, as while it does have a
  functioning constructor, instances should instead be created through the <code>getReader</code> method of a
  <code>ReadableStream</code> instance. Similarly, the <code>ReadableStreamController</code> class is not exposed,
  since its constructor is not independently useful outside of the <code>ReadableStream</code> implementation.
</div>

<div class="note">
  If by the time implementers begin implementing this standard, modules start becoming feasible in the relevant
  environments, then we should reconsider this requirement.
</div>

<h2 id="other-streams">Other Stream Implementations</h2>

<em>This section is non-normative.</em>

The <code>ReadableStream</code> and <code>WritableStream</code> classes defined in this specification are not expected
to be the only manifestations of the corresponding <a>readable stream</a> and <a>writable stream</a> concepts. They are
explicitly meant to cooperate with other stream instances that behave similarly. Those instances could be e.g.
platform- or developer-created subclasses of these classes, or they could be anything else that obeys the same public
API contract.

For example, we are already prototyping and planning an additional <code>ReadableByteStream</code> class, which will be
a <a>readable stream</a> while not being a subclass of <code>ReadableStream</code>. It will have the same set of
methods as a baseline <code>ReadableStream</code>, and can be used in the same way by most <a>consumers</a>, who will
be agnostic to which type of readable stream they are using. However, specialized consumers who know they are dealing
with a <code>ReadableByteStream</code> will be able to take advantage of extra APIs it provides for extremely efficient
"bring-your-own-buffer" memory management.

This kind of ecosystem is largely enabled by the genericness of the <code>pipeTo</code> method. Any object which has
the appropriate public writable stream APIs will work with <code>ReadableStream.prototype.pipeTo</code>. We strongly
recommend that any readable stream classs you create maintain this property for <em>its</em> <code>pipeTo</code>
method as well. Even if they use specialized algorithms when given a specific type of writable stream, they should
always fall back to an algorithm that works with any object obeying the writable stream contract.

<div class="note">
  The keen reader may be asking for more precise definitions of the "contracts" in play here, for both readable and
  writable streams. We're working on a test suite that should do the trick! For now, please check out our
  <a href="https://github.com/whatwg/streams/tree/master/reference-implementation/test/templated">templated tests</a>
  to see initial progress in that direction.
</div>

<h2 id="creating-examples">Examples of Creating Streams</h2>

<em>This section, and all its subsections, are non-normative.</em>

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

<h3 id="example-rs-push-no-backpressure">A readable stream with an underlying push source (no backpressure support)</h3>

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

<pre><code class="lang-javascript">
  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();
      }
    });
  }
</code></pre>

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

<pre><code class="lang-javascript">
  const webSocketStream = makeReadableWebSocketStream("wss://example.com", 443);

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

<h3 id="example-rs-push-backpressure">A readable stream with an underlying push source and backpressure support</h3>

The following function returns <a>readable streams</a> 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
<code>readStop</code> and <code>readStart</code> methods. In doing so, this example shows how to apply
<a>backpressure</a> to <a>underlying sources</a> that support it.

<pre><code class="lang-javascript">
  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();
      }
    });
  }
</code></pre>

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.

<h3 id="example-rs-pull">A readable stream with an underlying pull source</h3>

The following function returns <a>readable streams</a> that wrap portions of the
<a href="https://iojs.org/api/fs.html">io.js file system API</a> (which themselves map fairly directly to C's
<code>fopen</code>, <code>fread</code>, and <code>fclose</code> trio). Files are a typical example of <a>pull
sources</a>. Note how in contrast to the examples with push sources, most of the work here happens on-demand in the
<code>pull</code> function, and not at startup time in the <code>start</code> function.

<pre><code class="lang-javascript">
  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(buffer);
          }
        });
      },

      cancel() {
        return fs.close(fd);
      }
    });
  }
</code></pre>

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

<h3 id="example-ws-no-backpressure">A writable stream with no backpressure or success signals</h3>

The following function returns a <a>writable stream</a> that wraps a web socket [[HTML]]. Web sockets do not provide
any way to tell when a given chunk of data has been successfully sent, so this writable stream has no ability to
communicate <a>backpressure</a> signals or write success/failure to its <a>producers</a>. That is, it will always be in
the <code>"writable"</code> state, and the promise returned by its <code>write()</code> method will always fulfill
immediately.

<pre><code class="lang-javascript">
  function makeWritableWebSocketStream(url, protocols) {
    const ws = new WebSocket(url, protocols);

    return new WritableStream({
      start(error) {
        ws.onerror = error;
        return new Promise(resolve => ws.onopen = resolve);
      },

      write(chunk) {
        ws.send(chunk);
        // Return immediately, since the web socket gives us no way to tell
        // when the write completes.
      },

      close() {
        return new Promise((resolve, reject) => {
          ws.onclose = resolve;
          ws.close();
        });
      }
    });
  }
</code></pre>

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

<pre><code class="lang-javascript">
  const webSocketStream = makeWritableWebSocketStream("wss://example.com", 443);

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

<h3 id="example-ws-backpressure">A writable stream with backpressure and success signals</h3>

The following function returns <a>writable streams</a> that wrap portions of the
<a href="https://iojs.org/api/fs.html">io.js file system API</a> (which themselves map fairly directly to C's
  <code>fopen</code>, <code>fwrite</code>, and <code>fclose</code> trio). Since the API we are wrapping provides
a way to tell when a given write succeeds, this stream will be able to communicate <a>backpressure</a> signals as well
as whether an individual write succeeded or failed.

<pre><code class="lang-javascript">
  const fs = require("pr/fs"); // https://github.com/jden/pr

  function makeWritableFileStream(filename) {
    let fd;

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

      write(chunk) {
        return fs.write(fd, chunk, 0, chunk.length);
      }

      close() {
        return fs.close(fd);
      }
    });
  }
</code></pre>

We can then use this function to create a writable stream for a file, and write individual <a>chunks</a> of data to it:

<pre><code class="lang-javascript">
  const fileStream = makeWritableFileStream("/example/path/on/fs.txt");

  fileStream.write("To stream, or not to stream\n");
  fileStream.write("That is the question\n");

  fileStream.close()
    .then(() => console.log("chunks written and stream closed successfully!"))
    .catch(e => console.error(e));
</code></pre>

Note that if a particular call to <code>fs.write</code> takes a longer time, the returned promise will fulfill later.
In the meantime, additional writes can be queued up, which are stored in the stream's internal queue. The accumulation
of chunks in this queue can move the stream into a <code>"waiting"</code> state, which is a signal to <a>producers</a>
that they should back off and stop writing if possible.

The way in which the writable stream queues up writes is especially important in this case, since as stated in
<a href="https://iojs.org/api/fs.html#fs_fs_write_fd_data_position_encoding_callback">the documentation for
<code>fs.write</code></a>, "it is unsafe to use <code>fs.write</code> multiple times on the same file without waiting
for the [promise]." But we don't have to worry about that when writing the <code>makeWritableFileStream</code>
function, since the stream implementation guarantees that the <a>underlying sink</a>'s <code>write</code> method will
not be called until any promises returned by previous calls have fulfilled!

<h3 id="example-both">A { readable, writable } stream pair wrapping the same underlying resource</h3>

The following function returns an object of the form <code>{ readable, writable }</code>, with the
<code>readable</code> property containing a readable stream and the <code>writable</code> property containing a
writable stream, where both streams wrap the same underlying web socket resource. In essence, this combines
[[#example-rs-push-no-backpressure]] and [[#example-ws-no-backpressure]].

While doing so, it illustrates how you can use JavaScript classes to create reusable underlying sink and underlying
source abstractions.

<pre><code class="lang-javascript">
  function streamifyWebSocket(url, protocol) {
    const ws = new WebSocket(url, protocols);
    ws.binaryType = "arraybuffer";

    return {
      readable: new ReadableStream(new WebSocketSource(ws)),
      writable: new WritableStream(new WebSocketSink(ws))
    };
  }

  class WebSocketSource {
    constructor(ws) {
      this._ws = ws;
    }

    start(controller) {
      this._ws.onmessage = event => controller.enqueue(event.data);
      this._ws.onend = () => controller.close();

      this._ws.addEventListener("error", () => {
        controller.error(new Error("The WebSocket errored!"));
      });
    }

    cancel() {
      this._ws.close();
    }
  }

  class WebSocketSink {
    constructor(ws) {
      this._ws = ws;
    }

    start(error) {
      this._ws.addEventListener("error", () => {
        error(new Error("The WebSocket errored!"));
      });

      return new Promise(resolve => this._ws.onopen = resolve);
    }

    write(chunk) {
      this._ws.send(chunk);
    }

    close() {
      return new Promise((resolve, reject) => {
        this._ws.onclose = resolve;
        this._ws.close();
      });
    }
  });
</code></pre>

We can then use the objects created by this function to communicate with a remote web socket, using the standard stream
APIs:

<pre><code class="lang-javascript">
  const streamyWS = streamifyWebSocket("wss://example.com", 443);

  streamyWS.writable.write("Hello");
  streamyWS.writable.write("web socket!");

  streamyWS.readable.read().then(({ value, done }) => {
    console.log("The web socket says: ", value);
  });
</code></pre>

Note how in this setup canceling the <code>readable</code> side will implicitly close the <code>writable</code> side,
and similarly, closing or aborting the <code>writable</code> side will implicitly close the <code>readable</code> side.

<pre><code class="lang-javascript">
  streamyWS.readable.cancel().then(() => {
    assert(streamyWS.writable.state === "closed");
  });
</code></pre>


<h2 id="state-machines">State Machine Diagrams</h2>

<em>This section, and all its subsections, are non-normative.</em>

As explained in excruciating detail above, both readable and writable streams have fairly complex internal state
machines. In reaction to stimuli from various parts of the system, they transfer between several states. The public
<code>state</code> properties of each stream give a high-level overview of how developers should interact with the
rest of the stream's public API. However there are also a number of private flags used for tracking more subtle state
within the stream.

The diagrams in these sections aim to summarize, at least partially, the way in which readable and writable streams
transition between their states.

<h3 id="rs-state-diagram">The Readable Stream State Diagram</h3>

Readable streams transition in response to both actions from the <a>consumer</a> on the stream's public API, and
events instigated by the <a>underlying source</a> when the stream implementation calls the source's methods.

<figure>
  <img src="readable-stream.svg" width="670" alt="The readable stream state machine diagram." />

  <figcaption>
    <dl>
      <dt><span style="font-style: normal; font-weight: normal; font-family: monospace;">monospace</span></dt>
      <dd>Methods of the stream, called by consumers</dd>

      <dt><span style="font-style: normal; font-weight: bold;">bold</span></dt>
      <dd>Underlying source methods, called by the stream</dd>

      <dt><span style="font-style: italic; font-weight: normal;">italic</span></dt>
      <dd>Capabilities given to the underlying source, called by the underlying source methods</dd>
    </dl>
  </figcaption>
</figure>

<h3 id="ws-state-diagram">The Writable Stream State Diagram</h3>

TODO


<h2 id="conventions" class="no-num">Conventions</h2>

This specification uses algorithm conventions very similar to those of [[!ECMASCRIPT]]. However, it deviates in the
following ways, mostly for brevity. It is hoped (and vaguely planned) that eventually the conventions of ECMAScript
itself will evolve in these ways.

<ul>
  <li> We use destructuring notation in function and method declarations, and assume that the destructuring assignment
    procedure was performed before the algorithm starts.
  <li> We similarly use the default argument notation <code>= {}</code> in a couple of cases.
  <li> We use the notation <var>x</var>@\[[y]] to refer to internal slots of an object, instead of saying "the \[[y]]
    internal slot of <var>x</var>."
  <li> We do not alias <b>this</b> to a local variable in each algorithm, instead using it directly.
  <li> We use the phrase "call-with-rethrow AbstractOperation(<var>x</var>, <var>y</var>, <var>z</var>)" as a shorthand
    for "Let <var>opResult</var> be AbstractOperation(<var>x</var>, <var>y</var>, <var>z</var>).
    ReturnIfAbrupt(<var>opResult</var>)."
  <li> We use <a href="https://w3ctag.github.io/promises-guide/#shorthand-phrases">the shorthand phrases from the W3C
    TAG promises guide</a> to operate on promises at a higher level than the ECMAScript spec does.
</ul>

<h2 id="acks" class="no-num">Acknowledgments</h2>

The editor would like to thank
Anne van Kesteren,
Brian di Palma,
Calvin Metcalf,
Dominic Tarr,
Ed Hager,
Forbes Lindesay,
贺师俊 (hax),
Jens Nockert,
Mangala Sadhu Sangeet Singh Khalsa,
Marcos Caceres,
Michael Mior,
Tab Atkins,
Thorsten Lorenz,
Tim Caswell,
Trevor Norris,
tzik,
and
Xabier Rodríguez
for their contributions to this specification.

Special thanks to:
Bert Belder for bringing up <a href="https://github.com/whatwg/streams/issues/253">implementation concerns</a> that led
  to crucial API changes;
Forrest Norvell for his work on the initial reference implementation;
Gorgi Kosev for his breakthrough idea of separating piping into two methods, thus resolving
  <a href="https://github.com/whatwg/streams/issues/44">a major sticking point</a>;
Isaac Schlueter for his pioneering work on JavaScript streams in Node.js;
Jake Verbaten for his early involvement and support;
Janessa Det for the logo;
Will Chan for his help ensuring that the API allows high-performance network streaming;
and
平野裕 (Yutaka Hirano) for his help with the readable stream reader design,

This standard is written by <a href="https://domenic.me/">Domenic Denicola</a>
(<a href="https://google.com">Google</a>, <a href="mailto:d@domenic.me">d@domenic.me</a>) with substantial help from
吉野剛史 (Takeshi Yoshino, <a href="https://google.com">Google</a>,
<a href="mailto:tyoshino@chromium.org">tyoshino@chromium.org</a>).

Per <a href="https://creativecommons.org/publicdomain/zero/1.0/">CC0</a>, to the extent possible under law, the editor has waived all copyright and related or neighboring rights to this work.