Class Stream<T, E>

Type Parameters

T
Type of the 'values' on the stream.
E
Type of the 'errors' on the stream.

Hierarchy

HigherOrderStream<T, E>
- Stream

Index

Constructors

constructor

new Stream<T, E>(stream): Stream<T, E>
Type Parameters
- T
- E
Parameters
- stream: Readable
Returns Stream<T, E>
Inherited from HigherOrderStream<T, E>.constructor
- Defined in stream/base.ts:21

Properties

`Protected` stackTrace

stackTrace: string[] = []

`Protected` stream

stream: Readable

`Protected` traceComplete

traceComplete: boolean = false

`Static` StreamEnd

StreamEnd: symbol = ...

Marker for the end of a stream.

Methods

batch

batch(options): Stream<T[], E>
Batch items together on the stream and emit them as an array. The batch can be over some size, or can be over a period of time.
Parameters
- options: {
 byBucket?: ((value) => string | number);
 n?: number;
 timeout?: number;
 yieldEmpty?: boolean;
 yieldRemaining?: boolean;
 }
 - Optional byBucket?: ((value) => string | number)
 - (value): string | number
 Parameters
 value: T
 Returns string | number
 - Optional n?: number
 Batch up to n items (activates batching by count)
 - Optional timeout?: number
 Batch for timeout milliseconds (activates batching by timeout)
 - Optional yieldEmpty?: boolean
 If timeout is reached and no items have been emitted on the stream, still emit an empty array.
 - Optional yieldRemaining?: boolean
Returns Stream<T[], E>
Inherited from HigherOrderStream.batch
- Defined in stream/transforms.ts:427

exhaust

exhaust(): Promise<void>
Completely exhaust the stream, driving it to completion. This is particularly useful when side effects of the stream are desired, but the actual values of the stream are not needed.

Returns Promise<void>
Inherited from HigherOrderStream.exhaust
- Defined in stream/consumption.ts:20

`Protected` getTrace

getTrace(): string[]
Capture the current trace. Creates a clone of the trace to prevent it being modified.

Returns string[]
Inherited from HigherOrderStream.getTrace
- Defined in stream/base.ts:40

onFirst

onFirst(callback, options?): Stream<T, E>
Run the provided callback after the first atom passes through the stream.
Parameters
- callback: (() => void)
 Callback to run.
 - - (): void
 - Returns void
- Optional options: {
 atom?: boolean;
 } = {}
 - Optional atom?: boolean
 Run on any atom (true by default). Will only run after first ok atom if false.
Returns Stream<T, E>
Inherited from HigherOrderStream.onFirst
- Defined in stream/transforms.ts:579

onLast

onLast(callback, options?): Stream<T, E>
Run the provided callback after the last atom passes through the stream.
Parameters
- callback: (() => void)
 Callback to run.
 - - (): void
 - Returns void
- Optional options: {
 atom?: boolean;
 } = {}
 - Optional atom?: boolean
 Run on any atom (true by default). Will only run after last ok atom if false.
Returns Stream<T, E>
Inherited from HigherOrderStream.onLast
- Defined in stream/transforms.ts:604

`Protected` trace

trace(trace): string[]
Add a layer to the trace object. Returns a copy of the current trace.
Parameters
- trace: string
Returns string[]
Inherited from HigherOrderStream.trace
- Defined in stream/base.ts:28

`Static` writable

writable<T, E>(): {
stream: Stream<T, E>;
writable: Writable;
}
Create a stream and corresponding writable Node stream, where any writes to the writable Node stream will be emitted on the returned stream.
Type Parameters
- T
- E
Returns {
stream: Stream<T, E>;
writable: Writable;
}
- stream: Stream<T, E>
- writable: Writable
Inherited from HigherOrderStream.writable
- Defined in stream/base.ts:342

Atom

`Static` error

error<T, E>(value): Atom<T, E>
Create an error atom with the provided value.
Type Parameters
- T
- E
Parameters
- value: E
Returns Atom<T, E>
- Defined in stream/index.ts:37

`Static` exception

exception<T, E>(value, trace): Atom<T, E>
Create an exception atom with the provided value.
Type Parameters
- T
- E
Parameters
- value: unknown
- trace: string[]
Returns Atom<T, E>
- Defined in stream/index.ts:46

