Merge pull request #23 from sideeffect-io/refactor

flatMapLatest: uncomment tests

Author: twittemb

README.md

@@ -16,7 +16,7 @@
 * [AsyncThrowingBufferedChannel](./Sources/AsyncChannels/AsyncThrowingBufferedChannel.swift): Throwing buffered communication channel between tasks
 
 ### Subjects
-* [AsyncPassthroughSubject](./Sources/AsyncSubjects/AsyncPassthroughSubject.swift): Subject with a shared output
+* [AsyncPassthroughSubject](./Sources/AsyncSubjects/AsyncPassthroughSubject.swift): Subject with a shared output. The elements are not shared and will be spread across consumers (same as AsyncStream)
 * [AsyncThrowingPassthroughSubject](./Sources/AsyncSubjects/AsyncThrowingPassthroughSubject.swift): Throwing subject with a shared output
 * [AsyncCurrentValueSubject](./Sources/AsyncSubjects/AsyncCurrentValueSubject.swift): Subject with a shared output. Maintain an replays its current value
 * [AsyncThrowingCurrentValueSubject](./Sources/AsyncSubjects/AsyncThrowingCurrentValueSubject.swift): Throwing subject with a shared output. Maintain an replays its current value
@@ -45,13 +45,13 @@
 ### Operators
 * [`handleEvents()`](./Sources/Operators/AsyncHandleEventsSequence.swift): Executes closures during the lifecycle of the self
 * [`mapToResult()`](./Sources/Operators/AsyncMapToResultSequence.swift): Maps elements and failure from self to a `Result` type
-* [`multicast(_:)`](./Sources/Operators/AsyncMulticastSequence.swift): Shares values from self to several consumers thanks to a provided Suject
 * [`prepend(_:)`](./Sources/Operators/AsyncPrependSequence.swift): Prepends an element to self
 * [`scan(_:_:)`](./Sources/Operators/AsyncScanSequence.swift): Transforms elements from self by providing the current element to a closure along with the last value returned by the closure
 * [`assign(_:)`](./Sources/Operators/AsyncSequence+Assign.swift): Assigns elements from self to a property
 * [`collect(_:)`](./Sources/Operators/AsyncSequence+Collect.swift): Iterate over elements from self and execute a closure
 * [`eraseToAnyAsyncSequence()`](./Sources/Operators/AsyncSequence+EraseToAnyAsyncSequence.swift): Erases to AnyAsyncSequence
 * [`flatMapLatest(_:)`](./Sources/Operators/AsyncSequence+FlatMapLatest.swift): Transforms elements from self into a `AsyncSequence` and republishes elements sent by the most recently received `AsyncSequence` when self is an `AsyncSequence` of `AsyncSequence`
+* [`multicast(_:)`](./Sources/Operators/AsyncMulticastSequence.swift): Shares values from self to several consumers thanks to a provided Suject
 * [`share()`](./Sources/Operators/AsyncSequence+Share.swift): Shares values from self to several consumers
 * [`switchToLatest()`](./Sources/Operators/AsyncSwitchToLatestSequence.swift): Republishes elements sent by the most recently received `AsyncSequence` when self is an `AsyncSequence` of `AsyncSequence`
 

Tests/Operators/AsyncMulticastSequenceTests.swift

