Got the combineLatestMany algorithm working, using Swift 6.2+

Author: czechboy0

Sources/AsyncAlgorithms/CombineLatest/AsyncCombineLatestManySequence.swift

@@ -9,6 +9,8 @@
 //
 //===----------------------------------------------------------------------===//
 
+#if compiler(>=6.2)
+
 /// Creates an asynchronous sequence that combines the latest values from many `AsyncSequence` types
 /// by emitting a tuple of the values. ``combineLatestMany(_:)`` only emits a value whenever any of the base `AsyncSequence`s
 /// emit a value (so long as each of the bases have emitted at least one value).
@@ -58,13 +60,12 @@ public struct AsyncCombineLatestManySequence<Element: Sendable, Failure: Error>:
       }
 
       func next() async throws(Failure) -> [Element]? {
-          fatalError()
-//        guard let element = try await self.storage.next() else {
-//          return nil
-//        }
-//
-//        // This force unwrap is safe since there must be a third element.
-//        return element
+        guard let element = try await self.storage.next() else {
+          return nil
+        }
+
+        // This force unwrap is safe since there must be a third element.
+        return element
       }
     }
 
@@ -82,3 +83,5 @@ public struct AsyncCombineLatestManySequence<Element: Sendable, Failure: Error>:
 
 @available(*, unavailable)
 extension AsyncCombineLatestManySequence.Iterator: Sendable {}
+
+#endif

Sources/AsyncAlgorithms/CombineLatest/CombineLatestManyStateMachine.swift

@@ -9,6 +9,8 @@
 //
 //===----------------------------------------------------------------------===//
 
+#if compiler(>=6.2)
+
 import DequeModule
 
 /// State machine for combine latest
@@ -595,3 +597,5 @@ struct CombineLatestManyStateMachine<Element: Sendable, Failure: Error>: Sendabl
     }
   }
 }
+
+#endif

Sources/AsyncAlgorithms/CombineLatest/CombineLatestManyStorage.swift

@@ -9,6 +9,8 @@
 //
 //===----------------------------------------------------------------------===//
 
+#if compiler(>=6.2)
+
 @available(AsyncAlgorithms 1.1, *)
 final class CombineLatestManyStorage<Element: Sendable, Failure: Error>: Sendable {
   typealias StateMachine = CombineLatestManyStateMachine<Element, Failure>
@@ -35,9 +37,9 @@ final class CombineLatestManyStorage<Element: Sendable, Failure: Error>: Sendabl
     }
   }
 
-  func next() async throws -> [Element]? {
-    try await withTaskCancellationHandler {
-      let result = await withUnsafeContinuation { continuation in
+  func next() async throws(Failure) -> [Element]? {
+    let result = await withTaskCancellationHandler {
+      await withUnsafeContinuation { continuation in
         let action: StateMachine.NextAction? = self.stateMachine.withCriticalRegion { stateMachine in
           let action = stateMachine.next(for: continuation)
           switch action {
@@ -82,9 +84,6 @@ final class CombineLatestManyStorage<Element: Sendable, Failure: Error>: Sendabl
           break
         }
       }
-
-      return try result._rethrowGet()
-
     } onCancel: {
       let action = self.stateMachine.withCriticalRegion { stateMachine in
         stateMachine.cancelled()
@@ -96,19 +95,20 @@ final class CombineLatestManyStorage<Element: Sendable, Failure: Error>: Sendabl
         let task,
         let upstreamContinuations
       ):
-          task.cancel()
+        task.cancel()
         upstreamContinuations.forEach { $0.resume() }
 
         downstreamContinuation.resume(returning: .success(nil))
 
       case .cancelTaskAndUpstreamContinuations(let task, let upstreamContinuations):
-          task.cancel()
+        task.cancel()
         upstreamContinuations.forEach { $0.resume() }
 
       case .none:
         break
       }
     }
+    return try result.get()
   }
 
   private func startTask(
@@ -234,3 +234,5 @@ final class CombineLatestManyStorage<Element: Sendable, Failure: Error>: Sendabl
     stateMachine.taskIsStarted(task: task, downstreamContinuation: downstreamContinuation)
   }
 }
+
+#endif

Tests/AsyncAlgorithmsTests/TestCombineLatestMany.swift

@@ -9,6 +9,8 @@
 //
 //===----------------------------------------------------------------------===//
 
+#if compiler(>=6.2)
+
 import XCTest
 import AsyncAlgorithms
 
@@ -46,40 +48,42 @@ final class TestCombineLatestMany: XCTestCase {
     c.advance()
 
     value = await validator.validate()
-    XCTAssertEqual(value, [(1, "a", 4)])
+    XCTAssertEqual(value, [[1, 4, 7]])
     a.advance()
 
     value = await validator.validate()
-    XCTAssertEqual(value, [(1, "a", 4), (2, "a", 4)])
+    XCTAssertEqual(value, [[1, 4, 7], [2, 4, 7]])
     b.advance()
 
     value = await validator.validate()
-    XCTAssertEqual(value, [(1, "a", 4), (2, "a", 4), (2, "b", 4)])
+    XCTAssertEqual(value, [[1, 4, 7], [2, 4, 7], [2, 5, 7]])
     c.advance()
 
     value = await validator.validate()
-    XCTAssertEqual(value, [(1, "a", 4), (2, "a", 4), (2, "b", 4), (2, "b", 5)])
+    XCTAssertEqual(value, [[1, 4, 7], [2, 4, 7], [2, 5, 7], [2, 5, 8]])
     a.advance()
 
     value = await validator.validate()
-    XCTAssertEqual(value, [(1, "a", 4), (2, "a", 4), (2, "b", 4), (2, "b", 5), (3, "b", 5)])
+    XCTAssertEqual(value, [[1, 4, 7], [2, 4, 7], [2, 5, 7], [2, 5, 8], [3, 5, 8]])
     b.advance()
 
     value = await validator.validate()
-    XCTAssertEqual(value, [(1, "a", 4), (2, "a", 4), (2, "b", 4), (2, "b", 5), (3, "b", 5), (3, "c", 5)])
+    XCTAssertEqual(value, [[1, 4, 7], [2, 4, 7], [2, 5, 7], [2, 5, 8], [3, 5, 8], [3, 6, 8]])
     c.advance()
 
     value = await validator.validate()
     XCTAssertEqual(
       value,
-      [(1, "a", 4), (2, "a", 4), (2, "b", 4), (2, "b", 5), (3, "b", 5), (3, "c", 5), (3, "c", 6)]
+      [[1, 4, 7], [2, 4, 7], [2, 5, 7], [2, 5, 8], [3, 5, 8], [3, 6, 8], [3, 6, 9]]
     )
 
     await fulfillment(of: [finished], timeout: 1.0)
     value = validator.current
     XCTAssertEqual(
       value,
-      [(1, "a", 4), (2, "a", 4), (2, "b", 4), (2, "b", 5), (3, "b", 5), (3, "c", 5), (3, "c", 6)]
+      [[1, 4, 7], [2, 4, 7], [2, 5, 7], [2, 5, 8], [3, 5, 8], [3, 6, 8], [3, 6, 9]]
     )
   }
 }
+
+#endif