`Static` isError

isError<T, E>(atom): atom is AtomError<E>
Verify if the provided atom is of the error variant.
Type Parameters
- T
- E
Parameters
- atom: Atom<T, E>
Returns atom is AtomError<E>
- Defined in stream/index.ts:72

`Static` isException

isException<T, E>(atom): atom is AtomException
Verify if the provided atom is of the exception variant.
Type Parameters
- T
- E
Parameters
- atom: Atom<T, E>
Returns atom is AtomException
- Defined in stream/index.ts:81

`Static` isOk

isOk<T, E>(atom): atom is AtomOk<T>
Verify if the provided atom is of the ok variant.
Type Parameters
- T
- E
Parameters
- atom: Atom<T, E>
Returns atom is AtomOk<T>
- Defined in stream/index.ts:63

`Static` isUnknown

isUnknown<T, E>(atom): atom is AtomException
Type Parameters
- T
- E
Parameters
- atom: Atom<T, E>
Returns atom is AtomException
Deprecated
use isException instead
- Defined in stream/index.ts:89

`Static` ok

ok<T, E>(value): Atom<T, E>
Create an ok atom with the provided value.
Type Parameters
- T
- E
Parameters
- value: T
Returns Atom<T, E>
- Defined in stream/index.ts:28

`Static` unknown

unknown<T, E>(value, trace): Atom<T, E>
Type Parameters
- T
- E
Parameters
- value: unknown
- trace: string[]
Returns Atom<T, E>
Deprecated
use exception instead
- Defined in stream/index.ts:54

Consumption

[asyncIterator]

[asyncIterator](): AsyncIterator<Atom<T, E>, any, undefined>
Create an iterator that will emit each atom in the stream.

Returns AsyncIterator<Atom<T, E>, any, undefined>
Inherited from HigherOrderStream.[asyncIterator]
- Defined in stream/consumption.ts:12

serialise

serialise(options?): Readable
Serialise the stream, and produce a Node stream with the serialised result.
Parameters
- Optional options: {
  atoms?: boolean;
  single?: boolean;
  }
  - Optional atoms?: boolean
    By default, only ok values are serialised, however enabling this will serialise all values.
  - Optional single?: boolean
    Whether to emit an array for multiple items, or only a single item.
Returns Readable
See
Stream#toReadable if serialisation is not required
Inherited from HigherOrderStream.serialise
- Defined in stream/consumption.ts:92

toArray

toArray(options?): Promise<T[]>
Iterate through each atom in the stream, and return them as a single array.
Parameters
- Optional options: {
 atoms?: false;
 reject?: boolean;
 }
 - Optional atoms?: false
 Return every atom on the stream.
 - Optional reject?: boolean
 If an error or exception is encountered, reject the promise with it.
Returns Promise<T[]>
Inherited from HigherOrderStream.toArray
- Defined in stream/consumption.ts:64
toArray(options?): Promise<Atom<T, E>[]>
Parameters
- Optional options: {
 atoms: true;
 }
 - atoms: true
Returns Promise<Atom<T, E>[]>
Inherited from HigherOrderStream.toArray
- Defined in stream/consumption.ts:65

toReadable

toReadable(kind, options?): Readable
Produce a readable node stream with the values from the stream.
Parameters
- kind: "object" | "raw"
  What kind of readable stream to produce. When "raw" only strings and buffers can be emitted on the stream. Use "object" to preserve objects in the readable stream. Note that object values are not serialised, they are emitted as objects.
- Optional options: {
  atoms?: boolean;
  }
  Options for configuring how atoms are output on the stream
  - Optional atoms?: boolean
Returns Readable
See
Stream#serialize if the stream values should be serialized to json
Inherited from HigherOrderStream.toReadable
- Defined in stream/consumption.ts:151
toReadable(kind): Readable
Produce a readable node stream with the raw values from the stream
Parameters
- kind: "raw"
Returns Readable
Note
the stream must only contain atoms of type string or Buffer. If not, a stream error will be emitted.

