Interface Stream<T>

A lazy stream with fluent operators, composition, and terminal methods.

interface Stream<T> {
 "[asyncIterator]"(): StreamIterator<T>;
 bufferTime(windowMs: number, options: BufferTimeOptions): Stream<T[]>;
 catchError(
 handler: (
 error: unknown,
 context: ResourceContext,
 ) => Source | PromiseLike<Source>,
 ): Stream<T | U>;
 chunks(size: number): Stream<T[]>;
 concat(this: Stream<Source>, options?: HigherOrderOptions): Stream;
 concatMap(
 mapper: (value: T, context: OperatorContext) => Source,
 options?: HigherOrderOptions,
 ): Stream;
 deadline(durationMs: number): Stream<T>;
 dedupe<K = T>(
 keySelector?: (value: T, context: OperatorContext) => K,
 ): Stream<T>;
 delayStart(ms: number): Stream<T>;
 every(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 options?: ExecutionOptions,
 ): Promise<boolean>;
 filter<S>(
 predicate: (value: T, context: OperatorContext) => value is S,
 ): Stream<S>;
 filter(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 ): Stream<T>;
 filterMap(
 mapper: (value: T, context: OperatorContext) => U,
 ): Stream<NonNullable<Awaited>>;
 finalize(effect: () => unknown): Stream<T>;
 find<S>(
 predicate: (value: T, context: OperatorContext) => value is S,
 options?: ExecutionOptions,
 ): Promise<Awaited<S> | undefined>;
 find(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 options?: ExecutionOptions,
 ): Promise<Awaited<T> | undefined>;
 flatMap(
 mapper: (value: T, context: OperatorContext) => Source,
 options?: HigherOrderOptions,
 ): Stream;
 flatten(
 this: Stream<Source>,
 options?: HigherOrderOptions,
 ): Stream;
 forEach(
 effect: (value: T, context: OperatorContext) => unknown,
 options?: ExecutionOptions,
 ): Promise<void>;
 iterate(options?: ExecutionOptions): StreamIterator<T>;
 map(
 mapper: (value: T, context: MapContext) => U,
 options?: MapOptions,
 ): Stream<Awaited>;
 mapSettled(
 mapper: (value: T, context: MapContext) => U,
 options?: MapOptions,
 ): Stream<SettledResult<Awaited>>;
 pipe(): Stream<T>;
 pipe<A>(op1: CoreOperator<T, A>): Stream<A>;
 pipe<A, B>(op1: CoreOperator<T, A>, op2: CoreOperator<A, B>): Stream;
 pipe<A, B, C>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 ): Stream<C>;
 pipe<A, B, C, D>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 ): Stream<D>;
 pipe<A, B, C, D, E>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 op5: CoreOperator<D, E>,
 ): Stream<E>;
 pipe<A, B, C, D, E, F>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 op5: CoreOperator<D, E>,
 op6: CoreOperator<E, F>,
 ): Stream<F>;
 pipe<A, B, C, D, E, F, G>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 op5: CoreOperator<D, E>,
 op6: CoreOperator<E, F>,
 op7: CoreOperator<F, G>,
 ): Stream<G>;
 pipe<A, B, C, D, E, F, G, H>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 op5: CoreOperator<D, E>,
 op6: CoreOperator<E, F>,
 op7: CoreOperator<F, G>,
 op8: CoreOperator<G, H>,
 ): Stream<H>;
 rateLimit(options: RateLimitOptions): Stream<T>;
 reduce<A>(
 reducer: (
 accumulator: Awaited<A>,
 value: T,
 context: OperatorContext,
 ) => Awaited<A> | PromiseLike<Awaited<A>>,
 initial: A,
 options?: ExecutionOptions,
 ): Promise<Awaited<A>>;
 scan<A>(
 reducer: (
 accumulator: Awaited<A>,
 value: T,
 context: OperatorContext,
 ) => Awaited<A> | PromiseLike<Awaited<A>>,
 initial: A,
 ): Stream<Awaited<A>>;
 skip(count: number): Stream<T>;
 skipWhile(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 ): Stream<T>;
 some(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 options?: ExecutionOptions,
 ): Promise<boolean>;
 spaceout(intervalMs: number): Stream<T>;
 take(count: number): Stream<T>;
 takeWhile<S>(
 predicate: (value: T, context: OperatorContext) => value is S,
 ): Stream<S>;
 takeWhile(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 ): Stream<T>;
 tap(effect: (value: T, context: OperatorContext) => unknown): Stream<T>;
 timeoutEach(ms: number): Stream<T>;
 toArray(options?: ExecutionOptions): Promise<T[]>;
}

