<Svc, A, E, B, EE>(
peers: Record<string, Svc>,
pick: (service: Svc) => Stream.Stream<A, E>,
transform: (
streams: ReadonlyArray<NodeStream<A, E>>
) => Stream.Stream<B, EE>
): Stream.Stream<B, EE>Combine a stream field across every instance in peers: transform receives each node's stream
(node-attributed) and produces the combined stream — e.g. mergeNodeStreams to interleave
them all, or a latest-per-node fold. Pure; the live subscription lifecycle is the caller's.
export const const combineStream: <Svc, A, E, B, EE>(
peers: Record<string, Svc>,
pick: (service: Svc) => Stream.Stream<A, E>,
transform: (
streams: ReadonlyArray<NodeStream<A, E>>
) => Stream.Stream<B, EE>
) => Stream.Stream<B, EE>
Combine a stream field across every instance in peers: transform receives each node's stream
(node-attributed) and produces the combined stream — e.g.
mergeNodeStreams
to interleave
them all, or a latest-per-node fold. Pure; the live subscription lifecycle is the caller's.
combineStream = <function (type parameter) Svc in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>Svc, function (type parameter) A in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>A, function (type parameter) E in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>E, function (type parameter) B in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>B, function (type parameter) EE in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>EE>(
peers: Record<string, Svc>peers: type Record<K extends keyof any, T> = {
[P in K]: T
}
Construct a type with a set of properties K of type T
Record<string, function (type parameter) Svc in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>Svc>,
pick: (service: Svc) => Stream.Stream<A, E>pick: (service: Svcservice: function (type parameter) Svc in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>Svc) => 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 <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>A, function (type parameter) E in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>E>,
transform: (
streams: ReadonlyArray<NodeStream<A, E>>
) => Stream.Stream<B, EE>
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 <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>A, function (type parameter) E in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>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) B in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>B, function (type parameter) EE in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>EE>,
): 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) B in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>B, function (type parameter) EE in <Svc, A, E, B, EE>(peers: Record<string, Svc>, pick: (service: Svc) => Stream.Stream<A, E>, transform: (streams: ReadonlyArray<NodeStream<A, E>>) => Stream.Stream<B, EE>): Stream.Stream<B, EE>EE> =>
transform: (
streams: ReadonlyArray<NodeStream<A, E>>
) => Stream.Stream<B, EE>
var Object: ObjectConstructorProvides functionality common to all JavaScript objects.
Object.ObjectConstructor.entries<Svc>(o: {
[s: string]: Svc;
} | ArrayLike<Svc>): [string, Svc][] (+1 overload)
Returns an array of key/values of the enumerable own properties of an object
entries(peers: Record<string, Svc>peers).Array<[string, Svc]>.map<{
node: string;
stream: Stream.Stream<A, E, never>;
}>(callbackfn: (value: [string, Svc], index: number, array: [string, Svc][]) => {
node: string;
stream: Stream.Stream<A, E, never>;
}, thisArg?: any): {
node: string;
stream: Stream.Stream<A, E, never>;
}[]
Calls a defined callback function on each element of an array, and returns an array that contains the results.
map(([node: stringnode, service: Svcservice]) => ({ node: stringnode, stream: Stream.Stream<A, E, never>(property) 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; <…;
}
pick: (service: Svc) => Stream.Stream<A, E>pick(service: Svcservice) })),
);