See
Stream#serialize if the stream values should be serialized to json
Inherited from HigherOrderStream.toReadable
- Defined in stream/consumption.ts:163
toReadable(kind, options?): Readable
Produce a readable node stream in object mode with the values from the stream
Parameters
- kind: "object"
- Optional options: {
  atoms?: boolean;
  }
  - Optional atoms?: boolean
    By default, only ok values are emitted, however enabling this will emit all values.
Returns Readable
Note
When not using options.atoms, any null atom values will be skipped when piping to the readable stream

See
Stream#serialize if the stream values should be serialized to json
Inherited from HigherOrderStream.toReadable
- Defined in stream/consumption.ts:175

values

values(): AsyncIterableIterator<T>
Create an async iterator that will emit each value in the stream.

Returns AsyncIterableIterator<T>
Inherited from HigherOrderStream.values
- Defined in stream/consumption.ts:29

Creation

`Static` from

from<T, E>(value): Stream<T, E>
Create a stream from some kind of stream-like value. This can be an iterable, a promise that resolves to some value, or even another readable stream.
Type Parameters
- T
- E
Parameters
- value: Promise<MaybeAtom<T, E>> | Iterator<MaybeAtom<T, E>, any, undefined> | AsyncIterator<MaybeAtom<T, E>, any, undefined> | Iterable<MaybeAtom<T, E>> | AsyncIterable<MaybeAtom<T, E>> | MaybeAtom<T, E>[] | (() => Promise<MaybeAtom<T, E>>)
Returns Stream<T, E>
Inherited from HigherOrderStream.from
- Defined in stream/base.ts:50

`Static` fromArray

fromArray<T, E>(array): Stream<T, E>
Create a stream from an array.
Type Parameters
- T
- E
Parameters
- array: MaybeAtom<T, E>[]
 The array that values will be emitted from.
Returns Stream<T, E>
Note
The array will be shallow cloned internally, so that the original array won't be impacted.
Inherited from HigherOrderStream.fromArray
- Defined in stream/base.ts:182

`Static` fromCallback

fromCallback<T, E>(cb): Stream<T, E>
Create a stream from a node-style callback. A node-compatible callback function will be passed as the first parameter to the callback of this function.

The first parameter provided to the callback (the error) will be emitted as an Error atom, whilst the second parameter (the value) will be emitted as an Ok atom.
Type Parameters
- T
- E
Parameters
- cb: ((next) => void)
 - - (next): void
 - Parameters
 next: ((error, value) => unknown)
 (error, value): unknown
 Parameters
 error: E
 value: T
 Returns unknown
 Returns void
Returns Stream<T, E>
Example
```
$.fromCallback((next) => someAsyncMethod(paramA, paramB, next));
```
Inherited from HigherOrderStream.fromCallback
- Defined in stream/base.ts:100

`Static` fromIterable

fromIterable<T, E>(iterable): Stream<T, E>
Create a stream from an iterable.
Type Parameters
- T
- E
Parameters
- iterable: Iterable<MaybeAtom<T, E>> | AsyncIterable<MaybeAtom<T, E>>
 The iterable that will produce an iterator, which may be an async iterator.
Returns Stream<T, E>
Inherited from HigherOrderStream.fromIterable
- Defined in stream/base.ts:162

`Static` fromIterator

fromIterator<T, E>(iterator): Stream<T, E>
Create a stream from an iterator.
Type Parameters
- T
- E
Parameters
- iterator: Iterator<MaybeAtom<T, E>, any, undefined> | AsyncIterator<MaybeAtom<T, E>, any, undefined>
 The iterator that will produce values, which may be an async iterator.
Returns Stream<T, E>
Inherited from HigherOrderStream.fromIterator
- Defined in stream/base.ts:139

`Static` fromNext

fromNext<T, E>(next): Stream<T, E>
Create a new stream with the provided atom producer.
Type Parameters
- T
- E
Parameters
- next: (() => Promise<symbol | MaybeAtom<T, E>>)
 A callback method to produce the next atom. If no atom is available, then StreamEnd must be returned.
 - - (): Promise<symbol | MaybeAtom<T, E>>
 - Returns Promise<symbol | MaybeAtom<T, E>>
Returns Stream<T, E>
Inherited from HigherOrderStream.fromNext
- Defined in stream/base.ts:200

`Static` fromPromise

