Streams Standard

https://github.com/whatwg/streams
Admin
The Streams Standard defines a common model for representing streaming data with readable, writable,...
Tokens:19,170
Snippets:134
Trust Score:9.6
Update:1 day ago
Show doc for...
Tokens:
Raw
### Setup and helper functions for the append child demo

Source: https://github.com/whatwg/streams/blob/main/demos/append-child.html

Initializes DOM references, defines field names, and provides a function to create a table element with headers. Used before the main streaming logic.

```javascript
'use strict';
const loadButton = document.querySelector('#load');
const resetButton = document.querySelector('#reset');
const targetDiv = document.querySelector('#target');
const FIELDS = ['Hash', 'Date', 'Author', 'Subject'];
const FIELDS_LOWERCASE = FIELDS.map(string => string.toLowerCase());
function createTable(parentElement) {
  const table = document.createElement('table');
  table.id = 'commits';
  const tr = document.createElement('tr');
  for (const heading of FIELDS) {
    const th = document.createElement('th');
    th.textContent = heading;
    tr.appendChild(th);
  }
  table.appendChild(tr);
  parentElement.appendChild(table);
  return table;
}
```

--------------------------------

### Manual stream consumption with reader.read()

Source: https://github.com/whatwg/streams/blob/main/explainers/readable-stream-async-iteration.md

Shows the traditional way to consume a ReadableStream by acquiring a reader and repeatedly calling read() in a loop. This is the baseline example that the async iteration feature simplifies.

```javascript
async function getResponseSize(url) {
  const response = await fetch(url);
  const reader = response.body.getReader();
  let total = 0;

  while (true) {
    const {done, value} = await reader.read();
    if (done) return total;
    total += value.length;
  }
}
```

--------------------------------

### Simplified stream consumption with for await...of

Source: https://github.com/whatwg/streams/blob/main/explainers/readable-stream-async-iteration.md

Rewrites the original example using for await...of over the response body, eliminating the need for an explicit reader. This is the primary benefit of the async iteration support.

```javascript
async function getResponseSize(url) {
  const response = await fetch(url);
  let total = 0;
  for await (const chunk of response) {
    total += chunk.length;
  }
  return total;
}
```

--------------------------------

### Create a video processing pipe with background blur and logging

Source: https://github.com/whatwg/streams/blob/main/explainers/streams-for-raw-video.md

This example shows how to create a processing pipe for VideoFrame streams. It defines a custom TransformStream that logs every 30 frames and uses a native BackgroundBlurTransform. The stream types are set to 'owning' to take ownership of the VideoFrame chunks. The videoFrame must be closed when no longer needed, and the controller.error(e) is used to error the pipe on unrecoverable exceptions.

```javascript
function doBackgroundBlurOnVideoFrames(videoFrameStream, doLogging)
{
  // JavaScript custom transform.
  let frameCount = 0;
  const frameCountTransform = new TransformStream({
    transform: async (videoFrame, controller) => {
      try {
        // videoFrame is under the responsibility of the script and must be closed when no longer needed.
        controller.enqueue(videoFrame, { transfer: [videoFrame] });
        // controller.enqueue was called, videoFrame is transferred.
        if (!(++frameCount % 30) && doLogging)
            doLogging(frameCount);
      } catch (e) {
        // In case of exception, let's make sure videoFrame is closed. This is a no-op if videoFrame was previously transferred.
        videoFrame.close();
        // If exception is unrecoverable, let's error the pipe.
        controller.error(e);
      }
    },
    readableType: 'owning',
    writableType: 'owning'
  });
  // Native transform is of type 'owning'
  const backgroundBlurTransform = new BackgroundBlurTransform();

  return videoFrameStream.pipeThrough(backgroundBlurTransform)
                         .pipeThrough(frameCountTransform);
}
```

--------------------------------

### Pipe readable stream to writable stream

Source: https://github.com/whatwg/streams/blob/main/Requirements.md

Shows the primary way of consuming streams by piping a readable stream to a writable stream, queuing as little data as possible in memory. The first example pipes a file read stream to a file write stream, and the second pipes an HTTPS response to a file write stream.

```JavaScript
fs.createReadStream("source.txt")
    .pipeTo(fs.createWriteStream("dest.txt"));

https.get("https://example.com")
    .pipeTo(fs.createWriteStream("dest.txt"));
```

--------------------------------

### Transform streams via pipe chain

Source: https://github.com/whatwg/streams/blob/main/Requirements.md

Illustrates using transform streams in a pipe chain, where data is written to a readable/writable stream and transformed data is written out the other side. Examples include decompression, compression, video processing, and string decoding with database operations.

```JavaScript
fs.createReadStream("source.zip")
    .pipeThrough(zlib.createGzipDecompressor(options))
    .pipeTo(fs.createWriteStream("dest/"));

fs.createReadStream("index.html")
    .pipeThrough(zlib.createGzipCompressor(options))
    .pipeTo(httpServerResponse);

https.get("https://example.com/video.mp4")
    .pipeThrough(videoProcessingWebWorker)
    .pipeTo(document.query("video"));

fs.createReadStream("source.txt")
    .pipeThrough(new StringDecoder("utf-8"))
    .pipeThrough(database1.queryExecutor)
    .pipeTo(database2.tableWriter("table"));
```

