Function zipLatestAll

packages/effect/src/Stream.ts:3779–3816

(
  ...streams: T
)

 * @since 3.3.0
 */
export 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
> =>
  fromChannel(Channel.suspend(() => {
    const latest: Array<any> = []
    const emitted = new Set<number>()
    const readyLatch = Latch.makeUnsafe()
    return Channel.mergeAll(
      Channel.fromArray(
        streams.map((s, i) =>
          s.channel.pipe(
            Channel.flattenArray,
            Channel.mapEffect((a) => {
              latest[i] = a
              if (!emitted.has(i)) {
                emitted.add(i)
                if (emitted.size < streams.length) {
                  return readyLatch.await as Effect.Effect<undefined>
                }
                return Effect.as(readyLatch.open, Arr.of(latest.slice()))
              }
              return Effect.succeed(Arr.of(latest.slice()))
            }),
            Channel.filter(isNotUndefined)
          )
        )
      ),
      {
        concurrency: "unbounded",
        bufferSize: 0
      }
    )
  })) as any

/**
 * Combines two streams by emitting each new element with the latest value from the other stream.