@portable/streams

Functions

`Streams.abortable`

Wraps a transformer so that aborting signal errors its TransformStream with signal.reason. Erroring the stream makes pipeThrough cancel the source. The listener is removed when the stream ends, is cancelled, or fails because the transformer threw. Without a signal, the transformer is returned unchanged.

function abortable<input, output>(
  transformer: abortable.Transformer<input, output>,
  options: abortable.Options = {},
): abortable.Transformer<input, output>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the stream with `signal.reason` and cancels its source when aborted.

import * as Streams from '@portable/streams';
 
class UpperCaseStream extends TransformStream<string, string> {
  constructor(options: { signal?: AbortSignal | undefined } = {}) {
    super(
      Streams.abortable(
        {
          transform: (chunk, controller) =>
            controller.enqueue(chunk.toUpperCase()),
        },
        options,
      ),
    );
  }
}
 
text.pipeThrough(
  new UpperCaseStream({ signal: AbortSignal.timeout(5000) }),
);

`Streams.toArray`

Reads a stream or iterable to the end and returns its chunks in an array.

async function toArray<value>(
  source: Source<value>,
  options: toArray.Options = {},
): Promise<value[]>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Rejects with `signal.reason` and cancels the source when aborted.

import * as Streams from '@portable/streams';
import * as Lines from '@portable/lines';
 
const lines = await Streams.toArray(
  response.body!.pipeThrough(new Lines.DecoderStream()),
);

`Streams.toBlob`

Reads a stream or iterable to the end and returns its bytes in a Blob. String chunks are encoded as UTF-8. Unlike new Response(stream).blob(), it takes a maxBytes limit and a signal, and reads async iterables.

async function toBlob(
  source: Source<Uint8Array | string>,
  options: toBlob.Options = {},
): Promise<Blob>;

Option	Type	Default	Description
`maxBytes`	`number`	`Infinity`	Fail with `MaxBytesError` and cancel the source once it yields more bytes than this.
`signal`	`AbortSignal`		Rejects with `signal.reason` and cancels the source when aborted.
`type`	`string`	`''`	The `Blob`'s `type`, such as `'image/png'`.

import * as Streams from '@portable/streams';
 
const response = await fetch('https://example.com/image.png');
const blob = await Streams.toBlob(response.body!, {
  maxBytes: 10_000_000,
  type: 'image/png',
});

`Streams.toBytes`

Reads a stream or iterable to the end and returns its bytes in one Uint8Array. String chunks are encoded as UTF-8.

async function toBytes(
  source: Source<Uint8Array | string>,
  options: toBytes.Options = {},
): Promise<Uint8Array<ArrayBuffer>>;

Option	Type	Default	Description
`maxBytes`	`number`	`Infinity`	Fail with `MaxBytesError` and cancel the source once it yields more bytes than this.
`signal`	`AbortSignal`		Rejects with `signal.reason` and cancels the source when aborted.

import * as Streams from '@portable/streams';
 
const response = await fetch('https://example.com/image.png');
const bytes = await Streams.toBytes(response.body!, {
  maxBytes: 10_000_000,
});

`Streams.toJson`

Reads a stream or iterable to the end and parses it as JSON.

async function toJson(
  source: Source<Uint8Array | string>,
  options: toJson.Options = {},
): Promise<unknown>;

Option	Type	Default	Description
`encoding`	`string`	`'utf-8'`	Text encoding of byte chunks, as a `TextDecoder` label.
`maxBytes`	`number`	`Infinity`	Fail with `MaxBytesError` and cancel the source once it yields more bytes than this. String chunks count as their UTF-8 length.
`signal`	`AbortSignal`		Rejects with `signal.reason` and cancels the source when aborted.

import * as Streams from '@portable/streams';
 
const body = await Streams.toJson(request.body!, { maxBytes: 100_000 });

`Streams.toText`

Reads a stream or iterable to the end and decodes it as text. String chunks are kept as they are, and byte chunks are decoded with TextDecoder, so a character split across chunks decodes correctly.

async function toText(
  source: Source<Uint8Array | string>,
  options: toText.Options = {},
): Promise<string>;

Option	Type	Default	Description
`encoding`	`string`	`'utf-8'`	Text encoding of byte chunks, as a `TextDecoder` label.
`maxBytes`	`number`	`Infinity`	Fail with `MaxBytesError` and cancel the source once it yields more bytes than this. String chunks count as their UTF-8 length.
`signal`	`AbortSignal`		Rejects with `signal.reason` and cancels the source when aborted.

