namespace AsyncStream

A possibly infinite asynchronous sequence of elements of type T. See the Stream documentation and the AsyncStream API documentation

Companion interface: AsyncStream<T>

Interfaces

Name	Description
AsyncStream.NonEmpty<T>	A non-empty and possibly infinite asynchronous sequence of elements of type T. See the Stream documentation and the AsyncStream API documentation

Static 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>(open: () => MaybePromise<R>, createSource: (resource: R) => AsyncStreamSource.NonEmpty<T>, close: (resource: R) => MaybePromise<void>): AsyncStream.NonEmpty<T>;

fromResource<T, R>(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
open	() => MaybePromise<R>
createSource	(resource: R) => AsyncStreamSource.NonEmpty<T>
close	(resource: R) => MaybePromise<void>

live

Returns a tuple of a controller and a stream, where the controller can be used to submit values and close the stream, and the stream will emit the submitted values, with a buffer of maximum length thte given maxSize. Each stream client will start buffering from the moment that it is created.

Definition

live<T>(maxSize?: number): [{
    submit(value: T): void;
    close: () => void;
  }, AsyncStream<T>];

Type parameters

Name	Description
T	the element type

Parameters

Name	Type	Description
maxSize	number	the maximum buffer size

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>, length: L): {
    [K in keyof T]: AsyncStream.NonEmpty<T[K]>;
  };

unzip<T extends readonly unknown[] & {
    length: L;
  }, L extends number>(source: AsyncStream<T>, 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
length	L	the tuple length

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