Type Parameters

Hierarchy (View Summary)

CoreStream<T>
- Stream

Methods

[asyncIterator]

"[asyncIterator]"(): StreamIterator<T>

Creates a new execution that starts reading on the first next() call.

Returns StreamIterator<T>

Inherited from CoreStream.[asyncIterator]

bufferTime

bufferTime(windowMs: number, options: BufferTimeOptions): Stream<T[]>
Collects batches bounded by time and maxItems.
Parameters
- windowMs: number
- options: BufferTimeOptions
Returns Stream<T[]>

catchError

catchError(
 handler: (
 error: unknown,
 context: ResourceContext,
 ) => Source | PromiseLike<Source>,
): Stream<T | U>
Recovers once from an upstream failure using a fallback source.
Type Parameters
- U
Parameters
- handler: (error: unknown, context: ResourceContext) => Source | PromiseLike<Source>
Returns Stream<T | U>

chunks

chunks(size: number): Stream<T[]>
Groups values into arrays of at most size items.
Parameters
- size: number
Returns Stream<T[]>

concat

concat(this: Stream<Source>, options?: HigherOrderOptions): Stream
Flattens child sources in child order. Only available on streams of sources.
Type Parameters
- U
Parameters
- this: Stream<Source>
- Optionaloptions: HigherOrderOptions
Returns Stream

concatMap

concatMap(
mapper: (value: T, context: OperatorContext) => Source,
options?: HigherOrderOptions,
): Stream
Maps values to child sources and emits in child order.
Type Parameters
- U
Parameters
- mapper: (value: T, context: OperatorContext) => Source
- Optionaloptions: HigherOrderOptions
Returns Stream

deadline

deadline(durationMs: number): Stream<T>
Limits the duration of this stage.
Parameters
- durationMs: number
Returns Stream<T>

dedupe

dedupe<K = T>(
keySelector?: (value: T, context: OperatorContext) => K,
): Stream<T>
Drops consecutive equal values or keys.
Type Parameters
- K = T
Parameters
- OptionalkeySelector: (value: T, context: OperatorContext) => K
Returns Stream<T>

delayStart

delayStart(ms: number): Stream<T>
Delays opening upstream.
Parameters
- ms: number
Returns Stream<T>

every

every(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 options?: ExecutionOptions,
): Promise<boolean>
Returns false on the first non-matching value and closes upstream early, or true if all values match.
Parameters
- predicate: (value: T, context: OperatorContext) => boolean | PromiseLike<boolean>
- options: ExecutionOptions = {}
Returns Promise<boolean>

Inherited from CoreStream.every

filter

filter<S>(
predicate: (value: T, context: OperatorContext) => value is S,
): Stream<S>
Keeps matching values; synchronous type guards narrow the output type.
Type Parameters
- S
Parameters
- predicate: (value: T, context: OperatorContext) => value is S
Returns Stream<S>
filter(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
): Stream<T>
Keeps matching values; synchronous type guards narrow the output type.
Parameters
- predicate: (value: T, context: OperatorContext) => boolean | PromiseLike<boolean>
Returns Stream<T>

filterMap

filterMap(
mapper: (value: T, context: OperatorContext) => U,
): Stream<NonNullable<Awaited>>
Maps values sequentially, omitting null and undefined results.
Type Parameters
- U
Parameters
- mapper: (value: T, context: OperatorContext) => U
Returns Stream<NonNullable<Awaited>>

finalize

finalize(effect: () => unknown): Stream<T>
Runs an effect during cleanup and waits for it to finish.
Parameters
- effect: () => unknown
Returns Stream<T>