import * as Streams from '@portable/streams';
 
const response = await fetch('https://example.com/');
const html = await Streams.toText(response.body!, {
  maxBytes: 1_000_000,
});

`Streams.concat`

Joins sources end to end: every chunk of the first source, then every chunk of the next. A source is read only once the one before it has ended. An error in any source errors the output and cancels the sources not yet read, and cancelling the output cancels them all.

function concat<chunk>(
  sources: readonly Source<chunk>[],
  options: concat.Options = {},
): ReadableStream<chunk>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the output with `signal.reason` and cancels every source when aborted.

import * as Streams from '@portable/streams';
 
const body = Streams.concat([[header], file.stream(), [footer]]);

`Streams.limitBytes`

A TransformStream that passes chunks through unchanged and fails with MaxBytesError once more than maxBytes bytes have passed. The chunk that crosses the limit is not passed on, and the source is cancelled. String chunks count as their UTF-8 length.

function limitBytes<chunk extends Uint8Array | string>(
  maxBytes: number,
  options: limitBytes.Options = {},
): TransformStream<chunk, chunk>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the stream with `signal.reason` and cancels its source when aborted.

import * as Streams from '@portable/streams';
 
await request
  .body!.pipeThrough(Streams.limitBytes(1_000_000))
  .pipeTo(upload);

`Streams.merge`

Interleaves sources: reads all of them at once and passes each chunk on as it arrives. The output ends when every source has ended. An error in any source errors the output and cancels the others, and cancelling the output cancels them all.

function merge<chunk>(
  sources: readonly Source<chunk>[],
  options: merge.Options = {},
): ReadableStream<chunk>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the output with `signal.reason` and cancels every source when aborted.

import * as Streams from '@portable/streams';
 
const events = Streams.merge([serverA.body!, serverB.body!], {
  signal,
});

`Streams.fromIterable`

Creates a ReadableStream from an iterable or async iterable, such as an array or a generator. The stream pulls one value each time its reader asks for one, so a slow reader slows the generator down. Cancelling the stream calls the iterator's return(), which runs the generator's finally blocks. Uses ReadableStream.from where the runtime has it.

function fromIterable<chunk>(
  iterable: Iterable<chunk | PromiseLike<chunk>> | AsyncIterable<chunk>,
): ReadableStream<chunk>;

import * as Streams from '@portable/streams';
 
async function* countdown() {
  for (let i = 3; i > 0; i--) {
    yield `${i}\n`;
  }
}
 
return new Response(
  Streams.fromIterable(countdown()).pipeThrough(
    new TextEncoderStream(),
  ),
);

`Streams.toAsyncIterable`

Returns an async iterator over a ReadableStream, for for await loops in runtimes whose streams aren't async iterable yet. Leaving the loop early with break, return or a throw cancels the stream, or with preventCancel: true only releases its lock. Uses the stream's own values() where the runtime has it.

function toAsyncIterable<chunk>(
  stream: ReadableStream<chunk>,
  options: toAsyncIterable.Options = {},
): AsyncIterableIterator<chunk>;

Option	Type	Default	Description
`preventCancel`	`boolean`	`false`	Release the stream's lock instead of cancelling it when the loop exits early.

import * as Streams from '@portable/streams';
 
for await (const chunk of Streams.toAsyncIterable(response.body!)) {
  if (done(chunk)) {
    break; // cancels the response body
  }
}

`Streams.transform`

Builds a transform from an async generator function. The function receives the input chunks as an async iterable and yields output chunks, so state that spans chunks is a local variable. Use the result with pipeThrough.

Output is pulled, so the generator runs only as fast as the reader reads. An error thrown by the generator errors the output and cancels the input. Cancelling the output ends the input iterable and calls the generator's return(), so its finally blocks run.

function transform<input, output>(
  fn: (source: AsyncIterable<input>) => AsyncIterable<output>,
  options: transform.Options = {},
): ReadableWritablePair<output, input>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the output with `signal.reason`, cancels the input and ends the generator when aborted.

import * as Streams from '@portable/streams';
 
const sum = Streams.transform(async function* (
  source: AsyncIterable<number>,
) {
  let total = 0;
  for await (const n of source) {
    yield (total += n);
  }
});
 
numbers.pipeThrough(sum);

`Streams.batch`

A TransformStream that groups chunks into arrays of size. The last array holds whatever is left and may be shorter.