fromPromise<T, E>(promise): Stream<T, E>
Create a stream from a promise. The promise will be awaited, and the resulting value only ever emitted once.
Type Parameters
- T
- E
Parameters
- promise: Promise<MaybeAtom<T, E>>
 The promise to create the stream from.
Returns Stream<T, E>
Inherited from HigherOrderStream.fromPromise
- Defined in stream/base.ts:118

`Static` fromPusher

fromPusher<T, E>(): {
 done: (() => void);
 push: ((value) => void);
 stream: Stream<T, E>;
}
Create a new stream, and use the provided push and done methods to add values to it, and complete the stream.
- push: Adds the provided value to the stream.
- done: Indicatest that the stream is done, meaning that any future calls to push or done will be ignored.
Type Parameters
- T
- E
Returns {
 done: (() => void);
 push: ((value) => void);
 stream: Stream<T, E>;
}
- done: (() => void)
 - - (): void
 - Returns void
- push: ((value) => void)
 - - (value): void
 - Parameters
 value: MaybeAtom<T, E>
 Returns void
- stream: Stream<T, E>
Inherited from HigherOrderStream.fromPusher
- Defined in stream/base.ts:235

`Static` of

of<T, E>(value): Stream<T, E>
Create a new stream containing a single value. Unless an atom is provided, it will be converted to an ok atom.
Type Parameters
- T
- E
Parameters
- value: MaybeAtom<T, E>
Returns Stream<T, E>
Inherited from HigherOrderStream.of
- Defined in stream/base.ts:300

`Static` ofError

ofError<T, E>(value): Stream<T, E>
Create a new stream containing a single error atom.
Type Parameters
- T
- E
Parameters
- value: E
Returns Stream<T, E>
Inherited from HigherOrderStream.ofError
- Defined in stream/base.ts:317

`Static` ofException

ofException<T, E>(value): Stream<T, E>
Create a new stream containing a single exception atom.
Type Parameters
- T
- E
Parameters
- value: unknown
Returns Stream<T, E>
Inherited from HigherOrderStream.ofException
- Defined in stream/base.ts:326

`Static` ofUnknown

ofUnknown<T, E>(value): Stream<T, E>
Type Parameters
- T
- E
Parameters
- value: unknown
Returns Stream<T, E>
Deprecated
use ofException instead
Inherited from HigherOrderStream.ofUnknown
- Defined in stream/base.ts:334

Higher Order

cachedFlatMap

cachedFlatMap(cb, keyFn): Stream<U, E>
Map over each value in the stream, produce a stream from it, cache the resultant stream and flatten all the value streams together
Type Parameters
- U
Parameters
- cb: ((value) => MaybePromise<Stream<U, E>>)
 - - (value): MaybePromise<Stream<U, E>>
 - Parameters
 value: T
 Returns MaybePromise<Stream<U, E>>
- keyFn: ((value) => string | number | symbol)
 - - (value): string | number | symbol
 - Parameters
 value: T
 Returns string | number | symbol
Returns Stream<U, E>
Inherited from HigherOrderStream.cachedFlatMap
- Defined in stream/higher-order.ts:144

flatMap

flatMap(cb): Stream<U, E>
Map over each value in the stream, produce a stream from it, and flatten all the value streams together
Type Parameters
- U
Parameters
- cb: ((value) => MaybePromise<Stream<U, E>>)
 - - (value): MaybePromise<Stream<U, E>>
 - Parameters
 value: T
 Returns MaybePromise<Stream<U, E>>
Returns Stream<U, E>
Inherited from HigherOrderStream.flatMap
- Defined in stream/higher-order.ts:81

flatMapError

flatMapError<F>(cb): Stream<T, F>
Map over each error in the stream, produce a stream from it, and flatten all the value streams together.
Type Parameters
- F
Parameters
- cb: ((value) => MaybePromise<Stream<T, F>>)
 - - (value): MaybePromise<Stream<T, F>>
 - Parameters
 value: E
 Returns MaybePromise<Stream<T, F>>
Returns Stream<T, F>
Inherited from HigherOrderStream.flatMapError
- Defined in stream/higher-order.ts:98

flatMapException