find

find<S>(
predicate: (value: T, context: OperatorContext) => value is S,
options?: ExecutionOptions,
): Promise<Awaited<S> | undefined>
Returns the first value matching predicate (or undefined if none match) and closes upstream.
Type Parameters
- S
Parameters
- predicate: (value: T, context: OperatorContext) => value is S
- Optionaloptions: ExecutionOptions
Returns Promise<Awaited<S> | undefined>

Inherited from CoreStream.find
find(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 options?: ExecutionOptions,
): Promise<Awaited<T> | undefined>
Returns the first value matching predicate (or undefined if none match) and closes upstream.
Parameters
- predicate: (value: T, context: OperatorContext) => boolean | PromiseLike<boolean>
- Optionaloptions: ExecutionOptions
Returns Promise<Awaited<T> | undefined>

Inherited from CoreStream.find

flatMap

flatMap(
mapper: (value: T, context: OperatorContext) => Source,
options?: HigherOrderOptions,
): Stream
Maps values to child sources and flattens them concurrently.
Type Parameters
- U
Parameters
- mapper: (value: T, context: OperatorContext) => Source
- Optionaloptions: HigherOrderOptions
Returns Stream

flatten

flatten(this: Stream<Source>, options?: HigherOrderOptions): Stream
Flattens child sources concurrently. Only available on streams of sources.
Type Parameters
- U
Parameters
- this: Stream<Source>
- Optionaloptions: HigherOrderOptions
Returns Stream

forEach

forEach(
effect: (value: T, context: OperatorContext) => unknown,
options?: ExecutionOptions,
): Promise<void>
Runs effect sequentially for each value and waits for cleanup without collecting results.
Parameters
- effect: (value: T, context: OperatorContext) => unknown
- options: ExecutionOptions = {}
Returns Promise<void>

Inherited from CoreStream.forEach

iterate

iterate(options?: ExecutionOptions): StreamIterator<T>
Returns a controlled iterator for this stream. Equivalent to iterate(this, options).
Parameters
- options: ExecutionOptions = {}
Returns StreamIterator<T>

Inherited from CoreStream.iterate

map

map(
mapper: (value: T, context: MapContext) => U,
options?: MapOptions,
): Stream<Awaited>
Maps values with the same concurrency, ordering, and retry options as map.
Type Parameters
- U
Parameters
- mapper: (value: T, context: MapContext) => U
- Optionaloptions: MapOptions
Returns Stream<Awaited>

mapSettled

mapSettled(
mapper: (value: T, context: MapContext) => U,
options?: MapOptions,
): Stream<SettledResult<Awaited>>
Maps values to fulfilled or rejected outcomes.
Type Parameters
- U
Parameters
- mapper: (value: T, context: MapContext) => U
- Optionaloptions: MapOptions
Returns Stream<SettledResult<Awaited>>

pipe

pipe(): Stream<T>

Composes up to eight operators from left to right without starting execution.

Returns Stream<T>

Overrides CoreStream.pipe
pipe<A>(op1: CoreOperator<T, A>): Stream<A>
Composes up to eight operators from left to right without starting execution.
Type Parameters
- A
Parameters
- op1: CoreOperator<T, A>
Returns Stream<A>

Overrides CoreStream.pipe
pipe<A, B>(op1: CoreOperator<T, A>, op2: CoreOperator<A, B>): Stream
Composes up to eight operators from left to right without starting execution.
Type Parameters
- A
- B
Parameters
- op1: CoreOperator<T, A>
- op2: CoreOperator<A, B>
Returns Stream

Overrides CoreStream.pipe
pipe<A, B, C>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
): Stream<C>
Composes up to eight operators from left to right without starting execution.
Type Parameters
- A
- B
- C
Parameters
- op1: CoreOperator<T, A>
- op2: CoreOperator<A, B>
- op3: CoreOperator<B, C>
Returns Stream<C>

