Cookbook: streaming pipelines

Cookbook: streaming pipelines

Consume records without a pipeline

When you just need the records, skip the machinery — io.records parses JSONL / json-seq for you:

for await (const user of io.records.get('https://api.example.com/users/export')) {
  process(user);
}

Subscribe to Server-Sent Events

io.sse is the push counterpart — parsed events, automatic reconnects, Last-Event-ID:

for await (const event of io.sse('https://api.example.com/notifications')) {
  handle(JSON.parse(event.data));
}

Everything below is the stream-chain layer — byte streams in and out of io.stream, with real pipeline processing in between.

Recipes for streaming bodies and building processing pipes with stream-chain. Import its Web build, stream-chain/web. Background: Streaming.

Download a large response as a stream

const body = await io.stream.get('https://example.com/big.ndjson');
for await (const chunk of body) process(chunk); // Uint8Array chunks

Download straight to a file (CLI)

import {Readable} from 'node:stream';
import {createWriteStream} from 'node:fs';
import {pipeline} from 'node:stream/promises';

const body = await io.stream.get(url);
await pipeline(Readable.fromWeb(body), createWriteStream('out.bin'));

Upload a file / stream

Pass a Web ReadableStream (a Blob/File in the browser, or Readable.toWeb(...) on the CLI) as the body — set the type with as:

import {Readable} from 'node:stream';
import {createReadStream} from 'node:fs';

await io.put(url, Readable.toWeb(createReadStream('in.ndjson')), {as: 'ndjson'});

Process a stream and upload the result

Build the pipe, then hand it to io.put — a chain ({readable}) works as the body directly:

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

const pipe = chain([
  await io.stream.get(inputUrl),
  jsonlParser(),
  record => ((record.n += 1), record),
  jsonlStringer()
]);

const {etag} = await io.full.put(outputUrl, pipe, {as: 'ndjson'});

Server-processing round-trip (mid-pipe io.stream.put)

Use io.stream.put (or .post / .patch) inside the chain when the server transforms your upload and streams a response back that you keep processing. It returns a duplex {writable, readable, response}: writable = request up, readable = response down.

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
  jsonlParser(),
  jsonlStringer(),
  io.stream.put(processUrl, {as: 'ndjson'}), // upload → server → response streams on
  jsonlParser(),
  record => console.log('processed:', record)
]);

A non-2xx from io.stream.put errors the readable (so it propagates down the pipe); keep a reference to the duplex and await its .response for status/headers, or pass {ignoreBadStatus: true} to stream the error body through.

See also

• records · sse · Streaming · Core API · Requests & options