--------------------------------

### Finding a byte in a stream with for await...of

Source: https://github.com/whatwg/streams/blob/main/explainers/readable-stream-async-iteration.md

Shows how to search for a specific character in a stream using for await...of. Note that breaking out of the loop automatically cancels the stream; use values({ preventCancel: true }) to prevent cancellation.

```javascript
async function example() {
  const find = 'J';
  const findCode = find.codePointAt(0);
  const response = await fetch('https://html.spec.whatwg.org');
  let bytes = 0;

  for await (const chunk of response.body) {
    const index = chunk.indexOf(findCode);

    if (index != -1) {
      bytes += index;
      console.log(`Found ${find} at byte ${bytes}.`);
      break;
    }

    bytes += chunk.length;
  }
}
```

--------------------------------

### Create a readable byte stream with an underlying pull source using byobRequest

Source: https://github.com/whatwg/streams/blob/main/explainers/byte-streams.md

Returns a ReadableStream with type "bytes" and a pull method that fills the developer-supplied buffer via controller.byobRequest, enabling zero-copy reading. The autoAllocateChunkSize is set to 1024, allowing the default reader to work automatically.

```javascript
const DEFAULT_CHUNK_SIZE = 1024;

function makeReadableByteStream() {
  return new ReadableStream({
    type: "bytes",

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

    autoAllocateChunkSize: DEFAULT_CHUNK_SIZE
  });
}
```

--------------------------------

### Button event handlers for load and reset

Source: https://github.com/whatwg/streams/blob/main/demos/append-child.html

Sets up click handlers for the load and reset buttons. Load disables itself and triggers the fetch pipeline; reset clears the target div and re-enables load.

```javascript
loadButton.onclick = () => {
  loadButton.disabled = true;
  setTimeout(fetchThenJSONToDOM, 0);
};
resetButton.onclick = () => {
  targetDiv.innerHTML = '';
  loadButton.disabled = false;
};
```

--------------------------------

### Fetch and pipe stream into DOM element with backpressure

Source: https://github.com/whatwg/streams/blob/main/demos/streaming-element-backpressure.html

When the load button is clicked, this code fetches a local HTML file and pipes the response body through a TextDecoderStream into the target element's writable side, applying backpressure. The button is disabled during the operation.

```javascript
'use strict';
const loadButton = document.querySelector('#load');
const resetButton = document.querySelector('#reset');
const targetDiv = document.querySelector('#target');

loadButton.onclick = async () => {
  loadButton.disabled = true;
  fetchDirectlyIntoDOM();
};

// BEGIN SOURCE TO VIEW
async function fetchDirectlyIntoDOM() {
  const response = await fetch('data/commits.include', {
    mode: 'same-origin',
    headers: {
      'Cache-Control': 'no-cache, no-store'
    }
  });
  await response.body
    .pipeThrough(new TextDecoderStream())
    .pipeTo(targetDiv.writable);
}
// END SOURCE TO VIEW

resetButton.onclick = () => {
  targetDiv.reset();
  loadButton.disabled = false;
};
```

--------------------------------

### pipeThrough sugar for pipeTo

Source: https://github.com/whatwg/streams/blob/main/FAQ.md

Shows that src.pipeThrough(through).pipeTo(dest) is equivalent to piping into the writable side and out the readable side of a duplex stream.

```javascript
src.pipeThrough(through).pipeTo(dest);
```

```javascript
src.pipeTo(through.writable);
through.readable.pipeTo(dest);
```

--------------------------------

### Enable debug output for streams tests

Source: https://github.com/whatwg/streams/blob/main/reference-implementation/README.md

Sets the DEBUG environment variable to 'streams:*' to enable diagnostic output from the reference implementation when running tests. Useful for understanding asynchronous behavior.

```bash
DEBUG=streams:* npm test
```

--------------------------------

### Transfer a ReadableStream to a Worker

Source: https://github.com/whatwg/streams/blob/main/explainers/transferable-streams.md

Creates a ReadableStream and transfers it to a worker using postMessage(rs, [rs]). The stream must be transferred, not cloned; cloning throws a DataCloneError.

```javascript
const rs = new ReadableStream({
  start(controller) {
    controller.enqueue('hello');
  }
});
const w = new Worker('worker.js');
w.postMessage(rs, [rs]);
```

```javascript
onmessage = async (evt) => {
  const rs = evt.data;
  const reader = rs.getReader();
  const {value, done} = await reader.read();
  console.log(value); // logs 'hello'.
};
```

--------------------------------

### Append child WritableStream and fetch pipeline

Source: https://github.com/whatwg/streams/blob/main/demos/append-child.html