Overrides CoreStream.pipe
pipe<A, B, C, D>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
): Stream<D>
Composes up to eight operators from left to right without starting execution.
Type Parameters
- A
- B
- C
- D
Parameters
- op1: CoreOperator<T, A>
- op2: CoreOperator<A, B>
- op3: CoreOperator<B, C>
- op4: CoreOperator<C, D>
Returns Stream<D>

Overrides CoreStream.pipe
pipe<A, B, C, D, E>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 op5: CoreOperator<D, E>,
): Stream<E>
Composes up to eight operators from left to right without starting execution.
Type Parameters
- A
- B
- C
- D
- E
Parameters
- op1: CoreOperator<T, A>
- op2: CoreOperator<A, B>
- op3: CoreOperator<B, C>
- op4: CoreOperator<C, D>
- op5: CoreOperator<D, E>
Returns Stream<E>

Overrides CoreStream.pipe
pipe<A, B, C, D, E, F>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 op5: CoreOperator<D, E>,
 op6: CoreOperator<E, F>,
): Stream<F>
Composes up to eight operators from left to right without starting execution.
Type Parameters
- A
- B
- C
- D
- E
- F
Parameters
- op1: CoreOperator<T, A>
- op2: CoreOperator<A, B>
- op3: CoreOperator<B, C>
- op4: CoreOperator<C, D>
- op5: CoreOperator<D, E>
- op6: CoreOperator<E, F>
Returns Stream<F>

Overrides CoreStream.pipe
pipe<A, B, C, D, E, F, G>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 op5: CoreOperator<D, E>,
 op6: CoreOperator<E, F>,
 op7: CoreOperator<F, G>,
): Stream<G>
Composes up to eight operators from left to right without starting execution.
Type Parameters
- A
- B
- C
- D
- E
- F
- G
Parameters
- op1: CoreOperator<T, A>
- op2: CoreOperator<A, B>
- op3: CoreOperator<B, C>
- op4: CoreOperator<C, D>
- op5: CoreOperator<D, E>
- op6: CoreOperator<E, F>
- op7: CoreOperator<F, G>
Returns Stream<G>

Overrides CoreStream.pipe
pipe<A, B, C, D, E, F, G, H>(
 op1: CoreOperator<T, A>,
 op2: CoreOperator<A, B>,
 op3: CoreOperator<B, C>,
 op4: CoreOperator<C, D>,
 op5: CoreOperator<D, E>,
 op6: CoreOperator<E, F>,
 op7: CoreOperator<F, G>,
 op8: CoreOperator<G, H>,
): Stream<H>
Composes up to eight operators from left to right without starting execution.
Type Parameters
- A
- B
- C
- D
- E
- F
- G
- H
Parameters
- op1: CoreOperator<T, A>
- op2: CoreOperator<A, B>
- op3: CoreOperator<B, C>
- op4: CoreOperator<C, D>
- op5: CoreOperator<D, E>
- op6: CoreOperator<E, F>
- op7: CoreOperator<F, G>
- op8: CoreOperator<G, H>
Returns Stream<H>

Overrides CoreStream.pipe

rateLimit

rateLimit(options: RateLimitOptions): Stream<T>
Limits the rate of upstream reads.
Parameters
- options: RateLimitOptions
Returns Stream<T>

reduce

reduce<A>(
 reducer: (
 accumulator: Awaited<A>,
 value: T,
 context: OperatorContext,
 ) => Awaited<A> | PromiseLike<Awaited<A>>,
 initial: A,
 options?: ExecutionOptions,
): Promise<Awaited<A>>
Reduces values sequentially from initial and returns the final accumulator after cleanup.
Type Parameters
- A
Parameters
- reducer: (
 accumulator: Awaited<A>,
 value: T,
 context: OperatorContext,
 ) => Awaited<A> | PromiseLike<Awaited<A>>
- initial: A
- options: ExecutionOptions = {}
Returns Promise<Awaited<A>>

Inherited from CoreStream.reduce

scan

