Implement output in (#180)

Implement `output(in:)` similar to Combine - https://developer.apple.com/documentation/combine/publisher/output(in:)
Incoming ranges are reduced to `Range<Int>` same way as Apple does it in Combine.

Author: iThinker

Sources/Afluent/SequenceOperators/OutputSequence.swift

@@ -36,6 +36,35 @@ extension AsyncSequences {
         }
     }
 
+    public struct OutputIn<Upstream: AsyncSequence & Sendable>: AsyncSequence, Sendable {
+        public typealias Element = Upstream.Element
+        
+        let upstream: Upstream
+        let range: Range<Int>
+
+        public struct AsyncIterator: AsyncIteratorProtocol {
+            var upstreamIterator: Upstream.AsyncIterator
+            let range: Range<Int>
+            var nextIndex = 0
+
+            public mutating func next() async throws -> Element? {
+                guard nextIndex < range.upperBound else { return nil }
+                while let next = try await upstreamIterator.next() {
+                    if range.contains(nextIndex) {
+                        nextIndex &+= 1
+                        return next
+                    }
+                    nextIndex &+= 1
+                }
+                return nil
+            }
+        }
+
+        public func makeAsyncIterator() -> AsyncIterator {
+            AsyncIterator(upstreamIterator: upstream.makeAsyncIterator(), range: range)
+        }
+    }
+    
 }
 
 extension AsyncSequence where Self: Sendable {
@@ -48,4 +77,24 @@ extension AsyncSequence where Self: Sendable {
         AsyncSequences.OutputAt(upstream: self, index: index)
     }
 
+    /// Returns an async sequence that contains, in order, the elements of the base sequence specified by the range.
+    ///
+    /// ### Discussion:
+    /// Completes normally after range is exhausted.
+    /// 
+    /// ### Example:
+    /// ```swift
+    ///  let originalSequence = [0, 3, 5, 7, 9].async
+    ///  for try await element in originalSequence.output(in: 1..<4) {
+    ///      print("\(element)")
+    ///  }
+    ///  // Prints 3, 5, 7
+    /// ```
+    ///
+    /// - Parameter range: A range that indicates which elements to include.
+    /// - Returns: An async sequence that contains, in order, the elements of the base sequence specified by the range.
+    public func output<R>(in range: R) -> AsyncSequences.OutputIn<Self> where R : RangeExpression, R.Bound == Int, R: Sendable {
+        AsyncSequences.OutputIn(upstream: self,
+                                range: range.relative(to: 0..<Int.max))
+    }
 }

Tests/AfluentTests/SequenceTests/OutputSequenceTests.swift

@@ -5,17 +5,17 @@
 //  Created by Roman Temchenko on 2025-03-07.
 //
 
-import Afluent
+@testable 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)
+        let originalSequence = [0, 3, 5].async
+        let result = try await originalSequence.output(at: 1).collect().first()
+        #expect(result == [3])
     }
 
     @Test func testOutputAtWithEmptySequence() async throws {
@@ -25,8 +25,8 @@ struct OutputSequenceTests {
     }
 
     @Test func testOutputAtOutOfBounds() async throws {
-        let emptySequence = [0, 3, 5].async
-        let result = try await emptySequence.output(at: 5).first()
+        let originalSequence = [0, 3, 5].async
+        let result = try await originalSequence.output(at: 5).first()
         #expect(result == nil)
     }
 
@@ -60,4 +60,67 @@ struct OutputSequenceTests {
         #expect(result == nil)
     }
 
+    @Test func testOutputIn() async throws {
+        let originalSequence = [0, 3, 5, 7, 9].async
+        let result = try await originalSequence.output(in: 1..<4).collect().first()
+        #expect(result == [3, 5, 7])
+    }
+    
+    @Test func testOutputInClosedRange() async throws {
+        let originalSequence = [0, 3, 5, 7, 9].async
+        let result = try await originalSequence.output(in: 1...3).collect().first()
+        #expect(result == [3, 5, 7])
+    }
+    
+    @Test func testOutputInPartialRangeUpTo() async throws {
+        let originalSequence = [0, 3, 5, 7, 9].async
+        let result = try await originalSequence.output(in: ..<3).collect().first()
+        #expect(result == [0, 3, 5])
+    }
+    
+    @Test func testOutputInPartialRangeThrough() async throws {
+        let originalSequence = [0, 3, 5, 7, 9].async
+        let result = try await originalSequence.output(in: ...3).collect().first()
+        #expect(result == [0, 3, 5, 7])
+    }
+    
+    @Test func testOutputInPartialRangeFrom() async throws {
+        let originalSequence = [0, 3, 5, 7, 9].async
+        let result = try await originalSequence.output(in: 2...).collect().first()
+        #expect(result == [5, 7, 9])
+    }
+    
+    @Test func testOutputInWithEmptySequence() async throws {
+        let emptySequence = [Int]().async
+        let result = try await emptySequence.output(in: 0...).first()
+        try #require(result == nil)
+    }
+    
+    @Test func testOutputInOutOfBounds() async throws {
+        let originalSequence = [0, 3, 5].async
+        let result = try await originalSequence.output(in: 5...).first()
+        #expect(result == nil)
+    }
+    
+    @Test func testOutputInCancellation() async throws {
+        let (stream, continuation) = AsyncThrowingStream.makeStream(of: Int.self)
+        
+        let task = Task {
+            let result = try await stream.output(in: 5...).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)
+    }
+    
 }