@portable/streams

Collects, combines and transforms web streams and iterables, and makes your own transforms cancellable. It works without ReadableStream.from and for await over a stream, which some browsers lack. Runs in browsers, Cloudflare Workers, Deno, Bun and Node 24+.

Replaces get-stream, concat-stream, bl (as a collector), stream-to-array, merge2, merge-stream, through2 and the limit handling of raw-body.

npm install @portable/streams

import * as Streams from '@portable/streams';
 
async function* rows() {
  yield 'a';
  yield 'b';
}
 
const upper = Streams.transform(async function* (
  source: AsyncIterable<string>,
) {
  for await (const chunk of source) {
    yield chunk.toUpperCase();
  }
});
 
await Streams.toText(Streams.fromIterable(rows()).pipeThrough(upper)); // 'AB'

Collect

Each collector reads a ReadableStream, an async iterable (such as a Node Readable) or an iterable to the end. It releases the stream's lock when it finishes, and cancels the stream if it stops early.

toBytes, toBlob, toText and toJson take Uint8Array and string chunks, and reject with a TypeError for any other chunk. When the source yields more than maxBytes bytes (string chunks count as their UTF-8 length), the collector cancels the source and rejects with MaxBytesError, whose maxBytes property holds the limit. Aborting signal cancels the source and rejects with signal.reason.

const body = await Streams.toJson(request.body!, { maxBytes: 100_000 });

Combine

Sources are streams, async iterables or iterables. An error in one source errors the output and cancels the others, and cancelling the output cancels every source.

tee splits in the other direction, and differs from ReadableStream.prototype.tee() in three ways:

• It makes any number of branches. The platform's tee() makes two, so three branches need a chain of two calls.
• It reads the source only as fast as the slowest branch. Once a branch holds highWaterMark unread chunks, tee stops reading the source until that branch reads, so the fastest branch is at most that far ahead. The platform's tee() keeps reading and buffers without limit for a branch that isn't read. highWaterMark defaults to 16 chunks. With unit: 'bytes' it counts bytes, as maxBytes does, and defaults to 65536.
• Cancelling a branch removes it, and tee then reads only as fast as the slowest remaining branch. The source is cancelled when every branch is cancelled, with the reason from the last one. The platform's tee() does the same, with both reasons in an array.

An error in the source errors every branch. Aborting signal errors every branch with signal.reason and cancels the source. Branches receive the same chunk objects, as with the platform's tee() on a stream that isn't a byte stream, so copy a chunk with chunk.slice() before changing it. Because of the limit, reading one branch to the end before reading the next stops after highWaterMark chunks. Read the branches at the same time, or pass highWaterMark: Infinity.

const upload = (await request.formData()).get('file') as File;
const [forExif, forStorage, forResize] = Streams.tee(
  upload.stream(),
  3,
  {
    highWaterMark: 1_000_000,
    unit: 'bytes',
  },
);
await Promise.all([
  Exif.parseStream(forExif), // cancels forExif once it has the metadata
  bucket.put(key, forStorage),
  bucket.put(`${key}.thumb`, resize(forResize)),
]);

Bridge streams and iterables

Aborting signal errors a transform with signal.reason, and pipeThrough then cancels its source. Aborting the signal of concat or merge errors the output with signal.reason and cancels every source. fromIterable and toAsyncIterable take no signal: cancel the stream they return, or pass signal to the pipeTo that reads it.

const lines = Streams.transform(async function* (
  source: AsyncIterable<string>,
) {
  let rest = '';
  for await (const chunk of source) {
    const parts = (rest + chunk).split('\n');
    rest = parts.pop()!;
    yield* parts;
  }
  if (rest) {
    yield rest;
  }
});

Cancel your own TransformStream

abortable(transformer, { signal }) wraps a Transformer so that aborting signal errors its TransformStream with signal.reason, and pipeThrough then cancels the source. It removes its listener on the signal when the stream ends, is cancelled, or fails because the transformer threw, and returns the transformer unchanged when there is no signal. The stream transforms in lines, base64, csv and the other Portable packages use it to support their signal option.

class UpperCaseStream extends TransformStream<string, string> {
  constructor(options: { signal?: AbortSignal | undefined } = {}) {
    super(
      Streams.abortable(
        {
          transform: (chunk, controller) =>
            controller.enqueue(chunk.toUpperCase()),
        },
        options,
      ),
    );
  }
}

Types

Why use this over get-stream, concat-stream and merge2

The libraries above were written for Node streams. This package reads web streams with getReader(), so it works in every runtime listed above, and it has no dependencies outside @portable/*.

• toBytes collected 6.8 MB of web stream chunks about 7.6× faster than get-stream, and about 1.4× faster than Response.arrayBuffer(). toBlob ran about 1.7× faster than Response.blob(), and toText ran at the same speed as get-stream.
• maxBytes cancels the source as soon as the limit is crossed. get-stream's maxBuffer kept buffering (get-stream #42), and raw-body's limit became NaN for ArrayBuffer chunks, which turned it off.
• An error in any source of concat or merge errors the output and cancels the other sources. merge-stream and merge2 didn't propagate errors (merge-stream #24, merge2 #7), and concat-stream never called back when the stream errored (concat-stream #50).
• Collectors don't need async iteration over ReadableStream, which get-stream needed and Chrome lacked (get-stream #116).
• The collectors reject a number chunk with a TypeError. In concat-stream it exposed uninitialized Buffer memory (concat-stream #57).

Operations per second from pnpm bench, higher is better, in Node 24.22.0, on Intel(R) Xeon(R) Processor @ 2.10GHz (4 cores). A dash means the library has no equivalent.

Sizes cover the imports in the usage example, or the equivalent imports of each library, bundled for the browser by scripts/bundle-size.ts. Runtime dependencies count every package installed with the published package. A library that uses Node built-ins needs polyfills in a browser and the nodejs_compat flag in Cloudflare Workers.

• maxBytes always counts bytes. get-stream's maxBuffer counts characters for getStream and elements for getStreamAsArray, and toArray has no limit.
• raw-body's limit accepts a string such as '1mb'. maxBytes takes a number of bytes, such as 1_000_000.
• get-stream's MaxBufferError carries the data read so far in bufferedData. A collector that fails has already cancelled its source and dropped what it read, so its errors carry no data.
• Byte collectors accept only Uint8Array (including Buffer) and string chunks. get-stream also accepted ArrayBuffer, DataView and other typed arrays, and concat-stream accepted numbers.
• Arrays and other iterables are valid sources. get-stream rejected them.
• merge2 called with several arguments, as in merge2(a, b, c), reads the sources one after another, which is concat. merge reads them at the same time.
• merge2's .add() and the { end: false } option have no equivalent, so pass every source when calling concat or merge.
• map passes on undefined like any other value. through2 dropped it. Use filter to drop chunks.
• A transform generator starts when the output is first read. A through2 transform starts when the first chunk is written.