```js import { arrayBuffer, blob, buffer, json, text, } from 'node:stream/consumers';

const data1 = await arrayBuffer(getReadableStreamSomehow());

const data2 = await blob(getReadableStreamSomehow());

const data3 = await buffer(getReadableStreamSomehow());

const data4 = await json(getReadableStreamSomehow());

const data5 = await text(getReadableStreamSomehow()); ```

```js /* * For instance, given a ReadableStream object, the stream.Readable.fromWeb() method * will create an return a Node.js stream.Readable object that can be used to consume * the ReadableStream's data: / import { Readable } from 'node:stream';

const readable = new ReadableStream(getSomeSource());

const nodeReadable = Readable.fromWeb(readable);

nodeReadable.on('data', console.log); ```

```js /* * The adaptation can also work the other way -- starting with a Node.js * stream.Readable and acquiring a web streams ReadableStream: / import { Readable } from 'node:stream';

const readable = new Readable({ read(size) { reader.push(Buffer.from('hello')); } });

const readableStream = Readable.toWeb(readable);

await readableStream.read();

```

```js const response = await fetch("https://www.example.org"); let total = 0;

// Iterate response.body (a ReadableStream) asynchronously for await (const chunk of response.body) { // Do something with each chunk // Here we just accumulate the size of the response. total += chunk.length; }

// Do something with the total console.log(total); ```

js match (res) { if (isEmpty(res)) { … } when ({ pages, data }) if (pages > 1) { … } when ({ pages, data }) if (pages === 1) { … } else { … } }

js match (arithmeticStr) { when (/(?<left>\d+) \+ (?<right>\d+)/) as ({ groups: { left, right } }) { process(left, right); } when (/(?<left>\d+) \+ (?<right>\d+)/) { process(left, right); } // maybe? else { ... } }

```js class Name { static Symbol.matcher { const pieces = matchable.split(' '); if (pieces.length === 2) { return { matched: true, value: pieces }; } } }

match ('Tab Atkins-Bittner') { when (^Name with [first, last]) if (last.includes('-')) { … } when (^Name with [first, last]) { … } else { ... } } ```

js const res = await fetch(jsonService) match (res) { when ({ status: 200, headers: { 'Content-Length': s } }) { console.log(`size is ${s}`); } when ({ status: 404 }) { console.log('JSON not found'); } when ({ status }) if (status >= 400) { throw new RequestError(res); } };

```js function todoApp(state = initialState, action) { return match (action) { when ({ type: 'set-visibility-filter', payload: visFilter }) { ({ ...state, visFilter }); } when ({ type: 'add-todo', payload: text }) { ({ ...state, todos: [...state.todos, { text, completed: false }] }); } when ({ type: 'toggle-todo', payload: index }) { const newTodos = state.todos.map((todo, i) => { return i !== index ? todo : { ...todo, completed: !todo.completed }; });

  ({
    ...state,
    todos: newTodos,
  });
}
else { state } // ignore unknown actions

} } ```

jsx <Fetch url={API_URL}> {props => match (props) { when ({ loading }) { <Loading />; } when ({ error }) { <Error error={error} />; } when ({ data }) { <Page data={data} />; } }} </Fetch>

```js import type { EntryContext } from "@remix-run/cloudflare"; import { RemixServer } from "@remix-run/react"; import isbot from "isbot"; import { renderToReadableStream } from "react-dom/server";

const ABORT_DELAY = 5000;

const handleRequest = async ( request: Request, responseStatusCode: number, responseHeaders: Headers, remixContext: EntryContext ) => { let didError = false;

const stream = await renderToReadableStream( <RemixServer context={remixContext} url={request.url} abortDelay={ABORT_DELAY} />, { onError: (error: unknown) => { didError = true; console.error(error);

    // You can also log crash/error report
  },
  signal: AbortSignal.timeout(ABORT_DELAY),
}

);

if (isbot(request.headers.get("user-agent"))) { await stream.allReady; }

responseHeaders.set("Content-Type", "text/html"); return new Response(stream, { headers: responseHeaders, status: didError ? 500 : responseStatusCode, }); };

export default handleRequest; ```

js const wss = new WebSocketStream(url); const { readable } = await wss.connection; const reader = readable.getReader(); while (true) { const { value, done } = await reader.read(); if (done) break; await process(value); } done();

js const wss = new WebSocketStream(url); const { writable } = await wss.connection; const writer = writable.getWriter(); for await (const message of messages) { await writer.write(message); }

js const controller = new AbortController(); const wss = new WebSocketStream(url, { signal: controller.signal }); setTimeout(() => controller.abort(), 1000);

```js const supportsRequestStreams = (() => { let duplexAccessed = false;

const hasContentType = new Request('', { body: new ReadableStream(), method: 'POST', get duplex() { duplexAccessed = true; return 'half'; }, }).headers.has('Content-Type');

return duplexAccessed && !hasContentType; })();

if (supportsRequestStreams) { // … } else { // … } ```

```js /* * Fetch and process the stream / async function process() { // Retrieve NDJSON from the server const response = await fetch('http://localhost:3000/request');

const results = response.body // From bytes to text: .pipeThrough(new TextDecoderStream()) // Buffer until newlines: .pipeThrough(splitStream('\n')) // Parse chunks as JSON: .pipeThrough(parseJSON());

// Loop through the results and write to the DOM writeToDOM(results.getReader()); }

/* * Read through the results and write to the DOM * @param {object} reader / function writeToDOM(reader) { reader.read().then(({ value, done }) => { if (done) { console.log("The stream was already closed!");

} else {
  // Build up the values
  let result = document.createElement('div');
  result.innerHTML =
    `<div>ID: ${value.id} - Phone: ${value.phone} - Result: $
        {value.result}</div><br>`;

  // Prepend to the target
  targetDiv.insertBefore(result, targetDiv.firstChild);

  // Recursively call
  writeToDOM(reader);
}

}, e => console.error("The stream became errored and cannot be read from!", e) ); } ```

Node.js

js import { renderToPipeableStream } from "react-dom/server.node"; import React from "react"; import http from "http"; const App = () => ( <html> <body> <h1>Hello World</h1> <p>This is an example.</p> </body> </html> ); var didError = false; http .createServer(function (req, res) { const stream = renderToPipeableStream(<App />, { onShellReady() { res.statusCode = didError ? 500 : 200; res.setHeader("Content-type", "text/html"); res.setHeader("Cache-Control", "no-transform"); stream.pipe(res); }, onShellError(error) { res.statusCode = 500; res.send( '<!doctype html><p>Loading...</p><script src="clientrender.js"></script>', ); }, onAllReady() { }, onError(err) { didError = true; console.error(err); }, }); }) .listen(3000);

Deno

```js import { renderToReadableStream } from "https://esm.run/react-dom/server"; import * as React from "https://esm.run/react";

const App = () => ( <html> <body>

Hello World

This is an example.

const headers = { headers: { "Content-Type": "text/html", "Cache-Control": "no-transform", }, };

Deno.serve( async (req) => { return new Response(await renderToReadableStream(<App />), headers); }, { port: 3000 }, ); ```

Bun

```js import { renderToReadableStream } from "react-dom/server"; const headers = { headers: { "Content-Type": "text/html", }, };

const App = () => ( <html> <body>

Hello World

This is an example.

Bun.serve({ port: 3000, async fetch(req) { return new Response(await renderToReadableStream(<App />), headers); }, }); ```