@@ -11,16 +11,16 @@ import XCTest
 private struct SpyAsyncSequenceForOnNextCall<Element>: AsyncSequence {
   typealias Element = Element
   typealias AsyncIterator = Iterator
-
+  
   let onNext: () -> Void
-
+  
   func makeAsyncIterator() -> AsyncIterator {
     Iterator(onNext: self.onNext)
   }
-
+  
   struct Iterator: AsyncIteratorProtocol {
     let onNext: () -> Void
-
+    
     func next() async throws -> Element? {
       self.onNext()
       try await Task.sleep(nanoseconds: 100_000_000_000)
@@ -32,26 +32,26 @@ private struct SpyAsyncSequenceForOnNextCall<Element>: AsyncSequence {
 private class SpyAsyncSequenceForNumberOfIterators<Element>: AsyncSequence {
   typealias Element = Element
   typealias AsyncIterator = Iterator
-
+  
   let element: Element
   let numberOfTimes: Int
-
+  
   var numberOfIterators = 0
-
+  
   init(element: Element, numberOfTimes: Int) {
     self.element = element
     self.numberOfTimes = numberOfTimes
   }
-
+  
   func makeAsyncIterator() -> AsyncIterator {
     self.numberOfIterators += 1
     return Iterator(element: self.element, numberOfTimes: self.numberOfTimes)
   }
-
+  
   struct Iterator: AsyncIteratorProtocol {
     let element: Element
     var numberOfTimes: Int
-
+    
     mutating func next() async throws -> Element? {
       guard self.numberOfTimes > 0 else { return nil }
       self.numberOfTimes -= 1
@@ -64,14 +64,14 @@ final class AsyncMulticastSequenceTests: XCTestCase {
   func test_multiple_loops_receive_elements_from_single_baseIterator() {
     let taskHaveIterators = expectation(description: "All tasks have their iterator")
     taskHaveIterators.expectedFulfillmentCount = 2
-
+    
     let tasksHaveFinishedExpectation = expectation(description: "Tasks have finished")
     tasksHaveFinishedExpectation.expectedFulfillmentCount = 2
-
+    
     let spyUpstreamSequence = SpyAsyncSequenceForNumberOfIterators(element: 1, numberOfTimes: 3)
     let stream = AsyncThrowingPassthroughSubject<Int, Error>()
     let sut = spyUpstreamSequence.multicast(stream)
-
+    
     Task {
       var receivedElement = [Int]()
       var iterator = sut.makeAsyncIterator()
@@ -82,7 +82,7 @@ final class AsyncMulticastSequenceTests: XCTestCase {
       XCTAssertEqual(receivedElement, [1, 1, 1])
       tasksHaveFinishedExpectation.fulfill()
     }
-
+    
     Task {
       var receivedElement = [Int]()
       var iterator = sut.makeAsyncIterator()
@@ -93,27 +93,27 @@ final class AsyncMulticastSequenceTests: XCTestCase {
       XCTAssertEqual(receivedElement, [1, 1, 1])
       tasksHaveFinishedExpectation.fulfill()
     }
-
+    
     wait(for: [taskHaveIterators], timeout: 1)
-
+    
     sut.connect()
-
+    
     wait(for: [tasksHaveFinishedExpectation], timeout: 1)
-
+    
     XCTAssertEqual(spyUpstreamSequence.numberOfIterators, 1)
   }
-
+  
   func test_multiple_loops_uses_provided_stream() {
     let taskHaveIterators = expectation(description: "All tasks have their iterator")
     taskHaveIterators.expectedFulfillmentCount = 3
-
+    
     let tasksHaveFinishedExpectation = expectation(description: "Tasks have finished")
     tasksHaveFinishedExpectation.expectedFulfillmentCount = 3
-
+    
     let stream = AsyncThrowingPassthroughSubject<Int, Error>()
     let spyUpstreamSequence = SpyAsyncSequenceForNumberOfIterators(element: 1, numberOfTimes: 3)
     let sut = spyUpstreamSequence.multicast(stream)
-
+    
     Task {
       var receivedElement = [Int]()
       var iterator = sut.makeAsyncIterator()
@@ -124,7 +124,7 @@ final class AsyncMulticastSequenceTests: XCTestCase {
       XCTAssertEqual(receivedElement, [1, 1, 1])
       tasksHaveFinishedExpectation.fulfill()
     }
-
+    
     Task {
       var receivedElement = [Int]()
       var iterator = sut.makeAsyncIterator()
@@ -135,7 +135,7 @@ final class AsyncMulticastSequenceTests: XCTestCase {
       XCTAssertEqual(receivedElement, [1, 1, 1])
       tasksHaveFinishedExpectation.fulfill()
     }
-
+    
     Task {
       var receivedElement = [Int]()
       var iterator = sut.makeAsyncIterator()
@@ -146,26 +146,26 @@ final class AsyncMulticastSequenceTests: XCTestCase {
       XCTAssertEqual(receivedElement, [1, 1, 1])
       tasksHaveFinishedExpectation.fulfill()
     }
-
+    
     wait(for: [taskHaveIterators], timeout: 1)
-
+    
     sut.connect()
-
+    
     wait(for: [tasksHaveFinishedExpectation], timeout: 1)
-
+    
     XCTAssertEqual(spyUpstreamSequence.numberOfIterators, 1)
   }
-
+  
   func test_multicast_propagates_error_when_autoconnect() async {
     let expectedError = MockError(code: Int.random(in: 0...100))
-
+    
     let stream = AsyncThrowingPassthroughSubject<Int, Error>()
-
+    
     let sut = AsyncFailSequence<Int>(expectedError)
       .prepend(1)
       .multicast(stream)
       .autoconnect()
-
+    
     var receivedElement = [Int]()
     do {
       for try await element in sut {
@@ -177,45 +177,45 @@ final class AsyncMulticastSequenceTests: XCTestCase {
       XCTAssertEqual(error as? MockError, expectedError)
     }
   }
-
+  
   func test_multicast_finishes_when_task_is_cancelled() {
     let taskHasFinishedExpectation = expectation(description: "Task has finished")
-
+    
     let stream = AsyncThrowingPassthroughSubject<Int, Error>()
     let sut = AsyncLazySequence<[Int]>([1, 2, 3, 4, 5])
       .multicast(stream)
       .autoconnect()
-
+    
     Task {
       for try await _ in sut {}
       taskHasFinishedExpectation.fulfill()
     }.cancel()
-
+    
     wait(for: [taskHasFinishedExpectation], timeout: 1)
   }
-
+  
   func test_multicast_finishes_when_task_is_cancelled_while_waiting_for_next() {
     let canCancelExpectation = expectation(description: "the task can be cancelled")
     let taskHasFinishedExpectation = expectation(description: "Task has finished")
-
+    
     let spyAsyncSequence = SpyAsyncSequenceForOnNextCall<Int> {
       canCancelExpectation.fulfill()
     }
-
+    
     let stream = AsyncThrowingPassthroughSubject<Int, Error>()
     let sut = spyAsyncSequence
       .multicast(stream)
       .autoconnect()
-
+    
     let task = Task {
       for try await _ in sut {}
       taskHasFinishedExpectation.fulfill()
     }
-
+    
     wait(for: [canCancelExpectation], timeout: 1)
-
+    
     task.cancel()
-
+    
     wait(for: [taskHasFinishedExpectation], timeout: 1)
   }
 }

Tests/Operators/AsyncSequence+ShareTests.swift

@@ -8,30 +8,79 @@
 import AsyncExtensions
 import XCTest
 
-final class AsyncSequence_ShareTests: XCTestCase {
-    func test_share_multicasts_values_to_clientLoops() {
-        let tasksHaveFinishedExpectation = expectation(description: "the tasks have finished")
-        tasksHaveFinishedExpectation.expectedFulfillmentCount = 2
-
-        let sut = AsyncLazySequence(["first", "second", "third"])
-            .share()
-
-        Task {
-            var received = [String]()
-            await sut
-                .collect { received.append($0) }
-            XCTAssertEqual(received, ["first", "second", "third"])
-            tasksHaveFinishedExpectation.fulfill()
-        }
+private extension DispatchTimeInterval {
+  var nanoseconds: UInt64 {
+    switch self {
+      case .nanoseconds(let value) where value >= 0: return UInt64(value)
+      case .microseconds(let value) where value >= 0: return UInt64(value) * 1000
+      case .milliseconds(let value) where value >= 0: return UInt64(value) * 1_000_000
+      case .seconds(let value) where value >= 0: return UInt64(value) * 1_000_000_000
+      case .never: return .zero
+      default: return .zero
+    }
+  }
+}
 
-        Task {
-            var received = [String]()
-            await sut
-                .collect { received.append($0) }
-            XCTAssertEqual(received, ["first", "second", "third"])
-            tasksHaveFinishedExpectation.fulfill()
-        }
+private struct LongAsyncSequence<Element>: AsyncSequence, AsyncIteratorProtocol {
+  typealias Element = Element
+  typealias AsyncIterator = LongAsyncSequence
+  
+  var elements: IndexingIterator<[Element]>
+  let interval: DispatchTimeInterval
+  var currentIndex = -1
+  let failAt: Int?
+  var hasEmitted = false
+  let onCancel: () -> Void
+  
+  init(elements: [Element], interval: DispatchTimeInterval = .seconds(0), failAt: Int? = nil, onCancel: @escaping () -> Void = {}) {
+    self.onCancel = onCancel
+    self.elements = elements.makeIterator()
+    self.failAt = failAt
+    self.interval = interval
+  }
+  
+  mutating func next() async throws -> Element? {
+    return try await withTaskCancellationHandler { [onCancel] in
+      onCancel()
+    } operation: {
+      try await Task.sleep(nanoseconds: self.interval.nanoseconds)
+      self.currentIndex += 1
+      if self.currentIndex == self.failAt {
+        throw MockError(code: 0)
+      }
+      return self.elements.next()
+    }
+  }
+  
+  func makeAsyncIterator() -> AsyncIterator {
+    self
+  }
+}
 
-        waitForExpectations(timeout: 5)
+final class AsyncSequence_ShareTests: XCTestCase {
+  func test_share_multicasts_values_to_clientLoops() {
+    let tasksHaveFinishedExpectation = expectation(description: "the tasks have finished")
+    tasksHaveFinishedExpectation.expectedFulfillmentCount = 2
+    
+    let sut = LongAsyncSequence(
+      elements: ["1", "2", "3"],
+      interval: .milliseconds(200)
+    ).share()
+    
+    Task(priority: .high) {
+      var received = [String]()
+      try await sut.collect { received.append($0) }
+      XCTAssertEqual(received, ["1", "2", "3"])
+      tasksHaveFinishedExpectation.fulfill()
+    }
+    
+    Task(priority: .high) {
+      var received = [String]()
+      try await sut.collect { received.append($0) }
+      XCTAssertEqual(received, ["1", "2", "3"])
+      tasksHaveFinishedExpectation.fulfill()
     }
+    
+    waitForExpectations(timeout: 5)
+  }
 }