flatMapException(cb): Stream<T, E>
Maps over each exception in the stream, producing a new stream from it, and flatten all the value streams together.
Parameters
- cb: ((value, trace) => MaybePromise<Stream<T, E>>)
 - - (value, trace): MaybePromise<Stream<T, E>>
 - Parameters
 value: unknown
 trace: string[]
 Returns MaybePromise<Stream<T, E>>
Returns Stream<T, E>
Inherited from HigherOrderStream.flatMapException
- Defined in stream/higher-order.ts:115

flatMapUnknown

flatMapUnknown(cb): Stream<T, E>
Parameters
- cb: ((value, trace) => MaybePromise<Stream<T, E>>)
 - - (value, trace): MaybePromise<Stream<T, E>>
 - Parameters
 value: unknown
 trace: string[]
 Returns MaybePromise<Stream<T, E>>
Returns Stream<T, E>
Deprecated
use flatMapException instead
Inherited from HigherOrderStream.flatMapUnknown
- Defined in stream/higher-order.ts:132

flatTap

flatTap(cb): Stream<T, E>
Produce a stream for each value in the stream. The resulting stream will be emptied, however the resulting values will just be dropped. This is analogous to an asynchronous side effect.
Parameters
- cb: ((value) => MaybePromise<Stream<unknown, unknown>>)
 - - (value): MaybePromise<Stream<unknown, unknown>>
 - Parameters
 value: T
 Returns MaybePromise<Stream<unknown, unknown>>
Returns Stream<T, E>
Inherited from HigherOrderStream.flatTap
- Defined in stream/higher-order.ts:236

flatten

flatten(): T extends Stream<U, E>
? Stream<U, E>
: Stream<T, E>
Produce a new stream from the stream that has any nested streams flattened

Returns T extends Stream<U, E>
? Stream<U, E>
: Stream<T, E>
Note
Any atoms that are not nested streams are emitted as-is
Inherited from HigherOrderStream.flatten
- Defined in stream/higher-order.ts:192

otherwise

otherwise(cbOrStream): Stream<T, E>
Emit items from provided stream if this stream is completely empty.
Parameters
- cbOrStream: CallbackOrStream<T, E>
Returns Stream<T, E>
Note
If there are any errors or exceptions on the stream, then the new stream won't be consumed.
Inherited from HigherOrderStream.otherwise
- Defined in stream/higher-order.ts:253

replaceWith

replaceWith<U, F>(cbOrStream): Stream<U, F>
Consume the entire stream, and completely replace it with a new stream. This will remove any errors and exceptions currently on the stream.

Equivalent to:
```
stream
 .filter(() => false)
 .otherwise(newStream);
```
`
Type Parameters
- U
- F
Parameters
- cbOrStream: CallbackOrStream<U, F>
Returns Stream<U, F>
Inherited from HigherOrderStream.replaceWith
- Defined in stream/higher-order.ts:288

Transform

bufferedMap

bufferedMap(cb, options?): Stream<U, E>
Map over each value in the stream, and apply some callback to it. This differs from map as the upstream is continually pulled, regardless if the downstream is ready for it and whether the current iteration has complete.

Example
Approximate timelines are shown below. It isn't 100% correct for what's actually happening, but gives the right idea.
- p: producer (upstream)
- m: map operation (asynchronous)
Map
```
 0ms 10ms 20ms 30ms
 |---------|---------|---------|---------|
 |<-- p -->|<-- m -->| | |
 | | |<-- p -->|<-- m -->|
```
Buffered Map
```
 0ms 10ms 20ms 30ms
 |---------|---------|---------|---------|
 |<-- p -->|<-- m -->| | |
 | |<-- p -->|<-- m -->| |
 | | |<-- p -->|<-- m -->|