scan<A>(
 reducer: (
 accumulator: Awaited<A>,
 value: T,
 context: OperatorContext,
 ) => Awaited<A> | PromiseLike<Awaited<A>>,
 initial: A,
): Stream<Awaited<A>>
Emits each updated accumulator after reducing a value.
Type Parameters
- A
Parameters
- reducer: (
 accumulator: Awaited<A>,
 value: T,
 context: OperatorContext,
 ) => Awaited<A> | PromiseLike<Awaited<A>>
- initial: A
Returns Stream<Awaited<A>>

skip

skip(count: number): Stream<T>
Skips the first count values.
Parameters
- count: number
Returns Stream<T>

skipWhile

skipWhile(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
): Stream<T>
Skips a matching prefix.
Parameters
- predicate: (value: T, context: OperatorContext) => boolean | PromiseLike<boolean>
Returns Stream<T>

some

some(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
 options?: ExecutionOptions,
): Promise<boolean>
Returns true on the first matching value and closes upstream early, or false if no values match.
Parameters
- predicate: (value: T, context: OperatorContext) => boolean | PromiseLike<boolean>
- options: ExecutionOptions = {}
Returns Promise<boolean>

Inherited from CoreStream.some

spaceout

spaceout(intervalMs: number): Stream<T>
Spaces delivery of values.
Parameters
- intervalMs: number
Returns Stream<T>

take

take(count: number): Stream<T>
Emits at most count values, then closes upstream.
Parameters
- count: number
Returns Stream<T>

takeWhile

takeWhile<S>(
predicate: (value: T, context: OperatorContext) => value is S,
): Stream<S>
Emits a matching prefix; synchronous type guards narrow the output type.
Type Parameters
- S
Parameters
- predicate: (value: T, context: OperatorContext) => value is S
Returns Stream<S>
takeWhile(
 predicate: (
 value: T,
 context: OperatorContext,
 ) => boolean | PromiseLike<boolean>,
): Stream<T>
Emits a matching prefix; synchronous type guards narrow the output type.
Parameters
- predicate: (value: T, context: OperatorContext) => boolean | PromiseLike<boolean>
Returns Stream<T>

tap

tap(effect: (value: T, context: OperatorContext) => unknown): Stream<T>
Runs an effect for each value without changing the value.
Parameters
- effect: (value: T, context: OperatorContext) => unknown
Returns Stream<T>

timeoutEach

timeoutEach(ms: number): Stream<T>
Limits how long each upstream read may take.
Parameters
- ms: number
Returns Stream<T>

toArray

toArray(options?: ExecutionOptions): Promise<T[]>
Collects all emitted values into an array and waits for cleanup before resolving.
Parameters
- options: ExecutionOptions = {}
Returns Promise<T[]>

Inherited from CoreStream.toArray

Interface Stream<T>

Type Parameters

Hierarchy (View Summary)

Index

Methods

Methods

[asyncIterator]

Returns StreamIterator<T>

bufferTime

Parameters

Returns Stream<T[]>

catchError

Type Parameters

Parameters

Returns Stream<T | U>

chunks

Parameters

Returns Stream<T[]>

concat

Type Parameters

Parameters

Returns Stream<U>

concatMap

Type Parameters

Parameters

Returns Stream<U>

deadline

Parameters

Returns Stream<T>

dedupe

Type Parameters

Parameters

Returns Stream<T>

delayStart

Parameters

Returns Stream<T>

every

Parameters

Returns Promise<boolean>

filter

Type Parameters

Parameters

Returns Stream<S>

Parameters

Returns Stream<T>

filterMap

Type Parameters

Parameters

Returns Stream<NonNullable<Awaited<U>>>

finalize

Parameters

Returns Stream<T>

find

Type Parameters

Parameters

Returns Promise<Awaited<S> | undefined>

Parameters

Returns Promise<Awaited<T> | undefined>

flatMap

Type Parameters

Parameters

Returns Stream<U>

flatten

Type Parameters

Parameters

Returns Stream<U>

forEach

Parameters

Returns Promise<void>

iterate

Parameters

Returns StreamIterator<T>

map

Type Parameters

Parameters

Returns Stream<Awaited<U>>

mapSettled

Type Parameters

Parameters

Returns Stream<SettledResult<Awaited<U>>>