feat(Stream): zipLatestAll (#2832)

Co-authored-by: Tim <hello@timsmart.co>

.changeset/calm-bags-admire.md

@@ -0,0 +1,5 @@
+---
+"effect": minor
+---
+
+add `Stream.zipLatestAll` api

packages/effect/dtslint/Stream.ts

@@ -236,3 +236,16 @@ pipe(
     _scope // $ExpectType { a: number; b: string; }
   ) => true)
 )
+
+// -------------------------------------------------------------------------------------
+// zipLatestAll
+// -------------------------------------------------------------------------------------
+
+// $ExpectType Stream<never, never, never>
+Stream.zipLatestAll()
+
+// $ExpectType Stream<[number, string | number], never, never>
+Stream.zipLatestAll(numbers, numbersOrStrings)
+
+// $ExpectType Stream<[number, string | number, never], Error, never>
+Stream.zipLatestAll(numbers, numbersOrStrings, Stream.fail(new Error("")))

packages/effect/src/Stream.ts

@@ -4350,6 +4350,45 @@ export const zipLatest: {
   <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>): Stream<[A, A2], E | E2, R | R2>
 } = internal.zipLatest
 
+/**
+ * Zips multiple streams so that when a value is emitted by any of the streams,
+ * it is combined with the latest values from the other streams to produce a result.
+ *
+ * Note: tracking the latest value is done on a per-chunk basis. That means
+ * that emitted elements that are not the last value in chunks will never be
+ * used for zipping.
+ *
+ * @example
+ * import { Stream, Schedule, Console, Effect } from "effect"
+ *
+ * const stream = Stream.zipLatestAll(
+ *     Stream.fromSchedule(Schedule.spaced('1 millis')),
+ *     Stream.fromSchedule(Schedule.spaced('2 millis')),
+ *     Stream.fromSchedule(Schedule.spaced('4 millis')),
+ * ).pipe(Stream.take(6), Stream.tap(Console.log))
+ *
+ * Effect.runPromise(Stream.runDrain(stream))
+ * // Output:
+ * // [ 0, 0, 0 ]
+ * // [ 1, 0, 0 ]
+ * // [ 1, 1, 0 ]
+ * // [ 2, 1, 0 ]
+ * // [ 3, 1, 0 ]
+ * // [ 3, 1, 1 ]
+ * // .....
+ *
+ * @since 3.3.0
+ * @category zipping
+ */
+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
+> = internal.zipLatestAll
+
 /**
  * Zips the two streams so that when a value is emitted by either of the two
  * streams, it is combined with the latest value from the other stream to

packages/effect/src/internal/stream.ts

@@ -7609,6 +7609,27 @@ export const zipLatest = dual<
   ): Stream.Stream<[A, A2], E2 | E, R2 | R> => pipe(self, zipLatestWith(that, (a, a2) => [a, a2]))
 )
 
+export const zipLatestAll = <T extends ReadonlyArray<Stream.Stream<any, any, any>>>(
+  ...streams: T
+): Stream.Stream<
+  [T[number]] extends [never] ? never
+    : { [K in keyof T]: T[K] extends Stream.Stream<infer A, infer _E, infer _R> ? A : never },
+  [T[number]] extends [never] ? never : T[number] extends Stream.Stream<infer _A, infer _E, infer _R> ? _E : never,
+  [T[number]] extends [never] ? never : T[number] extends Stream.Stream<infer _A, infer _E, infer _R> ? _R : never
+> => {
+  if (streams.length === 0) {
+    return empty
+  } else if (streams.length === 1) {
+    return map(streams[0]!, (x) => [x]) as any
+  }
+  const [head, ...tail] = streams
+  return zipLatestWith(
+    head,
+    zipLatestAll(...tail),
+    (first, second) => [first, ...second]
+  ) as any
+}
+
 /** @internal */
 export const zipLatestWith = dual<
   <A2, E2, R2, A, A3>(

packages/effect/test/Stream/zipping.test.ts

@@ -466,4 +466,26 @@ describe("Stream", () => {
       )
       assert.deepStrictEqual(Array.from(result1), Array.from(result2))
     })))
+
+  it.effect("zipLatestAll", () =>
+    Effect.gen(function*($) {
+      const result = yield* $(
+        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))
+        ),
+        Stream.runCollect,
+        Effect.map(Chunk.toReadonlyArray)
+      )
+      assert.deepStrictEqual(result, [
+        [1, "a", true],
+        [2, "a", true],
+        [3, "a", true],
+        [3, "b", true],
+        [3, "c", true],
+        [3, "c", false],
+        [3, "c", true]
+      ])
+    }))
 })