interface Stream
extends AsyncIterable<Result<T, E>>

A TypeScript library that provides a powerful and flexible way to handle errors in asynchronous streams. It allows you to collect and manage errors alongside successful values in a stream, enabling you to process data while gracefully handling any issues that may arise.

The core concept is the Stream, which is an asynchronous iterable that emits results as either successful values or errors. You can use various methods on the Stream to transform, filter, and reduce both successful values and errors in a way that suits your application's needs.

Type Parameters

Methods

map<
U,
E2 = E,
>(fn: (value: T) => Promisable<U>): Stream<U, E | E2>

Similar to Array.prototype.map, but works on the stream of results. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.

When concurrency is greater than 1, results may be emitted out of order.

tryMap<
U,
F = never,
>(
fn: (value: T) => Promisable<U>,
errFn: (
error: unknown,
value: T,
) => Promisable<F>,
): Stream<U, E | F>

Maps successful values with fn and transforms errors with errFn. Both receive the original value so you can contextualize the mapping.

flatMap<U>(fn: (value: T) => Promisable<AsyncIterable<U> | Iterable<U>>): Stream<U, E>

Similar to Array.prototype.flatMap, but works on the stream of results. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.

When concurrency is greater than 1, results may be emitted out of order.

filter<U extends T>(fn: (value: T) => value is U): Stream<U, E>

Similar to Array.prototype.filter, but works on the stream of results. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.

filter(fn: (value: T) => Promisable<boolean>): Stream<T, E>
tap(fn: (value: T) => Promisable<void>): Stream<T, E>

Runs a side-effect function on each successful value without transforming it. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in place of the original value.

tapErr(fn: (error: E) => Promisable<void>): Stream<T, E>

Runs a side-effect function on each error without transforming it. If the provided function throws an error or returns a rejected promise, the new error replaces the original.

take(n: number): Stream<T, E>

Limits the stream to at most n successful values. Errors pass through without counting against the limit. After n successes are yielded, the stream stops immediately (any pending errors from earlier in the pipeline may still be yielded before stopping).

takeWhile(fn: (value: T) => Promisable<boolean>): Stream<T, E>

Yields successful values while the predicate returns true. Once the predicate returns false, iteration stops. Errors pass through until the stream is stopped. If the predicate throws, an error result is emitted and iteration stops.

fold<U>(
fn: (
acc: U,
value: T,
) => Promisable<U>,
initialValue: U,
): Promise<U>

Similar to Array.prototype.reduce, but works on the stream of results. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.

If any error results are present in the stream, they will be thrown as an AggregateError after the stream is exhausted. Use filterErr(() => false) upstream to explicitly drop errors if you want to fold only the successes.

scan<U>(
fn: (
acc: U,
value: T,
) => Promisable<U>,
initialValue: U,
): Stream<U, E>

Like fold but emits the running accumulator after each successful value, allowing downstream operations to react to intermediate states.

mapErr<F>(fn: (error: E) => Promisable<F>): Stream<T, F>

Similar to Array.prototype.map, but works on the stream of errors. If the provided function throws an error or returns a rejected promise, the new error will be collected and emitted as an error result in the stream.

filterErr<F extends E>(fn: (error: E) => error is F): Stream<T, F>

Similar to Array.prototype.filter, but works on the stream of errors. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.

filterErr(fn: (error: E) => Promisable<boolean>): Stream<T, E>
foldErr<F>(
fn: (
acc: F,
error: E,
) => Promisable<F>,
initialValue: F,
): Promise<F>

Similar to Array.prototype.reduce, but works on the stream of errors. If the provided function throws an error or returns a rejected promise, the new error will be collected and emitted as an error result in the stream.

recoverWhen<
E2 extends E,
U,
>(
guard: (error: E) => error is E2,
fn: (error: E2) => Promisable<U>,
): Stream<T | U, Exclude<E, E2>>

Recovers from specific error types by applying the provided function to transform them into successful values. This allows you to handle specific errors gracefully while still collecting other errors in the stream.

recover<U>(fn: (error: E) => Promisable<U>): Stream<T | U, never>

Recovers from all errors by applying the provided function to transform them into successful values. This allows you to handle all errors gracefully while still collecting successful values in the stream.

throwOn<E2 extends E>(guard: (error: E) => error is E2): Stream<T, Exclude<E, E2>>

Throws the specified error types if they are encountered in the stream. This allows you to handle specific errors immediately while continuing to process other errors.

successes(): AsyncIterable<T>

Returns an async iterable of all successful values emitted by the stream. If any errors were collected during the stream processing, they will be ignored in this iterable.

errors(): AsyncIterable<E>

Returns an async iterable of all errors collected during the stream processing. If any successful values were emitted during the stream processing, they will be ignored in this iterable.

collect(): Promise<T[]>

Collects all successful values emitted by the stream into an array. If any errors were collected during the stream processing, they will be thrown as an AggregateError containing all collected errors.

partition(): Promise<{ successes: T[]; errors: E[]; }>

Collects all results into separate successes and errors arrays. Unlike collect(), this never throws.

toArray(): Promise<Result<T, E>[]>

Collects all results emitted by the stream into an array of Result objects, which can represent either successful values or errors. This method allows you to see the full outcome of the stream processing, including both successes and errors, without throwing an aggregate error.

chunks(size: number): Stream<T[], E>

Collects consecutive successful values into fixed-size arrays. Errors pass through without breaking the current chunk.

zip<
U,
F,
>(other: Stream<U, F>): Stream<[T, U], E | F>

Combines this stream with another into tuples, yielding one result per pair of values. The shorter stream determines when zipping completes.

merge(other: Stream<T, E>): Stream<T, E>

Merges two streams by interleaving their results. Both streams must have compatible error types. The merged stream yields values from either stream as they become available.

[[Symbol.asyncIterator]](): AsyncIterator<Result<T, E>>

Usage

import { type Stream } from ".";