Implement async sequence output at (#178)

Feature parity with Combine - https://developer.apple.com/documentation/combine/publisher/output(at:)

1. Implement `output(at:)` similar to `collect`.
2. Document.
3. Test.
4. Extract some repeated test helpers to avoid copying them.

Author: iThinker

Sources/Afluent/SequenceOperators/OutputSequence.swift

@@ -0,0 +1,51 @@
+//
+//  OutputSequence.swift
+//  Afluent
+//
+//  Created by Roman Temchenko on 2025-03-05.
+//
+
+import Foundation
+
+extension AsyncSequences {
+    public struct OutputAt<Upstream: AsyncSequence & Sendable>: AsyncSequence, Sendable {
+        public typealias Element = Upstream.Element
+        let upstream: Upstream
+        let index: Int
+
+        public struct AsyncIterator: AsyncIteratorProtocol {
+            var upstreamIterator: Upstream.AsyncIterator
+            let index: Int
+            var nextIndex = 0
+
+            public mutating func next() async throws -> Element? {
+                guard nextIndex <= index else { return nil }
+                while let next = try await upstreamIterator.next() {
+                    if nextIndex == index {
+                        nextIndex &+= 1
+                        return next
+                    }
+                    nextIndex &+= 1
+                }
+                return nil
+            }
+        }
+
+        public func makeAsyncIterator() -> AsyncIterator {
+            AsyncIterator(upstreamIterator: upstream.makeAsyncIterator(), index: index)
+        }
+    }
+    
+}
+
+extension AsyncSequence where Self: Sendable {
+    
+    /// Returns a sequence containing a specific indexed element.
+    /// If the sequence finishes normally or with an error before emitting the specified element, then the sequence doesn’t produce any elements.
+    /// - Parameter index: The index that indicates the element needed.
+    /// - Returns: A sequence containing a specific indexed element.
+    public func output(at index: Int) -> AsyncSequences.OutputAt<Self> {
+        AsyncSequences.OutputAt(upstream: self, index: index)
+    }
+    
+}

Tests/AfluentTests/SequenceTests/CollectSequenceTests.swift

@@ -37,29 +37,11 @@ struct CollectSequenceTests {
     }
 
     @Test func testCollectWithSequenceThrowingError() async throws {
-        enum TestError: Error, Equatable {
-            case someError
-        }
-
         let errorSequence = AsyncThrowingStream<Int, Error> { continuation in
-            continuation.finish(throwing: TestError.someError)
+            continuation.finish(throwing: GeneralError.e1)
         }
-        do {
+        await #expect(throws: GeneralError.e1, performing: {
             _ = try await errorSequence.collect().first()
-            Issue.record("Collect should throw the error encountered in the sequence")
-        } catch {
-            #expect(error as? TestError == .someError, "Collect should throw the correct error")
-        }
-    }
-}
-
-extension Array where Element: Sendable {
-    fileprivate var async: AsyncStream<Element> {
-        AsyncStream { continuation in
-            for element in self {
-                continuation.yield(element)
-            }
-            continuation.finish()
-        }
+        })
     }
 }

Tests/AfluentTests/SequenceTests/DelaySequenceTests.swift

@@ -130,14 +130,3 @@ struct DelaySequenceTests {
         }
     }
 }
-
-extension Array where Element: Sendable {
-    fileprivate var async: AsyncStream<Element> {
-        AsyncStream { continuation in
-            for element in self {
-                continuation.yield(element)
-            }
-            continuation.finish()
-        }
-    }
-}

Tests/AfluentTests/SequenceTests/HandleEventsSequenceTests.swift

@@ -159,14 +159,3 @@ struct HandleEventsSequenceTests {
         }
     }
 }
-
-extension Array where Element: Sendable {
-    fileprivate var async: AsyncStream<Element> {
-        AsyncStream { continuation in
-            for element in self {
-                continuation.yield(element)
-            }
-            continuation.finish()
-        }
-    }
-}

Tests/AfluentTests/SequenceTests/OutputSequenceTests.swift

@@ -0,0 +1,63 @@
+//
+//  OutputSequenceTests.swift
+//  Afluent
+//
+//  Created by Roman Temchenko on 2025-03-07.
+//
+
+import Afluent
+import Foundation
+import Testing
+import ConcurrencyExtras
+
+struct OutputSequenceTests {
+    
+    @Test func testOutputAt() async throws {
+        let emptySequence = [0, 3, 5].async
+        let result = try await emptySequence.output(at: 1).first()
+        #expect(result == 3)
+    }
+    
+    @Test func testOutputAtWithEmptySequence() async throws {
+        let emptySequence = [Int]().async
+        let result = try await emptySequence.output(at: 0).first()
+        try #require(result == nil)
+    }
+    
+    @Test func testOutputAtOutOfBounds() async throws {
+        let emptySequence = [0, 3, 5].async
+        let result = try await emptySequence.output(at: 5).first()
+        #expect(result == nil)
+    }
+    
+    @Test func testOutputAtWithSequenceThrowingError() async throws {
+        let errorSequence = AsyncThrowingStream<Int, Error> { continuation in
+            continuation.finish(throwing: GeneralError.e1)
+        }
+        await #expect(throws: GeneralError.e1, performing: {
+            _ = try await errorSequence.output(at: 0).first()
+        })
+    }
+    
+    @Test func testOutputAtCancellation() async throws {
+        let (stream, continuation) = AsyncThrowingStream.makeStream(of: Int.self)
+        
+        let task = Task {
+            let result = try await stream.output(at: 4).first()
+            #expect(result == nil)
+            return result
+        }
+        
+        continuation.yield(0)
+        task.cancel()
+        
+        // Give task cancellation time to propagate.
+        await Task.megaYield()
+        
+        continuation.finish(throwing: GeneralError.e1)
+        
+        let result = try await task.value
+        #expect(result == nil)
+    }
+    
+}

Tests/AfluentTests/SequenceTests/ShareSequenceTests.swift

@@ -326,17 +326,6 @@ func throwOn<T: Equatable>(_ toThrowOn: T, _ value: T) throws -> T {
     return value
 }
 
-extension Array where Element: Sendable {
-    fileprivate var async: AsyncStream<Element> {
-        AsyncStream<Element> {
-            for item in self {
-                $0.yield(item)
-            }
-            $0.finish()
-        }
-    }
-}
-
 extension RangeReplaceableCollection {
     fileprivate init<Source: AsyncSequence>(_ source: Source) async rethrows
     where Source.Element == Element {

Tests/AfluentTests/TestHelpers/ArrayExtensions.swift

@@ -0,0 +1,19 @@
+//
+//  ArrayExtensions.swift
+//  Afluent
+//
+//  Created by Roman Temchenko on 2025-03-07.
+//
+
+import Foundation
+
+extension Array where Element: Sendable {
+    var async: AsyncStream<Element> {
+        AsyncStream<Element> {
+            for item in self {
+                $0.yield(item)
+            }
+            $0.finish()
+        }
+    }
+}