feat: Add initial naive implementation of flatMapLatest

This introduces a simplified version of the flatMapLatest operator for AsyncSequence, along with a basic unit test. Note that this implementation uses unstructured concurrency and may have race conditions regarding task cancellation.

Author: peterfriese

Sources/AsyncAlgorithms/AsyncFlatMapLatestSequence.swift

@@ -0,0 +1,68 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2022 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+//
+//===----------------------------------------------------------------------===//
+
+import Foundation
+
+@available(AsyncAlgorithms 1.0, *)
+extension AsyncSequence where Self: Sendable {
+  
+  /// Transforms elements into new asynchronous sequences, emitting elements
+  /// from the most recent inner sequence.
+  ///
+  /// When a new element is emitted by this sequence, the `transform`
+  /// is called to produce a new inner sequence. Iteration on the
+  /// previous inner sequence is cancelled, and iteration begins
+  /// on the new one.
+  public func flatMapLatest<T: AsyncSequence & Sendable>(
+    _ transform: @escaping @Sendable (Element) -> T
+  ) -> AsyncThrowingStream<T.Element, Error>
+  where T.Element: Sendable {
+    
+    AsyncThrowingStream { continuation in
+      let outerIterationTask = Task {
+        var innerIterationTask: Task<Void, Never>? = nil
+        
+        do {
+          for try await element in self {
+            innerIterationTask?.cancel()
+            
+            let innerSequence = transform(element)
+            
+            innerIterationTask = Task {
+              do {
+                for try await innerElement in innerSequence {
+                  try Task.checkCancellation()
+                  continuation.yield(innerElement)
+                }
+              } catch is CancellationError {
+                // Inner task was cancelled, this is normal
+              } catch {
+                // Inner sequence threw an error
+                continuation.finish(throwing: error)
+              }
+            }
+          }
+        } catch {
+          // Outer sequence threw an error
+          continuation.finish(throwing: error)
+        }
+        
+        // Outer sequence finished
+        await innerIterationTask?.value
+        continuation.finish()
+      }
+      
+      continuation.onTermination = { @Sendable _ in
+        outerIterationTask.cancel()
+      }
+    }
+  }
+}

Tests/AsyncAlgorithmsTests/TestFlatMapLatest.swift

@@ -0,0 +1,38 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2022 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+//
+//===----------------------------------------------------------------------===//
+
+import XCTest
+import AsyncAlgorithms
+
+@available(macOS 15.0, *)
+final class TestFlatMapLatest: XCTestCase {
+  
+  func test_simple_sequence() async throws {
+    let source = [1, 2, 3].async
+    let transformed = source.flatMapLatest { intValue in
+      return [intValue, intValue * 10].async
+    }
+
+    var expected = [3, 30]
+    do {
+      for try await element in transformed {
+        let (e, ex) = (element, expected.removeFirst())
+        print("\(e) == \(ex)")
+        
+        XCTAssertEqual(e, ex)
+      }
+    } catch {
+      XCTFail("Unexpected error: \(error)")
+    }
+    XCTAssertTrue(expected.isEmpty)
+  }
+
+}