Concepts: streaming

Streaming

double-meh streams over Web streams (ReadableStream / WritableStream) in both directions — the same API in the browser and on the CLI, and it composes directly with stream-chain. There is no node:stream code; bridge to Node streams with Readable.fromWeb / Readable.toWeb when you need to.

Streaming has two independent axes:

Axis	How you turn it on
Response (read the body as a stream)	the stream: true option, or the io.stream namespace
Request (send a stream as the body)	pass a ReadableStream / Blob / File as the body — no flag

They compose freely: one axis each on a write = stream up and down (that is what the io.stream write duplexes give you).

Response streaming

{stream: true} makes the pipeline hand back the raw response.body instead of parsing it:

const body = await io.get(url, null, {stream: true}); // a ReadableStream

io.stream.get is the GET-only sugar for that — no null placeholder:

const body = await io.stream.get(url); // ≡ io.get(url, null, {stream: true})
const rows = await io.stream.get(url, {page: 2}); // query still composes

io.stream is a return-shape namespace alongside io.full. get is its only read verb: HEAD/OPTIONS/DELETE have no body to stream, and write verbs take the body positionally, so io.post(url, data, {stream: true}) already streams the response with no placeholder. Its write verbs — io.stream.put/post/patch — are the request/response duplexes below. track and cache stand down for streamed responses (a stream is single-consumer).

Request streaming

Pass a Web ReadableStream (or Blob / File / FormData) as the body and it streams up — the fetch transport sets duplex: 'half' automatically, and retry stands down (a stream cannot be replayed):

await io.put(url, fileStream, {as: 'octet'}); // Content-Type application/octet-stream

A streamed body carries no inherent Content-Type (unlike a Blob), so set one — the terse way is the as shorthand ({as: 'ndjson'}, or a full type {as: 'application/x-ndjson'}). See Requests & options.

io.stream.put/post/patch — the request/response duplex

Request-body streaming for writes lives on the same namespace — io.stream.put / io.stream.post / io.stream.patch (url, options?). Each returns a Web duplex synchronously, so it drops straight into a stream-chain pipe as a node:

import io from 'double-meh';

const up = io.stream.put('https://example.com/data', {as: 'ndjson'});
// up.writable → the request body streams up (duplex:'half')
// up.readable → the streamed response body
// up.response → a promise for the envelope (status/headers), resolved at headers-time

• No data arg — the body arrives through the pipe, so the 2nd positional is options.
• Non-2xx errors up.readable and rejects up.response; pass {ignoreBadStatus: true} to stream the error-response body through instead.
• Response streaming is intrinsic — a duplex never takes {stream: true}.
• There is no read-verb duplex (GET has no body to stream — use io.stream.get).

{readable} body passthrough

io.put accepts a stream-chain chain (or any {readable} duplex) directly and uses its .readable as the body — so you can terminate a pipeline into an upload:

const pipe = chain([source, transform, jsonlStringer()]);
await io.put(url, pipe, {as: 'ndjson'}); // or io.put(url, pipe.readable, …)

Records and events — the framed layer

Byte streams are the floor; two helpers consume framed streams directly:

• io.records.get/post — a lazy async iterable of parsed JSONL / json-seq records (for await (const record of io.records.get(url))).
• io.sse — Server-Sent Events with EventSource-style reconnects, riding the pipeline (auth headers, POST, inspectors on every reconnect).

They cover consuming simply. Record processing pipelines and the encode/upload side are stream-chain's job (below) — double-meh deliberately duplicates none of it.

With stream-chain

Import from stream-chain/web (its Web-streams build) and stream-chain/web/jsonl (the JSONL factories). io.stream.get is a pipe source; io.stream.put is a mid-pipe duplex (upload, then the server's streamed response flows on):

import {chain} from 'stream-chain/web';
import {jsonlParser, jsonlStringer} from 'stream-chain/web/jsonl';
import io from 'double-meh';

const pipe = chain([
  await io.stream.get(inputUrl), // source: streamed GET
  jsonlParser(), // bytes → records
  record => ((record.ts = Date.now()), record),
  jsonlStringer(), // records → bytes
  io.stream.put(outputUrl, {as: 'ndjson'}) // duplex: upload, response streams back
]);

More recipes: Cookbook: streaming pipelines.

See also

• records · sse · Core API · Requests & options · fetch transport