<A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<A, E>Interleave every node's stream into one (a transform for combineStream).
export const const mergeNodeStreams: <A, E>(
streams: ReadonlyArray<NodeStream<A, E>>
) => Stream.Stream<A, E>
Interleave every node's stream into one (a transform for
combineStream
).
mergeNodeStreams = <function (type parameter) A in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<A, E>A, function (type parameter) E in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<A, E>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<A, E>A, function (type parameter) E in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<A, E>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<function (type parameter) A in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<A, E>A, function (type parameter) E in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<A, E>E> =>
streams: ReadonlyArray<NodeStream<A, E>>streams.ReadonlyArray<NodeStream<A, E>>.reduce<Stream.Stream<A, E, never>>(callbackfn: (previousValue: Stream.Stream<A, E, never>, currentValue: NodeStream<A, E>, currentIndex: number, array: readonly NodeStream<A, E>[]) => Stream.Stream<A, E, never>, initialValue: Stream.Stream<A, E, never>): Stream.Stream<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<function (type parameter) A in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<A, E>A, function (type parameter) E in <A, E>(streams: ReadonlyArray<NodeStream<A, E>>): Stream.Stream<A, E>E>>((acc: Stream.Stream<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: <A, E, never, A, E, never>(self: Stream.Stream<A, E, never>, that: Stream.Stream<A, E, never>, options?: {
readonly haltStrategy?: Stream.HaltStrategy | undefined;
} | undefined) => Stream.Stream<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<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>;
}
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; <…;
}
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);