interface AsyncStreamConstructors

An interface describing all factory functions used to create AsyncStream instances. Implementations of this interface are exposed via the global AsyncStream value and the @rimbu/stream/async-custom sub-package.

Methods

always

undocumented

Definition

always<T>(value: AsyncOptLazy<T>): AsyncStream.NonEmpty<T>;

Type parameters

Name	Description
T

Parameters

Name	Type	Description
value	AsyncOptLazy<T>

empty

undocumented

Definition

empty<T>(): AsyncStream<T>;

Type parameters

Name	Description
T

flatten

Returns an AsyncStream concatenating the given source AsyncStreamSource containing StreamSources.

Definitions

flatten<T extends AsyncStreamSource.NonEmpty<S>, S>(source: AsyncStreamSource.NonEmpty<T>): AsyncStream.NonEmpty<S>;

flatten<T extends AsyncStreamSource<S>, S>(source: AsyncStreamSource<T>): AsyncStream<S>;

Type parameters

Name	Constraints	Description
T	AsyncStreamSource.NonEmpty<S>
S

Parameters

Name	Type	Description
source	AsyncStreamSource.NonEmpty<T>	a StreamSource containing nested StreamSources

example

await AsyncStream.flatten(AsyncStream.of([[1, 2], [3], [], [4]])).toArray()  // => [1, 2, 3, 4]
await AsyncStream.flatten(AsyncStream.of(['ma', 'r', '', 'mot')).toArray()   // => ['m', 'a', 'r', 'm', 'o', 't']

from

undocumented

Definitions

from<T>(...sources: ArrayNonEmpty<AsyncStreamSource.NonEmpty<T>>): AsyncStream.NonEmpty<T>;

from<T>(...sources: ArrayNonEmpty<AsyncStreamSource<T>>): AsyncStream<T>;

Type parameters

Name	Description
T

Parameters

Name	Type	Description
sources	ArrayNonEmpty<AsyncStreamSource.NonEmpty<T>>

fromResource

undocumented

Definitions

fromResource<T, R>(options: {
    open: () => MaybePromise<R>;
    createSource: (resource: R) => AsyncStreamSource.NonEmpty<T>;
    close?: (resource: R) => MaybePromise<void>;
  }): AsyncStream.NonEmpty<T>;

fromResource<T, R>(options: {
    open: () => MaybePromise<R>;
    createSource: (resource: R) => MaybePromise<AsyncStreamSource<T>>;
    close?: (resource: R) => MaybePromise<void>;
  }): AsyncStream<T>;

Type parameters

Name	Description
T
R

Parameters

Name	Type	Description
options	{     open: () => MaybePromise<R>;     createSource: (resource: R) => AsyncStreamSource.NonEmpty<T>;     close?: (resource: R) => MaybePromise<void>;   }

of

undocumented

Definition

of<T>(...values: ArrayNonEmpty<AsyncOptLazy<T>>): AsyncStream.NonEmpty<T>;

Type parameters

Name	Description
T

Parameters

Name	Type	Description
values	ArrayNonEmpty<AsyncOptLazy<T>>

unfold

Returns a possibly infinite Stream starting with given init value, followed by applying given next function to the previous value.

Definition

unfold<T>(init: T, next: (current: T, index: number, stop: Token) => MaybePromise<T | Token>): AsyncStream.NonEmpty<T>;

Type parameters

Name	Description
T

Parameters

Name	Type	Description
init	T	an initial value
next	(current: T, index: number, stop: Token) => MaybePromise<T | Token>	a function taking the last value, its index, and a stop token, and returning a new value or a stop token

example

Stream.unfold(2, v => v * v).take(4).toArray()   // => [2, 4, 16, 256]

unzip

Returns an array containing an AsyncStream for each tuple element resulting from given source AsyncStream.

Definitions

unzip<T extends readonly unknown[] & {
    length: L;
  }, L extends number>(source: AsyncStream.NonEmpty<T>, options: {
    length: L;
  }): {
    [K in keyof T]: AsyncStream.NonEmpty<T[K]>;
  };

unzip<T extends readonly unknown[] & {
    length: L;
  }, L extends number>(source: AsyncStream<T>, options: {
    length: L;
  }): {
    [K in keyof T]: AsyncStream<T[K]>;
  };

Type parameters

Name	Constraints	Description
T	readonly unknown[] & {     length: L;   }
L	number

Parameters

Name	Type	Description
source	AsyncStream.NonEmpty<T>	a Stream containing tuple elements
options	{     length: L;   }

example

const [a, b] = AsyncStream.unzip(AsyncStream.of([[1, 'a'], [2, 'b']]), 2)
await a.toArray()   // => [1, 2]
await b.toArray()   // => ['a', 'b']

zip

Returns an AsyncStream with tuples containing each successive value from the given sources.

Definitions

zip<I extends readonly [unknown, ...unknown[]]>(...sources: {
    [K in keyof I]: AsyncStreamSource.NonEmpty<I[K]>;
  } & unknown[]): AsyncStream.NonEmpty<I>;

zip<I extends readonly [unknown, ...unknown[]]>(...sources: {
    [K in keyof I]: AsyncStreamSource<I[K]>;
  } & unknown[]): AsyncStream<I>;

Type parameters

Name	Constraints	Description
I	readonly [unknown, ...unknown[]]

Parameters

Name	Type	Description
sources	{     [K in keyof I]: AsyncStreamSource.NonEmpty<I[K]>;   } & unknown[]	the input async stream sources

example

await AsyncStream.zip(
[1, 2, 3],
[4, 5],
['a', 'b', 'c']
).toArray()
// => [[1, 4, 'a'], [2, 5, 'b']]

note

ends the AsyncStream when any of the given streams ends

zipAll

Returns an AsyncStream with tuples containing each successive value from the given sources, adding given fillValue to any streams that end before all streams have ended.

Definitions

zipAll<I extends readonly [unknown, ...unknown[]], O>(fillValue: AsyncOptLazy<O>, ...sources: {
    [K in keyof I]: AsyncStreamSource.NonEmpty<I[K]>;
  } & unknown[]): AsyncStream.NonEmpty<{
    [K in keyof I]: I[K] | O;
  }>;

zipAll<I extends readonly [unknown, ...unknown[]], O>(fillValue: AsyncOptLazy<O>, ...sources: {
    [K in keyof I]: AsyncStreamSource<I[K]>;
  } & unknown[]): AsyncStream<{
    [K in keyof I]: I[K] | O;
  }>;

Type parameters

Name	Constraints	Description
I	readonly [unknown, ...unknown[]]
O

Parameters

Name	Type	Description
fillValue	AsyncOptLazy<O>	the AsyncOptLazy value to add to streams that end early
sources	{     [K in keyof I]: AsyncStreamSource.NonEmpty<I[K]>;   } & unknown[]	the input async stream sources

example

await AsyncStream.zipAll(
0,
[1, 2, 3],
[4, 5],
['a', 'b', 'c']
).toArray()
// => [[1, 4, 'a'], [2, 5, 'b'], [3, 0, 'c']]

note

ends the AsyncStream when any of the given streams ends

zipAllWith

Returns an AsyncStream with the result of applying given zipFun to each successive value resulting from the given sources, adding given fillValue to any Streams that end before all streams have ended.

Definitions

zipAllWith<I extends readonly [unknown, ...unknown[]]>(...sources: {
    [K in keyof I]: AsyncStreamSource.NonEmpty<I[K]>;
  } & unknown[]): <O, R>(fillValue: AsyncOptLazy<O>, zipFun: (...values: {
    [K in keyof I]: I[K] | O;
  }) => MaybePromise<R>) => AsyncStream.NonEmpty<R>;

zipAllWith<I extends readonly [unknown, ...unknown[]]>(...sources: {
    [K in keyof I]: AsyncStreamSource<I[K]>;
  } & unknown[]): <O, R>(fillValue: AsyncOptLazy<O>, zipFun: (...values: {
    [K in keyof I]: I[K] | O;
  }) => MaybePromise<R>) => AsyncStream<R>;

Type parameters

Name	Constraints	Description
I	readonly [unknown, ...unknown[]]

Parameters

Name	Type	Description
sources	{     [K in keyof I]: AsyncStreamSource.NonEmpty<I[K]>;   } & unknown[]	the input async stream sources

example

await AsyncStream.zipAllWith(
[1, 2],
[3, 4, 5],
[6, 7]
)(
async () => 0,
async (a, b, c) => a + b + c,
).toArray()
// => [10, 13, 5]

zipWith

Returns an AsyncStream with the result of applying given zipFun to each successive value resulting from the given sources.

Definitions

zipWith<I extends readonly [unknown, ...unknown[]]>(...sources: {
    [K in keyof I]: AsyncStreamSource.NonEmpty<I[K]>;
  } & unknown[]): <R>(zipFun: (...values: I) => R) => AsyncStream.NonEmpty<R>;

zipWith<I extends readonly [unknown, ...unknown[]]>(...sources: {
    [K in keyof I]: AsyncStreamSource<I[K]>;
  } & unknown[]): <R>(zipFun: (...values: I) => R) => AsyncStream<R>;

Type parameters

Name	Constraints	Description
I	readonly [unknown, ...unknown[]]

Parameters

Name	Type	Description
sources	{     [K in keyof I]: AsyncStreamSource.NonEmpty<I[K]>;   } & unknown[]	the input async stream sources

example

await AsyncStream.zipWith(
[1, 2],
[3, 4, 5],
[true, false]
)(
async (a, b, c) => c ? a + b : a - b,
).toArray()
// => [4, -2]

note

ends the AsyncStream when any of the given streams ends