diff --git a/CHANGELOG.md b/CHANGELOG.md index ba628b4..5a02bc5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ **Unreleased:** - Subjects: fix a deadlock when sending values or termination concurrently with consumer cancellation (https://github.com/sideeffect-io/AsyncExtensions/issues/52). +- Multicast: preserve upstream element order and prevent termination from overtaking values when consumers advance concurrently. **v0.5.2 - Oxygen:** diff --git a/Sources/Operators/AsyncMulticastSequence.swift b/Sources/Operators/AsyncMulticastSequence.swift index 1cc4971..697912a 100644 --- a/Sources/Operators/AsyncMulticastSequence.swift +++ b/Sources/Operators/AsyncMulticastSequence.swift @@ -128,14 +128,15 @@ where Base.Element == Subject.Element, Subject.Failure == Error, Base.AsyncItera } self.state.withCriticalRegion { state in + // Broadcast before another consumer can advance upstream, including its termination. + // This lock belongs to multicast and is not acquired by subject cancellation handlers. + switch toSend { + case .success(.some(let element)): self.subject.send(element) + case .success(.none): self.subject.send(.finished) + case .failure(let error): self.subject.send(.failure(error)) + } state = .available(iterator) } - - switch toSend { - case .success(.some(let element)): self.subject.send(element) - case .success(.none): self.subject.send(.finished) - case .failure(let error): self.subject.send(.failure(error)) - } }.value } diff --git a/Tests/Operators/AsyncMulticastSequenceTests.swift b/Tests/Operators/AsyncMulticastSequenceTests.swift index ffe9828..8693e11 100644 --- a/Tests/Operators/AsyncMulticastSequenceTests.swift +++ b/Tests/Operators/AsyncMulticastSequenceTests.swift @@ -40,6 +40,52 @@ private class SpyAsyncSequenceForNumberOfIterators: AsyncSequence { } final class AsyncMulticastSequenceTests: XCTestCase { + func test_concurrent_consumers_receive_all_elements_in_order_before_finish() async { + await assertConcurrentDelivery() + } + + func test_concurrent_consumers_receive_all_elements_in_order_before_failure() async { + await assertConcurrentDelivery(failure: MockError(code: 1701)) + } + + private func assertConcurrentDelivery(failure: MockError? = nil) async { + let elements = Array(0..<100) + + for _ in 0..<20 { + let upstream = AsyncThrowingStream { continuation in + elements.forEach { continuation.yield($0) } + continuation.finish(throwing: failure) + } + let sut = upstream.multicast(AsyncThrowingPassthroughSubject()) + // Register every subscriber before allowing any of them to advance upstream. + let iterators = (0..<8).map { _ in sut.makeAsyncIterator() } + let finished = expectation(description: "All concurrent subscribers finish") + finished.expectedFulfillmentCount = iterators.count + sut.connect() + + let consumers = iterators.map { iterator in + Task { + defer { finished.fulfill() } + var iterator = iterator + var received = [Int]() + do { + while let element = try await iterator.next() { + received.append(element) + } + XCTAssertNil(failure) + } catch { + XCTAssertNotNil(failure) + XCTAssertEqual(error as? MockError, failure) + } + XCTAssertEqual(received, elements) + } + } + + await fulfillment(of: [finished], timeout: 5) + consumers.forEach { $0.cancel() } + } + } + func test_multiple_loops_receive_elements_from_single_baseIterator() { let taskHaveIterators = expectation(description: "All tasks have their iterator") taskHaveIterators.expectedFulfillmentCount = 2