<T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<
[T[number]] extends [never]
? never
: {
[K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R>
? A
: never
},
[T[number]] extends [never]
? never
: T[number] extends Stream<infer _A, infer _E, infer _R>
? _E
: never,
[T[number]] extends [never]
? never
: T[number] extends Stream<infer _A, infer _E, infer _R>
? _R
: never
>Zips multiple streams so that when a value is emitted by any stream, it is combined with the latest values from the other streams to produce a result.
When to use
Use when each stream should contribute its latest value after all streams have emitted at least once.
Gotchas
Note: tracking the latest value is done on a per-array basis. That means that emitted elements that are not the last value in arrays will never be used for zipping.
Example (Zipping latest values from many streams)
import { Console, Effect, Stream } from "effect"
const stream = Stream.zipLatestAll(
Stream.make(1, 2, 3).pipe(Stream.rechunk(1)),
Stream.make("a", "b", "c").pipe(Stream.rechunk(1)),
Stream.make(true, false, true).pipe(Stream.rechunk(1))
)
const program = Effect.gen(function*() {
const result = yield* Stream.runCollect(stream)
yield* Console.log(result)
})
Effect.runPromise(program)
// Output: [ [ 1, "a", true ], [ 2, "a", true ], [ 3, "a", true ], [ 3, "b", true ], [ 3, "c", true ], [ 3, "c", false ], [ 3, "c", true ] ]export const const zipLatestAll: <
T extends ReadonlyArray<Stream<any, any, any>>
>(
...streams: T
) => Stream<
[T[number]] extends [never]
? never
: {
[K in keyof T]: T[K] extends Stream<
infer A,
infer _E,
infer _R
>
? A
: never
},
[T[number]] extends [never]
? never
: T[number] extends Stream<
infer _A,
infer _E,
infer _R
>
? _E
: never,
[T[number]] extends [never]
? never
: T[number] extends Stream<
infer _A,
infer _E,
infer _R
>
? _R
: never
>
Zips multiple streams so that when a value is emitted by any stream, it is
combined with the latest values from the other streams to produce a result.
When to use
Use when each stream should contribute its latest value after all streams have
emitted at least once.
Gotchas
Note: tracking the latest value is done on a per-array basis. That means
that emitted elements that are not the last value in arrays will never be
used for zipping.
Example (Zipping latest values from many streams)
import { Console, Effect, Stream } from "effect"
const stream = Stream.zipLatestAll(
Stream.make(1, 2, 3).pipe(Stream.rechunk(1)),
Stream.make("a", "b", "c").pipe(Stream.rechunk(1)),
Stream.make(true, false, true).pipe(Stream.rechunk(1))
)
const program = Effect.gen(function*() {
const result = yield* Stream.runCollect(stream)
yield* Console.log(result)
})
Effect.runPromise(program)
// Output: [ [ 1, "a", true ], [ 2, "a", true ], [ 3, "a", true ], [ 3, "b", true ], [ 3, "c", true ], [ 3, "c", false ], [ 3, "c", true ] ]
zipLatestAll = <function (type parameter) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T extends interface ReadonlyArray<T>ReadonlyArray<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<any, any, any>>>(
...streams: T extends ReadonlyArray<Stream<any, any, any>>streams: function (type parameter) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T
): 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) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T[number]] extends [never] ? never
: { [function (type parameter) KK in keyof function (type parameter) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T]: function (type parameter) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T[function (type parameter) KK] extends 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<infer function (type parameter) AA, infer function (type parameter) _E_E, infer function (type parameter) _R_R> ? function (type parameter) AA : never },
[function (type parameter) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T[number]] extends [never] ? never : function (type parameter) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T[number] extends 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<infer function (type parameter) _A_A, infer function (type parameter) _E_E, infer function (type parameter) _R_R> ? function (type parameter) _E_E : never,
[function (type parameter) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T[number]] extends [never] ? never : function (type parameter) T in <T extends ReadonlyArray<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in keyof T]: T[K] extends Stream<infer A, infer _E, infer _R> ? A : never; }, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<infer _A, infer _E, infer _R> ? _R : never>T[number] extends 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<infer function (type parameter) _A_A, infer function (type parameter) _E_E, infer function (type parameter) _R_R> ? function (type parameter) _R_R : never
> =>
const fromChannel: <
Arr extends Arr.NonEmptyReadonlyArray<any>,
E,
R
>(
channel: Channel.Channel<
Arr,
E,
void,
unknown,
unknown,
unknown,
R
>
) => Stream<
Arr extends Arr.NonEmptyReadonlyArray<infer A>
? A
: never,
E,
R
>
Creates a stream from a array-emitting Channel.
Example (Creating a stream from an array-emitting channel)
import { Channel, Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const channel = Channel.succeed([1, 2, 3] as const)
const stream = Stream.fromChannel(channel)
const result = yield* Stream.runCollect(stream)
yield* Console.log(result)
})
// Output: [ 1, 2, 3 ]
fromChannel(import ChannelChannel.suspend(() => {
const const latest: any[]latest: interface Array<T>Array<any> = []
const const emitted: Set<number>emitted = new var Set: SetConstructor
new <number>(iterable?: Iterable<number> | null | undefined) => Set<number> (+1 overload)
const readyLatch: Latch.Latchconst readyLatch: {
open: Effect.Effect<boolean>;
openUnsafe: (this: Latch) => boolean;
release: Effect.Effect<boolean>;
await: Effect.Effect<void>;
close: Effect.Effect<boolean>;
closeUnsafe: (this: Latch) => boolean;
whenOpen: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
isOpen: (this: Latch) => boolean;
}
import LatchLatch.makeUnsafe()
return import ChannelChannel.mergeAll(
import ChannelChannel.fromArray(
streams: T extends ReadonlyArray<Stream<any, any, any>>streams.ReadonlyArray<Stream<any, any, any>>.map<any>(callbackfn: (value: Stream<any, any, any>, index: number, array: readonly Stream<any, any, any>[]) => any, thisArg?: any): any[]Calls a defined callback function on each element of an array, and returns an array that contains the results.
map((s: Stream<any, any, any>(parameter) s: {
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; <…;
}
i: numberi) =>
s: Stream<any, any, any>(parameter) s: {
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; <…;
}
Stream<out A, out E = never, out R = never>.channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>(property) Stream<out A, out E = never, out R = never>.channel: {
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 ChannelChannel.flattenArray,
import ChannelChannel.mapEffect((a: anya) => {
const latest: any[]latest[i: numberi] = a: anya
if (!const emitted: Set<number>emitted.Set<number>.has(value: number): booleanhas(i: numberi)) {
const emitted: Set<number>emitted.Set<number>.add(value: number): Set<number>Appends a new element with a specified value to the end of the Set.
add(i: numberi)
if (const emitted: Set<number>emitted.Set<number>.size: numbersize < streams: T extends ReadonlyArray<Stream<any, any, any>>streams.ReadonlyArray<T>.length: numberGets the length of the array. This is a number one higher than the highest element defined in an array.
length) {
return const readyLatch: Latch.Latchconst readyLatch: {
open: Effect.Effect<boolean>;
openUnsafe: (this: Latch) => boolean;
release: Effect.Effect<boolean>;
await: Effect.Effect<void>;
close: Effect.Effect<boolean>;
closeUnsafe: (this: Latch) => boolean;
whenOpen: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
isOpen: (this: Latch) => boolean;
}
import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<undefined>
}
return import EffectEffect.as(const readyLatch: Latch.Latchconst readyLatch: {
open: Effect.Effect<boolean>;
openUnsafe: (this: Latch) => boolean;
release: Effect.Effect<boolean>;
await: Effect.Effect<void>;
close: Effect.Effect<boolean>;
closeUnsafe: (this: Latch) => boolean;
whenOpen: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
isOpen: (this: Latch) => boolean;
}
import ArrArr.of(const latest: any[]latest.Array<any>.slice(start?: number, end?: number): any[]Returns a copy of a section of an array.
For both start and end, a negative index can be used to indicate an offset from the end of the array.
For example, -2 refers to the second to last element of the array.
slice()))
}
return import EffectEffect.succeed(import ArrArr.of(const latest: any[]latest.Array<any>.slice(start?: number, end?: number): any[]Returns a copy of a section of an array.
For both start and end, a negative index can be used to indicate an offset from the end of the array.
For example, -2 refers to the second to last element of the array.
slice()))
}),
import ChannelChannel.filter(import isNotUndefinedisNotUndefined)
)
)
),
{
concurrency: stringconcurrency: "unbounded",
bufferSize: numberbufferSize: 0
}
)
})) as any