```
Type Parameters
- U
Parameters
- cb: ((value) => MaybePromise<MaybeAtom<U, E>>)
 - - (value): MaybePromise<MaybeAtom<U, E>>
 - Parameters
 value: T
 Returns MaybePromise<MaybeAtom<U, E>>
- Optional options: {
 delay?: number;
 maxBufferPeriod?: number;
 maxBufferSize?: number;
 }
 - Optional delay?: number
 - Optional maxBufferPeriod?: number
 - Optional maxBufferSize?: number
Returns Stream<U, E>
Inherited from HigherOrderStream.bufferedMap
- Defined in stream/transforms.ts:341

collect

collect(): Stream<T[], E>
Collect the values of the stream atoms into an array then return a stream which emits that array

Returns Stream<T[], E>
Note
non-ok atoms are emitted as-is, the collected array is always emitted last

Note
empty streams will emit an empty array
Inherited from HigherOrderStream.collect
- Defined in stream/transforms.ts:33

compact

compact(): Stream<Truthy<T>, E>
Remove falsey values from the stream.

This is equivalent to doing .filter((value) => value).

Returns Stream<Truthy<T>, E>
Inherited from HigherOrderStream.compact
- Defined in stream/transforms.ts:189

consume

consume<U, F>(generator): Stream<U, F>
Consume the stream atoms, emitting new atoms from the generator.
Type Parameters
- U
- F
Parameters
- generator: ((it) => AsyncGenerator<Atom<U, F>, any, unknown>)
 - - (it): AsyncGenerator<Atom<U, F>, any, unknown>
 - Parameters
 it: AsyncIterable<Atom<T, E>>
 Returns AsyncGenerator<Atom<U, F>, any, unknown>
Returns Stream<U, F>
Inherited from HigherOrderStream.consume
- Defined in stream/transforms.ts:15

delay

delay(ms): Stream<T, E>
Delay emitting each value on the stream by ms.
Parameters
- ms: number
Returns Stream<T, E>
Inherited from HigherOrderStream.delay
- Defined in stream/transforms.ts:295

drop

drop(n, options?): Stream<T, E>
Drop the first n items from the stream.
Parameters
- n: number
- Optional options: {
 atoms?: boolean;
 }
 - Optional atoms?: boolean
Returns Stream<T, E>
Inherited from HigherOrderStream.drop
- Defined in stream/transforms.ts:268

filter

filter(condition): Stream<T, E>
Filter over each value in the stream.
Parameters
- condition: ((value) => unknown)
 - - (value): unknown
 - Parameters
 value: T
 Returns unknown
Returns Stream<T, E>
Inherited from HigherOrderStream.filter
- Defined in stream/transforms.ts:155

inspect

inspect(): Stream<T, E>
Inspect every atom that is emitted through the stream.

Returns Stream<T, E>
Inherited from HigherOrderStream.inspect
- Defined in stream/transforms.ts:138

map

map(cb): Stream<U, E>
Map over each value in the stream.
Type Parameters
- U
Parameters
- cb: ((value) => MaybePromise<MaybeAtom<U, E>>)
 - - (value): MaybePromise<MaybeAtom<U, E>>
 - Parameters
 value: T
 Returns MaybePromise<MaybeAtom<U, E>>
Returns Stream<U, E>
Inherited from HigherOrderStream.map
- Defined in stream/transforms.ts:54

mapError

mapError<F>(cb): Stream<T, F>
Map over each error in the stream.
Type Parameters
- F
Parameters
- cb: ((error) => MaybePromise<MaybeAtom<T, F>>)
 - - (error): MaybePromise<MaybeAtom<T, F>>
 - Parameters
 error: E
 Returns MaybePromise<MaybeAtom<T, F>>
Returns Stream<T, F>
Inherited from HigherOrderStream.mapError
- Defined in stream/transforms.ts:73

mapException

mapException(cb): Stream<T, E>
Map over each exception in the stream.
Parameters
- cb: ((error) => MaybePromise<MaybeAtom<T, E>>)
 - - (error): MaybePromise<MaybeAtom<T, E>>
 - Parameters
 error: unknown
 Returns MaybePromise<MaybeAtom<T, E>>
Returns Stream<T, E>
Inherited from HigherOrderStream.mapException
- Defined in stream/transforms.ts:92

mapUnknown

mapUnknown(cb): Stream<T, E>
Parameters
- cb: ((error) => MaybePromise<MaybeAtom<T, E>>)
 - - (error): MaybePromise<MaybeAtom<T, E>>
 - Parameters
 error: unknown
 Returns MaybePromise<MaybeAtom<T, E>>
Returns Stream<T, E>
Deprecated
use mapException instead
Inherited from HigherOrderStream.mapUnknown
- Defined in stream/transforms.ts:110

take

take(n, options?): Stream<T, E>
Return a stream containing the first n values. If options.atoms is true, then the first n atoms rather than values will be emitted.
Parameters
- n: number
- Optional options: {
 atoms?: boolean;
 }
 - Optional atoms?: boolean
 If enabled, first n atoms will be counted, otherwise values.
Returns Stream<T, E>
Inherited from HigherOrderStream.take
- Defined in stream/transforms.ts:239

tap

tap(cb): Stream<T, E>
Run a callback for each value in the stream, ideal for side effects on stream items.
Parameters
- cb: ((value) => unknown)
 - - (value): unknown
 - Parameters
 value: T
 Returns unknown
Returns Stream<T, E>
Inherited from HigherOrderStream.tap
- Defined in stream/transforms.ts:119

Transforms

reduce

reduce(cb, memo): Stream<U, E>
Operate on each item in the stream, reducing it into a single value. The resulting value is returned in its own stream.
Type Parameters
- U
Parameters
- cb: ((memo, value) => MaybePromise<MaybeAtom<U, E>>)
 - - (memo, value): MaybePromise<MaybeAtom<U, E>>
 - Parameters
 memo: U
 value: T
 Returns MaybePromise<MaybeAtom<U, E>>
- memo: U
Returns Stream<U, E>
Inherited from HigherOrderStream.reduce
- Defined in stream/transforms.ts:207

Class Stream<T, E>

Type Parameters

T

E

Hierarchy

Index

Constructors

Properties

Methods

Atom

Consumption

Creation

Higher Order

Transform

Transforms

Constructors

constructor

Type Parameters

T

E

Parameters

stream: Readable

Returns Stream<T, E>

Properties

Protected stackTrace

Protected stream

Protected traceComplete

Static StreamEnd

Methods

batch

Parameters

options: { byBucket?: ((value) => string | number); n?: number; timeout?: number; yieldEmpty?: boolean; yieldRemaining?: boolean; }

Optional byBucket?: ((value) => string | number)

Parameters

value: T

Returns string | number

Optional n?: number

Optional timeout?: number

Optional yieldEmpty?: boolean

Optional yieldRemaining?: boolean

Returns Stream<T[], E>

exhaust

Returns Promise<void>

Protected getTrace

Returns string[]

onFirst

Parameters

callback: (() => void)

Returns void

Optional options: { atom?: boolean; } = {}

Optional atom?: boolean

Returns Stream<T, E>

onLast

Parameters

callback: (() => void)

Returns void

Optional options: { atom?: boolean; } = {}

Optional atom?: boolean

Returns Stream<T, E>

Protected trace

Parameters

trace: string

Returns string[]

Static writable

Type Parameters

T

E

Returns { stream: Stream<T, E>; writable: Writable; }

stream: Stream<T, E>

writable: Writable

Atom

Static error

Type Parameters

T

E

Parameters

value: E

Returns Atom<T, E>

Static exception

Type Parameters

`Protected` stackTrace

`Protected` stream

`Protected` traceComplete

`Static` StreamEnd

options: {
byBucket?: ((value) => string | number);
n?: number;
timeout?: number;
yieldEmpty?: boolean;
yieldRemaining?: boolean;
}

`Optional` byBucket?: ((value) => string | number)

`Optional` n?: number

`Optional` timeout?: number

`Optional` yieldEmpty?: boolean

`Optional` yieldRemaining?: boolean

`Protected` getTrace

`Optional` options: {
atom?: boolean;
} = {}

`Optional` atom?: boolean

`Optional` options: {
atom?: boolean;
} = {}

`Optional` atom?: boolean

`Protected` trace

`Static` writable

Returns {
stream: Stream<T, E>;
writable: Writable;
}

`Static` error

`Static` exception

`Static` isError

`Static` isException

`Static` isOk

`Static` isUnknown

`Static` ok

`Static` unknown

`Optional` options: {
atoms?: boolean;
single?: boolean;
}

`Optional` atoms?: boolean

`Optional` single?: boolean

`Optional` options: {
atoms?: false;
reject?: boolean;
}

`Optional` atoms?: false

`Optional` reject?: boolean

`Optional` options: {
atoms: true;
}