<A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<
{ readonly node: string; readonly value: A },
E
>Interleave every node's stream into one, tagging each element with its node — attribution
when following peers (a transform for combineStream).
combinatorscombineStream
Source src/MultiNode.ts:1657 lines
export const const mergeNodeStreamsByNode: <A, E>(
streams: ReadonlyArray<NodeStream<A, E>>
) => Stream.Stream<
{ readonly node: string; readonly value: A },
E
>
Interleave every node's stream into one, tagging each element with its node — attribution
when following peers (a transform for
combineStream
).
mergeNodeStreamsByNode = <function (type parameter) A in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E>
function (type parameter) E in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E>
streams: ReadonlyArray<NodeStream<A, E>>streams: interface ReadonlyArray<T>ReadonlyArray<interface NodeStream<A, E = never>One node's stream for a gathered stream field.
NodeStream<function (type parameter) A in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E>
function (type parameter) E in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E>
import StreamStream.interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<{ readonly node: stringnode: string; readonly value: Avalue: function (type parameter) A in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E>
function (type parameter) E in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E>
streams: ReadonlyArray<NodeStream<A, E>>streams.ReadonlyArray<NodeStream<A, E>>.reduce<Stream.Stream<{
readonly node: string;
readonly value: A;
}, E, never>>(callbackfn: (previousValue: Stream.Stream<{
readonly node: string;
readonly value: A;
}, E, never>, currentValue: NodeStream<A, E>, currentIndex: number, array: readonly NodeStream<A, E>[]) => Stream.Stream<{
readonly node: string;
readonly value: A;
}, E, never>, initialValue: Stream.Stream<{
readonly node: string;
readonly value: A;
}, E, never>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E, never> (+2 overloads)
Calls the specified callback function for all the elements in an array. The return value of the callback function is the accumulated result, and is provided as an argument in the next call to the callback function.
reduce<import StreamStream.interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<{ readonly node: stringnode: string; readonly value: Avalue: function (type parameter) A in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E>
function (type parameter) E in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<{
readonly node: string;
readonly value: A;
}, E>
acc: Stream.Stream<
{
readonly node: string
readonly value: A
},
E,
never
>
(parameter) acc: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
s: NodeStream<A, E>(parameter) s: {
node: string;
stream: Stream.Stream<A, E>;
}
import StreamStream.const merge: <{
readonly node: string;
readonly value: A;
}, E, never, {
node: string;
value: A;
}, E, never>(self: Stream.Stream<{
readonly node: string;
readonly value: A;
}, E, never>, that: Stream.Stream<{
node: string;
value: A;
}, E, never>, options?: {
readonly haltStrategy?: Stream.HaltStrategy | undefined;
} | undefined) => Stream.Stream<{
readonly node: string;
readonly value: A;
} | {
node: string;
value: A;
}, E, never> (+1 overload)
Merges two streams, emitting elements from both as they arrive.
Details
By default, the merged stream ends when both streams end. Use
haltStrategy to change the termination behavior.
Example (Merging stream values)
import { Console, Effect, Stream } from "effect"
const fast = Stream.make(1, 2, 3)
const slow = Stream.fromEffect(Effect.delay(Effect.succeed(4), "50 millis"))
const program = Effect.gen(function*() {
const result = yield* Stream.runCollect(Stream.merge(fast, slow))
yield* Console.log(result)
})
Effect.runPromise(program)
// Output: [ 1, 2, 3, 4 ]
merge(acc: Stream.Stream<
{
readonly node: string
readonly value: A
},
E,
never
>
(parameter) acc: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
import StreamStream.const map: <A, E, never, {
node: string;
value: A;
}>(self: Stream.Stream<A, E, never>, f: (a: A, i: number) => {
node: string;
value: A;
}) => Stream.Stream<{
node: string;
value: A;
}, E, never> (+1 overload)
Transforms the elements of this stream using the supplied function.
Example (Mapping stream values)
import { Console, Effect, Stream } from "effect"
const stream = Stream.fromArray([1, 2, 3]).pipe(Stream.map((n, i) => n + i))
const program = Stream.runCollect(stream).pipe(
Effect.tap((values) => Console.log(values))
)
Effect.runPromise(program)
// [ 1, 3, 5 ]
map(s: NodeStream<A, E>(parameter) s: {
node: string;
stream: Stream.Stream<A, E>;
}
NodeStream<A, E>.stream: Stream.Stream<A, E, never>(property) NodeStream<A, E>.stream: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
value: Avalue) => ({ node: stringnode: s: NodeStream<A, E>(parameter) s: {
node: string;
stream: Stream.Stream<A, E>;
}
NodeStream<A, E>.node: stringnode, value: Avalue }))),
import StreamStream.const empty: Stream.Stream<
never,
never,
never
>
const empty: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
Creates an empty stream.
Example (Creating an empty stream)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const values = yield* Stream.empty.pipe(Stream.runCollect)
yield* Console.log(values)
})
Effect.runPromise(program)
// []
empty,
);