Defines a WritableStream that appends chunks to a parent node, and an async function that fetches JSON, transforms it into table rows, and pipes it to the stream. Includes a TransformStream for converting JSON objects to DOM elements.

```javascript
// BEGIN SOURCE TO VIEW
function appendChildWritableStream(parentNode) {
  return new WritableStream({
    write(chunk) {
      parentNode.appendChild(chunk);
    }
  });
}
async function fetchThenJSONToDOM() {
  const jsonToElementTransform = new TransformStream({
    transform(chunk, controller) {
      const tr = document.createElement('tr');
      for (const cell of FIELDS_LOWERCASE) {
        const td = document.createElement('td');
        td.textContent = chunk[cell];
        tr.appendChild(td);
      }
      controller.enqueue(tr);
    }
  });
  const response = await fetch('data/commits.json', {
    mode: 'same-origin',
    headers: {
      'Cache-Control': 'no-cache, no-store'
    }
  });
  const table = createTable(targetDiv);
  return response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(splitStream('\n'))
    .pipeThrough(parseJSON())
    .pipeThrough(jsonToElementTransform)
    .pipeTo(appendChildWritableStream(table));
}
// END SOURCE TO VIEW
```

--------------------------------

### Abort a fetch POST using controller.signal

Source: https://github.com/whatwg/streams/blob/main/explainers/writable-stream-abort-signal.md

Each write() triggers a POST to a remote endpoint using the fetch API. The controller.signal is passed to fetch so that writer.abort() aborts the ongoing request.

```javascript
const endpoint = 'https://endpoint/api';
const ws = new WritableStream({
  async write(controller, chunk) {
    const response = await fetch(endpoint, { signal: controller.signal,
                                             method: 'POST',
                                             body: chunk });
    await response.text();
  }
});
const writer = ws.getWriter();

writer.write('some data');
await writer.abort();
```

--------------------------------

### Update web-platform-tests submodule

Source: https://github.com/whatwg/streams/blob/main/reference-implementation/README.md

Command to update the web-platform-tests submodule to the latest remote commit. This is used when landing test changes that are maintained in the upstream repository.

```bash
git submodule update --remote web-platform-tests
```

--------------------------------

### Read bytes from a stream into a single memory buffer using a BYOB reader

Source: https://github.com/whatwg/streams/blob/main/explainers/byte-streams.md

Acquires a BYOB reader and reads the first 1024 bytes into a single ArrayBuffer, avoiding copies. After execution, the original startingAB is detached and cannot be used, but the returned buffer points to the same memory region.

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

let startingAB = new ArrayBuffer(1024);
const buffer = await readInto(startingAB);
console.log("The first 1024 bytes: ", buffer);

async function readInto(buffer) {
  let offset = 0;

  while (offset < buffer.byteLength) {
    const {value: view, done} =
     await reader.read(new Uint8Array(buffer, offset, buffer.byteLength - offset));
    buffer = view.buffer;
    if (done) {
      break;
    }
    offset += view.byteLength;
  }

  return buffer;
}
```

--------------------------------

### Stream fetched data directly into a DOM element

Source: https://github.com/whatwg/streams/blob/main/demos/streaming-element.html

Uses fetch with same-origin mode and cache-control headers, then pipes the response body through a TextDecoderStream into targetDiv.writable. The load button is disabled during streaming.

```javascript
'use strict'; const loadButton = document.querySelector('#load'); const resetButton = document.querySelector('#reset'); const targetDiv = document.querySelector('#target'); // BEGIN SOURCE TO VIEW async function streamDirectlyIntoDOM() { const response = await fetch('data/commits.include', { mode: 'same-origin', headers: { 'Cache-Control': 'no-cache, no-store' } }); await response.body .pipeThrough(new TextDecoderStream()) .pipeTo(targetDiv.writable); } loadButton.onclick = () => { loadButton.disabled = true; streamDirectlyIntoDOM(); }; // END SOURCE TO VIEW resetButton.onclick = () => { targetDiv.reset(); loadButton.disabled = false; };
```

--------------------------------

### ReadableStream.prototype[Symbol.asyncIterator]()

Source: https://github.com/whatwg/streams/blob/main/explainers/readable-stream-async-iteration.md

Same as values(). This method makes ReadableStream adhere to the ECMAScript AsyncIterable protocol, enabling for await...of loops.

```APIDOC
## ReadableStream.prototype[Symbol.asyncIterator]()

### Description
Same as `values()`. This method makes `ReadableStream` adhere to the ECMAScript AsyncIterable protocol, enabling `for await...of` loops to work.

### Method
Symbol.asyncIterator

### Endpoint
ReadableStream.prototype[Symbol.asyncIterator]()

### Parameters
None.

### Request Example
```javascript
for await (const chunk of stream) {
  // process chunk
}
```

### Response
#### Success Response
Returns an AsyncIterator object.

#### Response Example
```javascript
const iterator = stream[Symbol.asyncIterator]();
const { value, done } = await iterator.next();
```
```
Tokens:
Raw