function batch<chunk>(
  size: number,
  options: batch.Options = {},
): TransformStream<chunk, chunk[]>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the stream with `signal.reason` and cancels its source when aborted.

import * as Streams from '@portable/streams';
 
await rows
  .pipeThrough(Streams.batch(500))
  .pipeTo(new WritableStream({ write: insertMany }));

`Streams.filter`

A TransformStream that passes on only the chunks for which fn(chunk, index) returns true. fn may return a promise.

function filter<chunk, state extends chunk>(
  fn: (chunk: chunk, index: number) => chunk is state,
  options?: filter.Options,
): TransformStream<chunk, state>;
function filter<chunk>(
  fn: (chunk: chunk, index: number) => boolean | PromiseLike<boolean>,
  options?: filter.Options,
): TransformStream<chunk, chunk>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the stream with `signal.reason` and cancels its source when aborted.

import * as Streams from '@portable/streams';
 
lines.pipeThrough(Streams.filter((line) => line !== ''));

`Streams.map`

A TransformStream that replaces each chunk with fn(chunk, index). If fn returns a promise, the result is passed on once it resolves, and chunks stay in order.

function map<input, output>(
  fn: (chunk: input, index: number) => output | PromiseLike<output>,
  options: map.Options = {},
): TransformStream<input, output>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the stream with `signal.reason` and cancels its source when aborted.

import * as Streams from '@portable/streams';
import * as Lines from '@portable/lines';
 
const records = response
  .body!.pipeThrough(new Lines.DecoderStream())
  .pipeThrough(Streams.map((line) => JSON.parse(line)));

`Streams.tap`

A TransformStream that calls fn(chunk, index) for each chunk and passes the chunk on unchanged. If fn returns a promise, the chunk waits for it.

function tap<chunk>(
  fn: (chunk: chunk, index: number) => void | PromiseLike<void>,
  options: tap.Options = {},
): TransformStream<chunk, chunk>;

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the stream with `signal.reason` and cancels its source when aborted.

import * as Streams from '@portable/streams';
 
let received = 0;
response.body!.pipeThrough(
  Streams.tap((chunk) => {
    received += chunk.byteLength;
  }),
);

`Streams.tee`

Splits a source into count streams that each receive every chunk. The source is read only as fast as the slowest branch allows: once a branch holds highWaterMark unread chunks or bytes, the source isn't read until that branch reads. Cancelling a branch removes it, and the source is cancelled only when every branch is cancelled, with the last branch's reason. An error in the source errors every branch. Aborting signal errors every branch with its reason and cancels the source.

Branches receive the same chunk objects, as with ReadableStream.prototype.tee(), so a branch must copy a chunk (chunk.slice()) before changing it.

function tee<chunk>(
  source: Source<chunk>,
  count: number,
  options: tee.Options = {},
): ReadableStream<chunk>[];

Option	Type	Default	Description
`highWaterMark`	`number`	`16 chunks, or 65536 bytes`	How much the slowest branch may hold unread before the source stops being read, in `unit`s. The fastest branch can be this far ahead of the slowest.
`unit`	`"chunks" \| "bytes"`	`'chunks'`	What `highWaterMark` counts: `'chunks'`, or `'bytes'` for `Uint8Array` and string chunks, with strings counted as their UTF-8 length. With `'bytes'`, any other chunk errors every branch with a `TypeError`.
`signal`	`AbortSignal`		Aborting errors every branch with `signal.reason` and cancels the source.

import * as Streams from '@portable/streams';
 
const [forExif, forStorage, forResize] = Streams.tee(request.body!, 3, {
  highWaterMark: 1_000_000,
  unit: 'bytes',
});

Errors

`Streams.MaxBytesError`

Thrown when a stream has more bytes than maxBytes allows.

class MaxBytesError extends Errors.MaxBytesError {
  constructor(options: Errors.MaxBytesError.Options);
  readonly kind: string;
}

import * as Streams from '@portable/streams';
 
const response = await fetch('https://example.com/upload');
try {
  await Streams.toBytes(response.body!, { maxBytes: 1_000_000 });
} catch (error) {
  if (error instanceof Streams.MaxBytesError) {
    console.log(`larger than ${error.maxBytes} bytes`);
  }
}

Types

`Streams.Source`

Anything the collectors and combinators read from: a ReadableStream, an async iterable or an iterable.

type Source<chunk> =
  ReadableStream<chunk> | AsyncIterable<chunk> | Iterable<chunk>;