From 2eeb7c2bb3e36a823680bfec80050811a5b2d3e8 Mon Sep 17 00:00:00 2001 From: harry lachenmayer Date: Wed, 6 Dec 2023 17:36:33 +0000 Subject: [PATCH 1/4] fix: remove overlapping operators from swift-async-algorithms --- Package.resolved | 13 +- Package.swift | 10 +- README.md | 7 - .../Combiners/Merge/AsyncMerge2Sequence.swift | 63 -- .../Combiners/Merge/AsyncMerge3Sequence.swift | 69 --- Sources/Combiners/Zip/AsyncZip2Sequence.swift | 61 -- Sources/Combiners/Zip/AsyncZip3Sequence.swift | 66 --- Sources/Combiners/Zip/Zip2Runtime.swift | 214 ------- Sources/Combiners/Zip/Zip2StateMachine.swift | 373 ------------ Sources/Combiners/Zip/Zip3Runtime.swift | 252 -------- Sources/Combiners/Zip/Zip3StateMachine.swift | 542 ------------------ .../Operators/AsyncPrependSequenceTests.swift | 1 + .../Operators/AsyncSequence+AssignTests.swift | 3 +- .../AsyncSequence+FlatMapLatestTests.swift | 5 +- .../AsyncSwitchToLatestSequenceTests.swift | 9 +- 15 files changed, 30 insertions(+), 1658 deletions(-) delete mode 100644 Sources/Combiners/Merge/AsyncMerge2Sequence.swift delete mode 100644 Sources/Combiners/Merge/AsyncMerge3Sequence.swift delete mode 100644 Sources/Combiners/Zip/AsyncZip2Sequence.swift delete mode 100644 Sources/Combiners/Zip/AsyncZip3Sequence.swift delete mode 100644 Sources/Combiners/Zip/Zip2Runtime.swift delete mode 100644 Sources/Combiners/Zip/Zip2StateMachine.swift delete mode 100644 Sources/Combiners/Zip/Zip3Runtime.swift delete mode 100644 Sources/Combiners/Zip/Zip3StateMachine.swift diff --git a/Package.resolved b/Package.resolved index e378654..1241cb6 100644 --- a/Package.resolved +++ b/Package.resolved @@ -1,13 +1,22 @@ { "object": { "pins": [ + { + "package": "swift-async-algorithms", + "repositoryURL": "https://github.com/apple/swift-async-algorithms.git", + "state": { + "branch": null, + "revision": "da4e36f86544cdf733a40d59b3a2267e3a7bbf36", + "version": "1.0.0" + } + }, { "package": "swift-collections", "repositoryURL": "https://github.com/apple/swift-collections.git", "state": { "branch": null, - "revision": "f504716c27d2e5d4144fa4794b12129301d17729", - "version": "1.0.3" + "revision": "a902f1823a7ff3c9ab2fba0f992396b948eda307", + "version": "1.0.5" } } ] diff --git a/Package.swift b/Package.swift index 4f5d569..cb582c4 100644 --- a/Package.swift +++ b/Package.swift @@ -16,7 +16,10 @@ let package = Package( name: "AsyncExtensions", targets: ["AsyncExtensions"]), ], - dependencies: [.package(url: "https://github.com/apple/swift-collections.git", .upToNextMajor(from: "1.0.3"))], + dependencies: [ + .package(url: "https://github.com/apple/swift-async-algorithms.git", .upToNextMajor(from: "1.0.0")), + .package(url: "https://github.com/apple/swift-collections.git", .upToNextMajor(from: "1.0.3")) + ], targets: [ .target( name: "AsyncExtensions", @@ -32,7 +35,10 @@ let package = Package( ), .testTarget( name: "AsyncExtensionsTests", - dependencies: ["AsyncExtensions"], + dependencies: [ + "AsyncExtensions", + .product(name: "AsyncAlgorithms", package: "swift-async-algorithms") + ], path: "Tests"), ] ) diff --git a/README.md b/README.md index 19bac57..f20fbd7 100644 --- a/README.md +++ b/README.md @@ -55,12 +55,6 @@ the value. The active sender drains pending deliveries before returning. A new c or replay consumer receives the latest stored state, followed by subsequent sends. ### Combiners -* [`zip(_:_:)`](./Sources/Combiners/Zip/AsyncZip2Sequence.swift): Zips two `AsyncSequence` into an AsyncSequence of tuple of elements -* [`zip(_:_:_:)`](./Sources/Combiners/Zip/AsyncZip3Sequence.swift): Zips three `AsyncSequence` into an AsyncSequence of tuple of elements -* [`zip(_:)`](./Sources/Combiners/Zip/AsyncZipSequence.swift): Zips any async sequences into an array of elements -* [`merge(_:_:)`](./Sources/Combiners/Merge/AsyncMerge2Sequence.swift): Merges two `AsyncSequence` into an AsyncSequence of elements -* [`merge(_:_:_:)`](./Sources/Combiners/Merge/AsyncMerge3Sequence.swift): Merges three `AsyncSequence` into an AsyncSequence of elements -* [`merge(_:)`](./Sources/Combiners/Merge/AsyncMergeSequence.swift): Merges any `AsyncSequence` into an AsyncSequence of elements * [`withLatest(_:)`](./Sources/Combiners/WithLatestFrom/AsyncWithLatestFromSequence.swift): Combines elements from self with the last known element from an other `AsyncSequence` * [`withLatest(_:_:)`](./Sources/Combiners/WithLatestFrom/AsyncWithLatestFrom2Sequence.swift): Combines elements from self with the last known elements from two other async sequences @@ -69,7 +63,6 @@ or replay consumer receives the latest stored state, followed by subsequent send * [AsyncFailSequence](./Sources/Creators/AsyncFailSequence.swift): Creates an `AsyncSequence` that immediately fails * [AsyncJustSequence](./Sources/Creators/AsyncJustSequence.swift): Creates an `AsyncSequence` that emits an element an finishes * [AsyncThrowingJustSequence](./Sources/Creators/AsyncThrowingJustSequence.swift): Creates an `AsyncSequence` that emits an elements and finishes bases on a throwing closure -* [AsyncLazySequence](./Sources/Creators/AsyncLazySequence.swift): Creates an `AsyncSequence` of the elements from the base sequence * [AsyncTimerSequence](./Sources/Creators/AsyncTimerSequence.swift): Creates an `AsyncSequence` that emits a date value periodically * [AsyncStream Pipe](./Sources/Creators/AsyncStream+Pipe.swift): Creates an AsyncStream and returns a tuple standing for its inputs and outputs diff --git a/Sources/Combiners/Merge/AsyncMerge2Sequence.swift b/Sources/Combiners/Merge/AsyncMerge2Sequence.swift deleted file mode 100644 index 89a4b23..0000000 --- a/Sources/Combiners/Merge/AsyncMerge2Sequence.swift +++ /dev/null @@ -1,63 +0,0 @@ -// -// AsyncMerge2Sequence.swift -// -// -// Created by Thibault Wittemberg on 31/03/2022. -// - -/// Creates an asynchronous sequence of elements from two underlying asynchronous sequences -public func merge( - _ base1: Base1, - _ base2: Base2 -) -> AsyncMerge2Sequence { - AsyncMerge2Sequence(base1, base2) -} - -/// An asynchronous sequence of elements from two underlying asynchronous sequences -/// -/// In a `AsyncMerge2Sequence` instance, the *i*th element is the *i*th element -/// resolved in sequential order out of the two underlying asynchronous sequences. -/// Use the `merge(_:_:)` function to create an `AsyncMerge2Sequence`. -public struct AsyncMerge2Sequence: AsyncSequence -where Base1.Element == Base2.Element { - public typealias Element = Base1.Element - public typealias AsyncIterator = Iterator - - let base1: Base1 - let base2: Base2 - - public init(_ base1: Base1, _ base2: Base2) { - self.base1 = base1 - self.base2 = base2 - } - - public func makeAsyncIterator() -> Iterator { - Iterator( - base1: self.base1, - base2: self.base2 - ) - } - - public struct Iterator: AsyncIteratorProtocol { - let mergeStateMachine: MergeStateMachine - - init(base1: Base1, base2: Base2) { - self.mergeStateMachine = MergeStateMachine( - base1, - base2 - ) - } - - public mutating func next() async rethrows -> Element? { - let mergedElement = await self.mergeStateMachine.next() - switch mergedElement { - case .element(let result): - return try result._rethrowGet() - case .termination: - return nil - } - } - } -} - -extension AsyncMerge2Sequence: Sendable where Base1: Sendable, Base2: Sendable {} diff --git a/Sources/Combiners/Merge/AsyncMerge3Sequence.swift b/Sources/Combiners/Merge/AsyncMerge3Sequence.swift deleted file mode 100644 index 468b1ee..0000000 --- a/Sources/Combiners/Merge/AsyncMerge3Sequence.swift +++ /dev/null @@ -1,69 +0,0 @@ -// -// AsyncMerge3Sequence.swift -// -// -// Created by Thibault Wittemberg on 31/03/2022. -// - -/// Creates an asynchronous sequence of elements from three underlying asynchronous sequences -public func merge( - _ base1: Base1, - _ base2: Base2, - _ base3: Base3 -) -> AsyncMerge3Sequence { - AsyncMerge3Sequence(base1, base2, base3) -} - -/// An asynchronous sequence of elements from three underlying asynchronous sequences -/// -/// In a `AsyncMerge3Sequence` instance, the *i*th element is the *i*th element -/// resolved in sequential order out of the two underlying asynchronous sequences. -/// Use the `merge(_:_:_:)` function to create an `AsyncMerge3Sequence`. -public struct AsyncMerge3Sequence: AsyncSequence -where Base1.Element == Base2.Element, Base3.Element == Base2.Element { - public typealias Element = Base1.Element - public typealias AsyncIterator = Iterator - - let base1: Base1 - let base2: Base2 - let base3: Base3 - - public init(_ base1: Base1, _ base2: Base2, _ base3: Base3) { - self.base1 = base1 - self.base2 = base2 - self.base3 = base3 - } - - public func makeAsyncIterator() -> Iterator { - Iterator( - base1: self.base1, - base2: self.base2, - base3: self.base3 - ) - } - - public struct Iterator: AsyncIteratorProtocol { - let mergeStateMachine: MergeStateMachine - - init(base1: Base1, base2: Base2, base3: Base3) { - self.mergeStateMachine = MergeStateMachine( - base1, - base2, - base3 - ) - } - - public mutating func next() async rethrows -> Element? { - let mergedElement = await self.mergeStateMachine.next() - switch mergedElement { - case .element(let result): - return try result._rethrowGet() - case .termination: - return nil - } - } - } -} - -extension AsyncMerge3Sequence: Sendable where Base1: Sendable, Base2: Sendable, Base3: Sendable {} -extension AsyncMerge3Sequence.Iterator: Sendable where Base1: Sendable, Base2: Sendable, Base3: Sendable {} diff --git a/Sources/Combiners/Zip/AsyncZip2Sequence.swift b/Sources/Combiners/Zip/AsyncZip2Sequence.swift deleted file mode 100644 index d9ebe93..0000000 --- a/Sources/Combiners/Zip/AsyncZip2Sequence.swift +++ /dev/null @@ -1,61 +0,0 @@ -// -// AsyncZip2Sequence.swift -// -// -// Created by Thibault Wittemberg on 13/01/2022. -// - -/// `zip` produces an `AsyncSequence` that combines the latest elements from two sequences according to their temporality -/// and emits a tuple to the client. If any Async Sequence ends successfully or fails with an error, so to does the zipped -/// Async Sequence. -/// -/// ``` -/// let asyncSequence1 = [1, 2, 3, 4, 5].async -/// let asyncSequence2 = ["1", "2", "3", "4", "5"].async -/// -/// let zippedAsyncSequence = zip(asyncSequence1, asyncSequence2) -/// -/// for await element in zippedAsyncSequence { -/// print(element) // will print -> (1, "1") (2, "2") (3, "3") (4, "4") (5, "5") -/// } -/// ``` -/// Use the `zip(_:_:)` function to create an `AsyncZip2Sequence`. -public func zip( - _ base1: Base1, - _ base2: Base2 -) -> AsyncZip2Sequence { - AsyncZip2Sequence(base1, base2) -} - -public struct AsyncZip2Sequence: AsyncSequence -where Base1: Sendable, Base1.Element: Sendable, Base2: Sendable, Base2.Element: Sendable { - public typealias Element = (Base1.Element, Base2.Element) - public typealias AsyncIterator = Iterator - - let base1: Base1 - let base2: Base2 - - init(_ base1: Base1, _ base2: Base2) { - self.base1 = base1 - self.base2 = base2 - } - - public func makeAsyncIterator() -> AsyncIterator { - Iterator( - base1, - base2 - ) - } - - public struct Iterator: AsyncIteratorProtocol { - let runtime: Zip2Runtime - - init(_ base1: Base1, _ base2: Base2) { - self.runtime = Zip2Runtime(base1, base2) - } - - public func next() async rethrows -> Element? { - try await self.runtime.next() - } - } -} diff --git a/Sources/Combiners/Zip/AsyncZip3Sequence.swift b/Sources/Combiners/Zip/AsyncZip3Sequence.swift deleted file mode 100644 index 89d8a24..0000000 --- a/Sources/Combiners/Zip/AsyncZip3Sequence.swift +++ /dev/null @@ -1,66 +0,0 @@ -// -// AsyncZip3Sequence.swift -// -// -// Created by Thibault Wittemberg on 24/09/2022. -// - -/// `zip` produces an `AsyncSequence` that combines the latest elements from three sequences according to their temporality -/// and emits a tuple to the client. If any Async Sequence ends successfully or fails with an error, so to does the zipped -/// Async Sequence. -/// -/// ``` -/// let asyncSequence1 = [1, 2, 3, 4, 5].async -/// let asyncSequence2 = ["1", "2", "3", "4", "5"].async -/// let asyncSequence3 = ["A", "B", "C", "D", "E"].async -/// -/// let zippedAsyncSequence = zip(asyncSequence1, asyncSequence2, asyncSequence3) -/// -/// for await element in zippedAsyncSequence { -/// print(element) // will print -> (1, "1", "A") (2, "2", "B") (3, "3", "V") (4, "4", "D") (5, "5", "E") -/// } -/// ``` -/// Use the `zip(_:_:_:)` function to create an `AsyncZip3Sequence`. -public func zip( - _ base1: Base1, - _ base2: Base2, - _ base3: Base3 -) -> AsyncZip3Sequence { - AsyncZip3Sequence(base1, base2, base3) -} - -public struct AsyncZip3Sequence: AsyncSequence -where Base1: Sendable, Base1.Element: Sendable, Base2: Sendable, Base2.Element: Sendable, Base3: Sendable, Base3.Element: Sendable { - public typealias Element = (Base1.Element, Base2.Element, Base3.Element) - public typealias AsyncIterator = Iterator - - let base1: Base1 - let base2: Base2 - let base3: Base3 - - init(_ base1: Base1, _ base2: Base2, _ base3: Base3) { - self.base1 = base1 - self.base2 = base2 - self.base3 = base3 - } - - public func makeAsyncIterator() -> AsyncIterator { - Iterator( - base1, - base2, - base3 - ) - } - - public struct Iterator: AsyncIteratorProtocol { - let runtime: Zip3Runtime - - init(_ base1: Base1, _ base2: Base2, _ base3: Base3) { - self.runtime = Zip3Runtime(base1, base2, base3) - } - - public func next() async rethrows -> Element? { - try await self.runtime.next() - } - } -} diff --git a/Sources/Combiners/Zip/Zip2Runtime.swift b/Sources/Combiners/Zip/Zip2Runtime.swift deleted file mode 100644 index f59cbb6..0000000 --- a/Sources/Combiners/Zip/Zip2Runtime.swift +++ /dev/null @@ -1,214 +0,0 @@ -// -// Zip2Runtime.swift -// -// -// Created by Thibault Wittemberg on 24/09/2022. -// - -final class Zip2Runtime: Sendable -where Base1: Sendable, Base1.Element: Sendable, Base2: Sendable, Base2.Element: Sendable { - typealias ZipStateMachine = Zip2StateMachine - - private let stateMachine = ManagedCriticalState(ZipStateMachine()) - - init(_ base1: Base1, _ base2: Base2) { - self.stateMachine.withCriticalRegion { machine in - machine.taskIsStarted(task: Task { - await withTaskGroup(of: Void.self) { group in - group.addTask { - var base1Iterator = base1.makeAsyncIterator() - - do { - while true { - await withUnsafeContinuation { (continuation: UnsafeContinuation) in - let output = self.stateMachine.withCriticalRegion { machine in - machine.newLoopFromBase1(suspendedBase: continuation) - } - - self.handle(newLoopFromBaseOutput: output) - } - - guard let element1 = try await base1Iterator.next() else { - break - } - - let output = self.stateMachine.withCriticalRegion { machine in - machine.base1HasProducedElement(element: element1) - } - - self.handle(baseHasProducedElementOutput: output) - } - } catch { - let output = self.stateMachine.withCriticalRegion { machine in - machine.baseHasProducedFailure(error: error) - } - - self.handle(baseHasProducedFailureOutput: output) - } - - let output = self.stateMachine.withCriticalRegion { stateMachine in - stateMachine.baseIsFinished() - } - - self.handle(baseIsFinishedOutput: output) - } - - group.addTask { - var base2Iterator = base2.makeAsyncIterator() - - do { - while true { - await withUnsafeContinuation { (continuation: UnsafeContinuation) in - let output = self.stateMachine.withCriticalRegion { machine in - machine.newLoopFromBase2(suspendedBase: continuation) - } - - self.handle(newLoopFromBaseOutput: output) - } - - guard let element2 = try await base2Iterator.next() else { - break - } - - let output = self.stateMachine.withCriticalRegion { machine in - machine.base2HasProducedElement(element: element2) - } - - self.handle(baseHasProducedElementOutput: output) - } - } catch { - let output = self.stateMachine.withCriticalRegion { machine in - machine.baseHasProducedFailure(error: error) - } - - self.handle(baseHasProducedFailureOutput: output) - } - - let output = self.stateMachine.withCriticalRegion { machine in - machine.baseIsFinished() - } - - self.handle(baseIsFinishedOutput: output) - } - } - }) - } - } - - private func handle(newLoopFromBaseOutput: ZipStateMachine.NewLoopFromBaseOutput) { - switch newLoopFromBaseOutput { - case .none: - break - - case .resumeBases(let suspendedBases): - suspendedBases.forEach { $0.resume() } - - case .terminate(let task, let suspendedBase, let suspendedDemand): - suspendedBase.resume() - suspendedDemand?.resume(returning: nil) - task?.cancel() - } - } - - private func handle(baseHasProducedElementOutput: ZipStateMachine.BaseHasProducedElementOutput) { - switch baseHasProducedElementOutput { - case .none: - break - - case .resumeDemand(let suspendedDemand, let result1, let result2): - suspendedDemand?.resume(returning: (result1, result2)) - - case .terminate(let task, let suspendedBases): - suspendedBases?.forEach { $0.resume() } - task?.cancel() - } - } - - private func handle(baseHasProducedFailureOutput: ZipStateMachine.BaseHasProducedFailureOutput) { - switch baseHasProducedFailureOutput { - case .resumeDemandAndTerminate(let task, let suspendedDemand, let suspendedBases, let result1, let result2): - suspendedDemand?.resume(returning: (result1, result2)) - suspendedBases.forEach { $0.resume() } - task?.cancel() - - case .terminate(let task, let suspendedBases): - suspendedBases?.forEach { $0.resume() } - task?.cancel() - } - } - - private func handle(baseIsFinishedOutput: ZipStateMachine.BaseIsFinishedOutput) { - switch baseIsFinishedOutput { - case .terminate(let task, let suspendedBases, let suspendedDemands): - suspendedBases?.forEach { $0.resume() } - suspendedDemands?.forEach { $0?.resume(returning: nil) } - task?.cancel() - } - } - - func next() async rethrows -> (Base1.Element, Base2.Element)? { - try await withTaskCancellationHandler { - let results = await withUnsafeContinuation { (continuation: UnsafeContinuation<(Result, Result)?, Never>) in - let output = self.stateMachine.withCriticalRegion { stateMachine in - stateMachine.newDemandFromConsumer(suspendedDemand: continuation) - } - - self.handle(newDemandFromConsumerOutput: output) - } - - guard let results = results else { - return nil - } - - let output = self.stateMachine.withCriticalRegion { stateMachine in - stateMachine.demandIsFulfilled() - } - - self.handle(demandIsFulfilledOutput: output) - - return try (results.0._rethrowGet(), results.1._rethrowGet()) - } onCancel: { - let output = self.stateMachine.withCriticalRegion { stateMachine in - stateMachine.rootTaskIsCancelled() - } - - self.handle(rootTaskIsCancelledOutput: output) - } - } - - private func handle(rootTaskIsCancelledOutput: ZipStateMachine.RootTaskIsCancelledOutput) { - switch rootTaskIsCancelledOutput { - case .terminate(let task, let suspendedBases, let suspendedDemands): - suspendedBases?.forEach { $0.resume() } - suspendedDemands?.forEach { $0?.resume(returning: nil) } - task?.cancel() - } - } - - private func handle(newDemandFromConsumerOutput: ZipStateMachine.NewDemandFromConsumerOutput) { - switch newDemandFromConsumerOutput { - case .none: - break - - case .resumeBases(let suspendedBases): - suspendedBases.forEach { $0.resume() } - - case .terminate(let task, let suspendedBases, let suspendedDemands): - suspendedBases?.forEach { $0.resume() } - suspendedDemands?.forEach { $0?.resume(returning: nil) } - task?.cancel() - } - } - - private func handle(demandIsFulfilledOutput: ZipStateMachine.DemandIsFulfilledOutput) { - switch demandIsFulfilledOutput { - case .none: - break - - case .terminate(let task, let suspendedBases, let suspendedDemands): - suspendedBases?.forEach { $0.resume() } - suspendedDemands?.forEach { $0.resume(returning: nil) } - task?.cancel() - } - } -} diff --git a/Sources/Combiners/Zip/Zip2StateMachine.swift b/Sources/Combiners/Zip/Zip2StateMachine.swift deleted file mode 100644 index 4b3b43e..0000000 --- a/Sources/Combiners/Zip/Zip2StateMachine.swift +++ /dev/null @@ -1,373 +0,0 @@ -// -// Zip2StateMachine.swift -// -// -// Created by Thibault Wittemberg on 24/09/2022. -// - -struct Zip2StateMachine: Sendable -where Element1: Sendable, Element2: Sendable { - private enum State { - case initial - case started(task: Task) - case awaitingDemandFromConsumer( - task: Task?, - suspendedBases: [UnsafeContinuation] - ) - case awaitingBaseResults( - task: Task?, - result1: Result?, - result2: Result?, - suspendedBases: [UnsafeContinuation], - suspendedDemand: UnsafeContinuation<(Result, Result)?, Never>? - ) - case finished - } - - private var state: State = .initial - - mutating func taskIsStarted(task: Task) { - switch self.state { - case .initial: - self.state = .started(task: task) - - default: - assertionFailure("Inconsistent state, the task cannot start while the state is other than initial") - } - } - - enum NewDemandFromConsumerOutput { - case none - case resumeBases(suspendedBases: [UnsafeContinuation]) - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]?, - suspendedDemands: [UnsafeContinuation<(Result, Result)?, Never>?]? - ) - } - - mutating func newDemandFromConsumer( - suspendedDemand: UnsafeContinuation<(Result, Result)?, Never> - ) -> NewDemandFromConsumerOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: [suspendedDemand]) - - case .started(let task): - self.state = .awaitingBaseResults(task: task, result1: nil, result2: nil, suspendedBases: [], suspendedDemand: suspendedDemand) - return .none - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - self.state = .awaitingBaseResults(task: task, result1: nil, result2: nil, suspendedBases: [], suspendedDemand: suspendedDemand) - return .resumeBases(suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, _, _, let suspendedBases, let oldSuspendedDemand): - assertionFailure("Inconsistent state, a demand is already suspended") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: [oldSuspendedDemand, suspendedDemand]) - - case .finished: - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: [suspendedDemand]) - } - } - - enum NewLoopFromBaseOutput { - case none - case resumeBases(suspendedBases: [UnsafeContinuation]) - case terminate( - task: Task?, - suspendedBase: UnsafeContinuation, - suspendedDemand: UnsafeContinuation<(Result, Result)?, Never>? - ) - } - - mutating func newLoopFromBase1(suspendedBase: UnsafeContinuation) -> NewLoopFromBaseOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - - case .started(let task): - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: [suspendedBase]) - return .none - - case .awaitingDemandFromConsumer(let task, var suspendedBases): - assert(suspendedBases.count < 2, "There cannot be more than 2 suspended base at the same time") - suspendedBases.append(suspendedBase) - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: suspendedBases) - return .none - - case .awaitingBaseResults(let task, let result1, let result2, var suspendedBases, let suspendedDemand): - assert(suspendedBases.count < 2, "There cannot be more than 2 suspended bases at the same time") - if result1 != nil { - suspendedBases.append(suspendedBase) - self.state = .awaitingBaseResults(task: task, result1: result1, result2: result2, suspendedBases: suspendedBases, suspendedDemand: suspendedDemand) - return .none - } else { - self.state = .awaitingBaseResults(task: task, result1: result1, result2: result2, suspendedBases: suspendedBases, suspendedDemand: suspendedDemand) - return .resumeBases(suspendedBases: [suspendedBase]) - } - - case .finished: - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - } - } - - mutating func newLoopFromBase2(suspendedBase: UnsafeContinuation) -> NewLoopFromBaseOutput { - switch state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - - case .started(let task): - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: [suspendedBase]) - return .none - - case .awaitingDemandFromConsumer(let task, var suspendedBases): - assert(suspendedBases.count < 2, "There cannot be more than 2 suspended base at the same time") - suspendedBases.append(suspendedBase) - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: suspendedBases) - return .none - - case .awaitingBaseResults(let task, let result1, let result2, var suspendedBases, let suspendedDemand): - assert(suspendedBases.count < 2, "There cannot be more than 2 suspended bases at the same time") - if result2 != nil { - suspendedBases.append(suspendedBase) - self.state = .awaitingBaseResults(task: task, result1: result1, result2: result2, suspendedBases: suspendedBases, suspendedDemand: suspendedDemand) - return .none - } else { - self.state = .awaitingBaseResults(task: task, result1: result1, result2: result2, suspendedBases: suspendedBases, suspendedDemand: suspendedDemand) - return .resumeBases(suspendedBases: [suspendedBase]) - } - - case .finished: - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - } - } - - enum BaseHasProducedElementOutput { - case none - case resumeDemand( - suspendedDemand: UnsafeContinuation<(Result, Result)?, Never>?, - result1: Result, - result2: Result - ) - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]? - ) - } - - mutating func base1HasProducedElement(element: Element1) -> BaseHasProducedElementOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil) - - case .started(let task): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, _, let result2, let suspendedBases, let suspendedDemand): - if let result2 = result2 { - self.state = .awaitingBaseResults(task: task, result1: .success(element), result2: result2, suspendedBases: suspendedBases, suspendedDemand: nil) - return .resumeDemand(suspendedDemand: suspendedDemand, result1: .success(element), result2: result2) - } else { - self.state = .awaitingBaseResults(task: task, result1: .success(element), result2: nil, suspendedBases: suspendedBases, suspendedDemand: suspendedDemand) - return .none - } - - case .finished: - return .terminate(task: nil, suspendedBases: nil) - } - } - - enum BaseHasProducedFailureOutput { - case resumeDemandAndTerminate( - task: Task?, - suspendedDemand: UnsafeContinuation<(Result, Result)?, Never>?, - suspendedBases: [UnsafeContinuation], - result1: Result, - result2: Result - ) - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]? - ) - } - - mutating func baseHasProducedFailure(error: any Error) -> BaseHasProducedFailureOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil) - - case .started(let task): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, _, _, let suspendedBases, let suspendedDemand): - self.state = .finished - return .resumeDemandAndTerminate( - task: task, - suspendedDemand: suspendedDemand, - suspendedBases: suspendedBases, - result1: .failure(error), - result2: .failure(error) - ) - - case .finished: - return .terminate(task: nil, suspendedBases: nil) - } - } - - mutating func base2HasProducedElement(element: Element2) -> BaseHasProducedElementOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil) - - case .started(let task): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, let result1, _, let suspendedBases, let suspendedDemand): - if let result1 = result1 { - self.state = .awaitingBaseResults(task: task, result1: result1, result2: .success(element), suspendedBases: suspendedBases, suspendedDemand: nil) - return .resumeDemand(suspendedDemand: suspendedDemand, result1: result1, result2: .success(element)) - } else { - self.state = .awaitingBaseResults(task: task, result1: nil, result2: .success(element), suspendedBases: suspendedBases, suspendedDemand: suspendedDemand) - return .none - } - - case .finished: - return .terminate(task: nil, suspendedBases: nil) - } - } - - enum DemandIsFulfilledOutput { - case none - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]?, - suspendedDemands: [UnsafeContinuation<(Result, Result)?, Never>]? - ) - } - - mutating func demandIsFulfilled() -> DemandIsFulfilledOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - - case .started(let task): - assertionFailure("Inconsistent state, results are not yet available to be acknowledged") - self.state = .finished - return .terminate(task: task, suspendedBases: nil, suspendedDemands: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, results are not yet available to be acknowledged") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: nil) - - case .awaitingBaseResults(let task, let result1, let result2, let suspendedBases, let suspendedDemand): - assert(suspendedDemand == nil, "Inconsistent state, there cannot be a suspended demand when ackowledging the demand") - assert(result1 != nil && result2 != nil, "Inconsistent state, all results are not yet available to be acknowledged") - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: suspendedBases) - return .none - - case .finished: - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - } - } - - enum RootTaskIsCancelledOutput { - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]?, - suspendedDemands: [UnsafeContinuation<(Result, Result)?, Never>?]? - ) - } - - mutating func rootTaskIsCancelled() -> RootTaskIsCancelledOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - - case .started(let task): - self.state = .finished - return .terminate(task: task, suspendedBases: nil, suspendedDemands: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: nil) - - case .awaitingBaseResults(let task, _, _, let suspendedBases, let suspendedDemand): - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: [suspendedDemand]) - - case .finished: - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - } - } - - enum BaseIsFinishedOutput { - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]?, - suspendedDemands: [UnsafeContinuation<(Result, Result)?, Never>?]? - ) - } - - mutating func baseIsFinished() -> BaseIsFinishedOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - - case .started(let task): - self.state = .finished - return .terminate(task: task, suspendedBases: nil, suspendedDemands: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: nil) - - case .awaitingBaseResults(let task, _, _, let suspendedBases, let suspendedDemand): - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: [suspendedDemand]) - - case .finished: - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - } - } -} diff --git a/Sources/Combiners/Zip/Zip3Runtime.swift b/Sources/Combiners/Zip/Zip3Runtime.swift deleted file mode 100644 index 9b8cc18..0000000 --- a/Sources/Combiners/Zip/Zip3Runtime.swift +++ /dev/null @@ -1,252 +0,0 @@ -// -// Zip3Runtime.swift -// -// -// Created by Thibault Wittemberg on 24/09/2022. -// - -final class Zip3Runtime: Sendable -where Base1: Sendable, Base1.Element: Sendable, Base2: Sendable, Base2.Element: Sendable, Base3: Sendable, Base3.Element: Sendable { - typealias ZipStateMachine = Zip3StateMachine - - private let stateMachine = ManagedCriticalState(ZipStateMachine()) - - init(_ base1: Base1, _ base2: Base2, _ base3: Base3) { - self.stateMachine.withCriticalRegion { machine in - machine.taskIsStarted(task: Task { - await withTaskGroup(of: Void.self) { group in - group.addTask { - var base1Iterator = base1.makeAsyncIterator() - - do { - while true { - await withUnsafeContinuation { (continuation: UnsafeContinuation) in - let output = self.stateMachine.withCriticalRegion { machine in - machine.newLoopFromBase1(suspendedBase: continuation) - } - - self.handle(newLoopFromBaseOutput: output) - } - - guard let element1 = try await base1Iterator.next() else { - break - } - - let output = self.stateMachine.withCriticalRegion { machine in - machine.base1HasProducedElement(element: element1) - } - - self.handle(baseHasProducedElementOutput: output) - } - } catch { - let output = self.stateMachine.withCriticalRegion { machine in - machine.baseHasProducedFailure(error: error) - } - - self.handle(baseHasProducedFailureOutput: output) - } - - let output = self.stateMachine.withCriticalRegion { stateMachine in - stateMachine.baseIsFinished() - } - - self.handle(baseIsFinishedOutput: output) - } - - group.addTask { - var base2Iterator = base2.makeAsyncIterator() - - do { - while true { - await withUnsafeContinuation { (continuation: UnsafeContinuation) in - let output = self.stateMachine.withCriticalRegion { machine in - machine.newLoopFromBase2(suspendedBase: continuation) - } - - self.handle(newLoopFromBaseOutput: output) - } - - guard let element2 = try await base2Iterator.next() else { - break - } - - let output = self.stateMachine.withCriticalRegion { machine in - machine.base2HasProducedElement(element: element2) - } - - self.handle(baseHasProducedElementOutput: output) - } - } catch { - let output = self.stateMachine.withCriticalRegion { machine in - machine.baseHasProducedFailure(error: error) - } - - self.handle(baseHasProducedFailureOutput: output) - } - - let output = self.stateMachine.withCriticalRegion { machine in - machine.baseIsFinished() - } - - self.handle(baseIsFinishedOutput: output) - } - - group.addTask { - var base3Iterator = base3.makeAsyncIterator() - - do { - while true { - await withUnsafeContinuation { (continuation: UnsafeContinuation) in - let output = self.stateMachine.withCriticalRegion { machine in - machine.newLoopFromBase3(suspendedBase: continuation) - } - - self.handle(newLoopFromBaseOutput: output) - } - - guard let element3 = try await base3Iterator.next() else { - break - } - - let output = self.stateMachine.withCriticalRegion { machine in - machine.base3HasProducedElement(element: element3) - } - - self.handle(baseHasProducedElementOutput: output) - } - } catch { - let output = self.stateMachine.withCriticalRegion { machine in - machine.baseHasProducedFailure(error: error) - } - - self.handle(baseHasProducedFailureOutput: output) - } - - let output = self.stateMachine.withCriticalRegion { machine in - machine.baseIsFinished() - } - - self.handle(baseIsFinishedOutput: output) - } - } - }) - } - } - - private func handle(newLoopFromBaseOutput: ZipStateMachine.NewLoopFromBaseOutput) { - switch newLoopFromBaseOutput { - case .none: - break - - case .resumeBases(let suspendedBases): - suspendedBases.forEach { $0.resume() } - - case .terminate(let task, let suspendedBase, let suspendedDemand): - suspendedBase.resume() - suspendedDemand?.resume(returning: nil) - task?.cancel() - } - } - - private func handle(baseHasProducedElementOutput: ZipStateMachine.BaseHasProducedElementOutput) { - switch baseHasProducedElementOutput { - case .none: - break - - case .resumeDemand(let suspendedDemand, let result1, let result2, let result3): - suspendedDemand?.resume(returning: (result1, result2, result3)) - - case .terminate(let task, let suspendedBases): - suspendedBases?.forEach { $0.resume() } - task?.cancel() - } - } - - private func handle(baseHasProducedFailureOutput: ZipStateMachine.BaseHasProducedFailureOutput) { - switch baseHasProducedFailureOutput { - case .resumeDemandAndTerminate(let task, let suspendedDemand, let suspendedBases, let result1, let result2, let result3): - suspendedDemand?.resume(returning: (result1, result2, result3)) - suspendedBases.forEach { $0.resume() } - task?.cancel() - - case .terminate(let task, let suspendedBases): - suspendedBases?.forEach { $0.resume() } - task?.cancel() - } - } - - private func handle(baseIsFinishedOutput: ZipStateMachine.BaseIsFinishedOutput) { - switch baseIsFinishedOutput { - case .terminate(let task, let suspendedBases, let suspendedDemands): - suspendedBases?.forEach { $0.resume() } - suspendedDemands?.forEach { $0?.resume(returning: nil) } - task?.cancel() - } - } - - func next() async rethrows -> (Base1.Element, Base2.Element, Base3.Element)? { - try await withTaskCancellationHandler { - let results = await withUnsafeContinuation { (continuation: UnsafeContinuation<(Result, Result, Result)?, Never>) in - let output = self.stateMachine.withCriticalRegion { stateMachine in - stateMachine.newDemandFromConsumer(suspendedDemand: continuation) - } - - self.handle(newDemandFromConsumerOutput: output) - } - - guard let results = results else { - return nil - } - - let output = self.stateMachine.withCriticalRegion { stateMachine in - stateMachine.demandIsFulfilled() - } - - self.handle(demandIsFulfilledOutput: output) - - return try (results.0._rethrowGet(), results.1._rethrowGet(), results.2._rethrowGet()) - } onCancel: { - let output = self.stateMachine.withCriticalRegion { stateMachine in - stateMachine.rootTaskIsCancelled() - } - - self.handle(rootTaskIsCancelledOutput: output) - } - } - - private func handle(rootTaskIsCancelledOutput: ZipStateMachine.RootTaskIsCancelledOutput) { - switch rootTaskIsCancelledOutput { - case .terminate(let task, let suspendedBases, let suspendedDemands): - suspendedBases?.forEach { $0.resume() } - suspendedDemands?.forEach { $0?.resume(returning: nil) } - task?.cancel() - } - } - - private func handle(newDemandFromConsumerOutput: ZipStateMachine.NewDemandFromConsumerOutput) { - switch newDemandFromConsumerOutput { - case .none: - break - - case .resumeBases(let suspendedBases): - suspendedBases.forEach { $0.resume() } - - case .terminate(let task, let suspendedBases, let suspendedDemands): - suspendedBases?.forEach { $0.resume() } - suspendedDemands?.forEach { $0?.resume(returning: nil) } - task?.cancel() - } - } - - private func handle(demandIsFulfilledOutput: ZipStateMachine.DemandIsFulfilledOutput) { - switch demandIsFulfilledOutput { - case .none: - break - - case .terminate(let task, let suspendedBases, let suspendedDemands): - suspendedBases?.forEach { $0.resume() } - suspendedDemands?.forEach { $0.resume(returning: nil) } - task?.cancel() - } - } -} diff --git a/Sources/Combiners/Zip/Zip3StateMachine.swift b/Sources/Combiners/Zip/Zip3StateMachine.swift deleted file mode 100644 index 3d3ed82..0000000 --- a/Sources/Combiners/Zip/Zip3StateMachine.swift +++ /dev/null @@ -1,542 +0,0 @@ -// -// Zip3StateMachine.swift -// -// -// Created by Thibault Wittemberg on 24/09/2022. -// - -struct Zip3StateMachine: Sendable -where Element1: Sendable, Element2: Sendable, Element3: Sendable { - private enum State { - case initial - case started(task: Task) - case awaitingDemandFromConsumer( - task: Task?, - suspendedBases: [UnsafeContinuation] - ) - case awaitingBaseResults( - task: Task?, - result1: Result?, - result2: Result?, - result3: Result?, - suspendedBases: [UnsafeContinuation], - suspendedDemand: UnsafeContinuation<(Result, Result, Result)?, Never>? - ) - case finished - } - - private var state: State = .initial - - mutating func taskIsStarted(task: Task) { - switch self.state { - case .initial: - self.state = .started(task: task) - - default: - assertionFailure("Inconsistent state, the task cannot start while the state is other than initial") - } - } - - enum NewDemandFromConsumerOutput { - case none - case resumeBases(suspendedBases: [UnsafeContinuation]) - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]?, - suspendedDemands: [UnsafeContinuation<(Result, Result, Result)?, Never>?]? - ) - } - - mutating func newDemandFromConsumer( - suspendedDemand: UnsafeContinuation<(Result, Result, Result)?, Never> - ) -> NewDemandFromConsumerOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: [suspendedDemand]) - - case .started(let task): - self.state = .awaitingBaseResults( - task: task, - result1: nil, - result2: nil, - result3: nil, - suspendedBases: [], - suspendedDemand: suspendedDemand - ) - return .none - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - self.state = .awaitingBaseResults( - task: task, - result1: nil, - result2: nil, - result3: nil, - suspendedBases: [], - suspendedDemand: suspendedDemand - ) - return .resumeBases(suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, _, _, _, let suspendedBases, let oldSuspendedDemand): - assertionFailure("Inconsistent state, a demand is already suspended") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: [oldSuspendedDemand, suspendedDemand]) - - case .finished: - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: [suspendedDemand]) - } - } - - enum NewLoopFromBaseOutput { - case none - case resumeBases(suspendedBases: [UnsafeContinuation]) - case terminate( - task: Task?, - suspendedBase: UnsafeContinuation, - suspendedDemand: UnsafeContinuation<(Result, Result, Result)?, Never>? - ) - } - - mutating func newLoopFromBase1(suspendedBase: UnsafeContinuation) -> NewLoopFromBaseOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - - case .started(let task): - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: [suspendedBase]) - return .none - - case .awaitingDemandFromConsumer(let task, var suspendedBases): - assert(suspendedBases.count < 3, "There cannot be more than 3 suspended base at the same time") - suspendedBases.append(suspendedBase) - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: suspendedBases) - return .none - - case .awaitingBaseResults(let task, let result1, let result2, let result3, var suspendedBases, let suspendedDemand): - assert(suspendedBases.count < 3, "There cannot be more than 3 suspended bases at the same time") - if result1 != nil { - suspendedBases.append(suspendedBase) - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: result2, - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .none - } else { - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: result2, - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .resumeBases(suspendedBases: [suspendedBase]) - } - - case .finished: - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - } - } - - mutating func newLoopFromBase2(suspendedBase: UnsafeContinuation) -> NewLoopFromBaseOutput { - switch state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - - case .started(let task): - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: [suspendedBase]) - return .none - - case .awaitingDemandFromConsumer(let task, var suspendedBases): - assert(suspendedBases.count < 3, "There cannot be more than 3 suspended base at the same time") - suspendedBases.append(suspendedBase) - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: suspendedBases) - return .none - - case .awaitingBaseResults(let task, let result1, let result2, let result3, var suspendedBases, let suspendedDemand): - assert(suspendedBases.count < 3, "There cannot be more than 3 suspended bases at the same time") - if result2 != nil { - suspendedBases.append(suspendedBase) - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: result2, - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .none - } else { - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: result2, - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .resumeBases(suspendedBases: [suspendedBase]) - } - - case .finished: - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - } - } - - mutating func newLoopFromBase3(suspendedBase: UnsafeContinuation) -> NewLoopFromBaseOutput { - switch state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - - case .started(let task): - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: [suspendedBase]) - return .none - - case .awaitingDemandFromConsumer(let task, var suspendedBases): - assert(suspendedBases.count < 3, "There cannot be more than 3 suspended base at the same time") - suspendedBases.append(suspendedBase) - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: suspendedBases) - return .none - - case .awaitingBaseResults(let task, let result1, let result2, let result3, var suspendedBases, let suspendedDemand): - assert(suspendedBases.count < 3, "There cannot be more than 3 suspended bases at the same time") - if result3 != nil { - suspendedBases.append(suspendedBase) - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: result2, - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .none - } else { - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: result2, - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .resumeBases(suspendedBases: [suspendedBase]) - } - - case .finished: - return .terminate(task: nil, suspendedBase: suspendedBase, suspendedDemand: nil) - } - } - - enum BaseHasProducedElementOutput { - case none - case resumeDemand( - suspendedDemand: UnsafeContinuation<(Result, Result, Result)?, Never>?, - result1: Result, - result2: Result, - result3: Result - ) - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]? - ) - } - - mutating func base1HasProducedElement(element: Element1) -> BaseHasProducedElementOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil) - - case .started(let task): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, _, let result2, let result3, let suspendedBases, let suspendedDemand): - if let result2 = result2, let result3 = result3 { - self.state = .awaitingBaseResults( - task: task, - result1: .success(element), - result2: result2, - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: nil - ) - return .resumeDemand(suspendedDemand: suspendedDemand, result1: .success(element), result2: result2, result3: result3) - } else { - self.state = .awaitingBaseResults( - task: task, - result1: .success(element), - result2: result2, - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .none - } - - case .finished: - return .terminate(task: nil, suspendedBases: nil) - } - } - - enum BaseHasProducedFailureOutput { - case resumeDemandAndTerminate( - task: Task?, - suspendedDemand: UnsafeContinuation<(Result, Result, Result)?, Never>?, - suspendedBases: [UnsafeContinuation], - result1: Result, - result2: Result, - result3: Result - ) - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]? - ) - } - - mutating func base2HasProducedElement(element: Element2) -> BaseHasProducedElementOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil) - - case .started(let task): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, let result1, _, let result3, let suspendedBases, let suspendedDemand): - if let result1 = result1, let result3 = result3 { - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: .success(element), - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: nil - ) - return .resumeDemand(suspendedDemand: suspendedDemand, result1: result1, result2: .success(element), result3: result3) - } else { - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: .success(element), - result3: result3, - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .none - } - - case .finished: - return .terminate(task: nil, suspendedBases: nil) - } - } - - mutating func base3HasProducedElement(element: Element3) -> BaseHasProducedElementOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil) - - case .started(let task): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, let result1, let result2, _, let suspendedBases, let suspendedDemand): - if let result1 = result1, let result2 = result2 { - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: result2, - result3: .success(element), - suspendedBases: suspendedBases, - suspendedDemand: nil - ) - return .resumeDemand(suspendedDemand: suspendedDemand, result1: result1, result2: result2, result3: .success(element)) - } else { - self.state = .awaitingBaseResults( - task: task, - result1: result1, - result2: result2, - result3: .success(element), - suspendedBases: suspendedBases, - suspendedDemand: suspendedDemand - ) - return .none - } - - case .finished: - return .terminate(task: nil, suspendedBases: nil) - } - } - - mutating func baseHasProducedFailure(error: Error) -> BaseHasProducedFailureOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil) - - case .started(let task): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, a base can only produce an element when the consumer is awaiting for it") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases) - - case .awaitingBaseResults(let task, _, _, _, let suspendedBases, let suspendedDemand): - self.state = .finished - return .resumeDemandAndTerminate( - task: task, - suspendedDemand: suspendedDemand, - suspendedBases: suspendedBases, - result1: .failure(error), - result2: .failure(error), - result3: .failure(error) - ) - - case .finished: - return .terminate(task: nil, suspendedBases: nil) - } - } - - enum DemandIsFulfilledOutput { - case none - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]?, - suspendedDemands: [UnsafeContinuation<(Result, Result)?, Never>]? - ) - } - - mutating func demandIsFulfilled() -> DemandIsFulfilledOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - - case .started(let task): - assertionFailure("Inconsistent state, results are not yet available to be acknowledged") - self.state = .finished - return .terminate(task: task, suspendedBases: nil, suspendedDemands: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - assertionFailure("Inconsistent state, results are not yet available to be acknowledged") - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: nil) - - case .awaitingBaseResults(let task, let result1, let result2, let result3, let suspendedBases, let suspendedDemand): - assert(suspendedDemand == nil, "Inconsistent state, there cannot be a suspended demand when ackowledging the demand") - assert( - result1 != nil && result2 != nil && result3 != nil, - "Inconsistent state, all results are not yet available to be acknowledged" - ) - self.state = .awaitingDemandFromConsumer(task: task, suspendedBases: suspendedBases) - return .none - - case .finished: - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - } - } - - enum RootTaskIsCancelledOutput { - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]?, - suspendedDemands: [UnsafeContinuation<(Result, Result, Result)?, Never>?]? - ) - } - - mutating func rootTaskIsCancelled() -> RootTaskIsCancelledOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - - case .started(let task): - self.state = .finished - return .terminate(task: task, suspendedBases: nil, suspendedDemands: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: nil) - - case .awaitingBaseResults(let task, _, _, _, let suspendedBases, let suspendedDemand): - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: [suspendedDemand]) - - case .finished: - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - } - } - - enum BaseIsFinishedOutput { - case terminate( - task: Task?, - suspendedBases: [UnsafeContinuation]?, - suspendedDemands: [UnsafeContinuation<(Result, Result, Result)?, Never>?]? - ) - } - - mutating func baseIsFinished() -> BaseIsFinishedOutput { - switch self.state { - case .initial: - assertionFailure("Inconsistent state, the task is not started") - self.state = .finished - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - - case .started(let task): - self.state = .finished - return .terminate(task: task, suspendedBases: nil, suspendedDemands: nil) - - case .awaitingDemandFromConsumer(let task, let suspendedBases): - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: nil) - - case .awaitingBaseResults(let task, _, _, _, let suspendedBases, let suspendedDemand): - self.state = .finished - return .terminate(task: task, suspendedBases: suspendedBases, suspendedDemands: [suspendedDemand]) - - case .finished: - return .terminate(task: nil, suspendedBases: nil, suspendedDemands: nil) - } - } -} diff --git a/Tests/Operators/AsyncPrependSequenceTests.swift b/Tests/Operators/AsyncPrependSequenceTests.swift index c23c03d..ec0e606 100644 --- a/Tests/Operators/AsyncPrependSequenceTests.swift +++ b/Tests/Operators/AsyncPrependSequenceTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 01/01/2022. // +import AsyncAlgorithms import AsyncExtensions import XCTest diff --git a/Tests/Operators/AsyncSequence+AssignTests.swift b/Tests/Operators/AsyncSequence+AssignTests.swift index d738d14..03688f2 100644 --- a/Tests/Operators/AsyncSequence+AssignTests.swift +++ b/Tests/Operators/AsyncSequence+AssignTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 02/02/2022. // +import AsyncAlgorithms import AsyncExtensions import XCTest @@ -21,7 +22,7 @@ private class Root { final class AsyncSequence_AssignTests: XCTestCase { func testAssign_sets_elements_on_the_root() async throws { let root = Root() - let sut = AsyncLazySequence(["1", "2", "3"]) + let sut = ["1", "2", "3"].async try await sut.assign(to: \.property, on: root) XCTAssertEqual(root.successiveValues, ["1", "2", "3"]) } diff --git a/Tests/Operators/AsyncSequence+FlatMapLatestTests.swift b/Tests/Operators/AsyncSequence+FlatMapLatestTests.swift index c84a637..744674d 100644 --- a/Tests/Operators/AsyncSequence+FlatMapLatestTests.swift +++ b/Tests/Operators/AsyncSequence+FlatMapLatestTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 10/01/2022. // +import AsyncAlgorithms @testable import AsyncExtensions import XCTest @@ -166,10 +167,10 @@ final class AsyncSequence_FlatMapLatestTests: XCTestCase { func testFlatMapLatest_propagates_errors() async { let expectedError = MockError(code: Int.random(in: 0...100)) - let sut = AsyncLazySequence([1, 2]) + let sut = [1, 2].async .flatMapLatest { element -> AnyAsyncSequence in if element == 1 { - return AsyncLazySequence([1]).eraseToAnyAsyncSequence() + return [1].async.eraseToAnyAsyncSequence() } return AsyncFailSequence(expectedError).eraseToAnyAsyncSequence() diff --git a/Tests/Operators/AsyncSwitchToLatestSequenceTests.swift b/Tests/Operators/AsyncSwitchToLatestSequenceTests.swift index 8619cde..79bf5d1 100644 --- a/Tests/Operators/AsyncSwitchToLatestSequenceTests.swift +++ b/Tests/Operators/AsyncSwitchToLatestSequenceTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 04/01/2022. // +import AsyncAlgorithms @testable import AsyncExtensions import XCTest @@ -102,10 +103,10 @@ final class AsyncSwitchToLatestSequenceTests: XCTestCase { func testSwitchToLatest_propagates_errors_when_base_sequence_fails() async { let sequences = [ - AsyncLazySequence([1, 2, 3]).eraseToAnyAsyncSequence(), - AsyncLazySequence([4, 5, 6]).eraseToAnyAsyncSequence(), - AsyncLazySequence([7, 8, 9]).eraseToAnyAsyncSequence(), // should fail here - AsyncLazySequence([10, 11, 12]).eraseToAnyAsyncSequence(), + [1, 2, 3].async.eraseToAnyAsyncSequence(), + [4, 5, 6].async.eraseToAnyAsyncSequence(), + [7, 8, 9].async.eraseToAnyAsyncSequence(), // should fail here + [10, 11, 12].async.eraseToAnyAsyncSequence(), ] let sourceSequence = LongAsyncSequence(elements: sequences, interval: .milliseconds(100), failAt: 2) From 431569aec2f38b9b73326113e07e409583915982 Mon Sep 17 00:00:00 2001 From: harry lachenmayer Date: Wed, 6 Dec 2023 17:45:02 +0000 Subject: [PATCH 2/4] docs: remove sentence about overlaps in README --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index f20fbd7..b4110ff 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ **AsyncExtensions** provides a collection of operators that intends to ease the creation and combination of `AsyncSequences`. -**AsyncExtensions** can be seen as a companion to Apple [swift-async-algorithms](https://github.com/apple/swift-async-algorithms). For now there is an overlap between both libraries, but when **swift-async-algorithms** becomes stable the overlapping operators while be deprecated in **AsyncExtensions**. Nevertheless **AsyncExtensions** will continue to provide the operators that the community needs and are not provided by Apple. +**AsyncExtensions** can be seen as a companion to Apple [swift-async-algorithms](https://github.com/apple/swift-async-algorithms), which provides operators that the community needs and are not provided by Apple. ## Adding AsyncExtensions as a Dependency From b015aa7fef91b4101078f949d21e58a3e4ebaba8 Mon Sep 17 00:00:00 2001 From: Thibault Wittemberg Date: Sat, 3 Oct 2026 12:40:34 +0200 Subject: [PATCH 3/4] Complete Swift Async Algorithms coexistence migration Preserve variadic zip and merge and the explicit AsyncLazySequence constructor. Remove the conflicting Sequence.async extension and unused fixed-arity merge setup, repair test imports, and cover both modules together. --- CHANGELOG.md | 3 + Package.swift | 2 +- README.md | 43 ++- .../Combiners/Merge/MergeStateMachine.swift | 61 ---- Sources/Creators/AsyncLazySequence.swift | 8 - .../AsyncSubjectCancellationTests.swift | 1 + .../AsyncSubjectQueuedDeliveryTests.swift | 1 + .../AsyncAlgorithmsCompatibilityTests.swift | 70 ++++ .../Merge/AsyncMergeSequenceTests.swift | 71 +++-- .../AsyncWithLatestFrom2SequenceTests.swift | 1 + .../AsyncWithLatestFromSequenceTests.swift | 1 + .../Combiners/Zip/AsyncZipSequenceTests.swift | 301 ------------------ .../AsyncHandleEventsSequenceTests.swift | 1 + Tests/Operators/AsyncScanSequenceTests.swift | 1 + ...equence+EraseToAnyAsyncSequenceTests.swift | 1 + ...AsyncSwitchToLatestCancellationTests.swift | 1 + 16 files changed, 164 insertions(+), 403 deletions(-) create mode 100644 Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bfe0e4..6e750d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,8 @@ **Unreleased:** +- Breaking: use Swift Async Algorithms for `Sequence.async` and two- or three-input `zip`/`merge`, removing import ambiguities. Add the `AsyncAlgorithms` product dependency and import when migrating these APIs. Preserve variadic `zip`/`merge` and the explicit `AsyncLazySequence` constructor. +- SwiftPM: require Swift 5.8 or later for the Swift Async Algorithms test dependency. + - SwitchToLatest: finish cancelled collection while the latest channel or outer sequence remains open, and discard late producer results (https://github.com/sideeffect-io/AsyncExtensions/issues/53). - Subjects: fix a deadlock when sending values or termination concurrently with consumer cancellation (https://github.com/sideeffect-io/AsyncExtensions/issues/52). - Subjects: preserve a shared order for concurrent sends, keep current-value/replay delivery consistent with stored state, and queue termination after accepted values without locking during delivery (https://github.com/sideeffect-io/AsyncExtensions/issues/61). Concurrent sends may return while another sender drains their queued delivery. diff --git a/Package.swift b/Package.swift index cb582c4..5252a8b 100644 --- a/Package.swift +++ b/Package.swift @@ -1,4 +1,4 @@ -// swift-tools-version:5.5 +// swift-tools-version:5.8 // The swift-tools-version declares the minimum version of Swift required to build this package. import PackageDescription diff --git a/README.md b/README.md index b4110ff..70c2512 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ **AsyncExtensions** provides a collection of operators that intends to ease the creation and combination of `AsyncSequences`. -**AsyncExtensions** can be seen as a companion to Apple [swift-async-algorithms](https://github.com/apple/swift-async-algorithms), which provides operators that the community needs and are not provided by Apple. +**AsyncExtensions** complements Apple [swift-async-algorithms](https://github.com/apple/swift-async-algorithms) with subjects, buffered channels, and operators that Apple does not provide. ## Adding AsyncExtensions as a Dependency @@ -28,6 +28,44 @@ Include `"AsyncExtensions"` as a dependency for your executable target: Finally, add `import AsyncExtensions` to your source code. +## Using Swift Async Algorithms alongside AsyncExtensions + +SwiftPM requires Swift 5.8 or later. To use both libraries, declare both package dependencies and add the `AsyncAlgorithms` product to your target alongside `AsyncExtensions`: + +```swift +.package(url: "https://github.com/apple/swift-async-algorithms.git", from: "1.0.0"), +``` + +```swift +.target( + name: "", + dependencies: [ + "AsyncExtensions", + .product(name: "AsyncAlgorithms", package: "swift-async-algorithms") + ] +), +``` + +Import both modules in each source file that uses them: + +```swift +import AsyncAlgorithms +import AsyncExtensions + +let values = [1, 2, 3].async +let pairs = zip(values, ["a", "b", "c"].async) +let merged = merge(values, [4, 5, 6].async) +let rows = zip(values, values, values, values) +``` + +`Sequence.async`, two- and three-input `zip`/`merge`, and their corresponding sequence types now come from `AsyncAlgorithms`. This is a breaking API change: add that dependency and import when migrating these calls. + +AsyncExtensions retains variadic `zip` and `merge`, including support for four or more inputs. Use `AsyncExtensions.zip(...)` or `AsyncExtensions.merge(...)` explicitly when you want the variadic implementations with two or three inputs; variadic `zip` produces arrays, while Apple's fixed overloads produce tuples. + +Variadic inputs share one concrete sequence type. Use `eraseToAnyAsyncSequence()` when combining different sequence types through these variadic operators. + +The `AsyncLazySequence(sequence)` constructor remains available when you only need AsyncExtensions. The `.async` extension is supplied exclusively by AsyncAlgorithms. + ## Features ### Channels @@ -55,6 +93,8 @@ the value. The active sender drains pending deliveries before returning. A new c or replay consumer receives the latest stored state, followed by subsequent sends. ### Combiners +* [`zip(_:)`](./Sources/Combiners/Zip/AsyncZipSequence.swift): Zips any number of async sequences into arrays of elements +* [`merge(_:)`](./Sources/Combiners/Merge/AsyncMergeSequence.swift): Merges any number of async sequences into one sequence * [`withLatest(_:)`](./Sources/Combiners/WithLatestFrom/AsyncWithLatestFromSequence.swift): Combines elements from self with the last known element from an other `AsyncSequence` * [`withLatest(_:_:)`](./Sources/Combiners/WithLatestFrom/AsyncWithLatestFrom2Sequence.swift): Combines elements from self with the last known elements from two other async sequences @@ -63,6 +103,7 @@ or replay consumer receives the latest stored state, followed by subsequent send * [AsyncFailSequence](./Sources/Creators/AsyncFailSequence.swift): Creates an `AsyncSequence` that immediately fails * [AsyncJustSequence](./Sources/Creators/AsyncJustSequence.swift): Creates an `AsyncSequence` that emits an element an finishes * [AsyncThrowingJustSequence](./Sources/Creators/AsyncThrowingJustSequence.swift): Creates an `AsyncSequence` that emits an elements and finishes bases on a throwing closure +* [AsyncLazySequence](./Sources/Creators/AsyncLazySequence.swift): Creates an async sequence from an explicit synchronous sequence * [AsyncTimerSequence](./Sources/Creators/AsyncTimerSequence.swift): Creates an `AsyncSequence` that emits a date value periodically * [AsyncStream Pipe](./Sources/Creators/AsyncStream+Pipe.swift): Creates an AsyncStream and returns a tuple standing for its inputs and outputs diff --git a/Sources/Combiners/Merge/MergeStateMachine.swift b/Sources/Combiners/Merge/MergeStateMachine.swift index 2f7984a..e8e9027 100644 --- a/Sources/Combiners/Merge/MergeStateMachine.swift +++ b/Sources/Combiners/Merge/MergeStateMachine.swift @@ -29,67 +29,6 @@ struct MergeStateMachine: Sendable { let state: ManagedCriticalState let task: Task - init( - _ base1: Base1, - _ base2: Base2 - ) where Base1.Element == Element, Base2.Element == Element { - self.state = ManagedCriticalState(State(buffer: .idle, basesToTerminate: 2)) - - let regulator1 = Regulator(base1, onNextRegulatedElement: { [state] in Self.onNextRegulatedElement($0, state: state) }) - let regulator2 = Regulator(base2, onNextRegulatedElement: { [state] in Self.onNextRegulatedElement($0, state: state) }) - - self.requestNextRegulatedElements = { - regulator1.requestNextRegulatedElement() - regulator2.requestNextRegulatedElement() - } - - self.task = Task { - await withTaskGroup(of: Void.self) { group in - group.addTask { - await regulator1.iterate() - } - - group.addTask { - await regulator2.iterate() - } - } - } - } - - init( - _ base1: Base1, - _ base2: Base2, - _ base3: Base3 - ) where Base1.Element == Element, Base2.Element == Element, Base3.Element == Base1.Element { - self.state = ManagedCriticalState(State(buffer: .idle, basesToTerminate: 3)) - - let regulator1 = Regulator(base1, onNextRegulatedElement: { [state] in Self.onNextRegulatedElement($0, state: state) }) - let regulator2 = Regulator(base2, onNextRegulatedElement: { [state] in Self.onNextRegulatedElement($0, state: state) }) - let regulator3 = Regulator(base3, onNextRegulatedElement: { [state] in Self.onNextRegulatedElement($0, state: state) }) - - self.requestNextRegulatedElements = { - regulator1.requestNextRegulatedElement() - regulator2.requestNextRegulatedElement() - regulator3.requestNextRegulatedElement() - } - - self.task = Task { - await withTaskGroup(of: Void.self) { group in - group.addTask { - await regulator1.iterate() - } - - group.addTask { - await regulator2.iterate() - } - - group.addTask { - await regulator3.iterate() - } - } - } - } - init( _ bases: [Base] ) where Base.Element == Element { diff --git a/Sources/Creators/AsyncLazySequence.swift b/Sources/Creators/AsyncLazySequence.swift index b68d998..984f27e 100644 --- a/Sources/Creators/AsyncLazySequence.swift +++ b/Sources/Creators/AsyncLazySequence.swift @@ -5,14 +5,6 @@ // Created by Thibault Wittemberg on 01/01/2022. // -public extension Sequence { - /// Creates an AsyncSequence of the sequence elements. - /// - Returns: The AsyncSequence that outputs the elements from the sequence. - var async: AsyncLazySequence { - AsyncLazySequence(self) - } -} - /// `AsyncLazySequence` is an AsyncSequence that outputs elements from a traditional Sequence. /// If the parent task is cancelled while iterating then the iteration finishes. /// diff --git a/Tests/AsyncSubjets/AsyncSubjectCancellationTests.swift b/Tests/AsyncSubjets/AsyncSubjectCancellationTests.swift index a5676f3..fe9ec28 100644 --- a/Tests/AsyncSubjets/AsyncSubjectCancellationTests.swift +++ b/Tests/AsyncSubjets/AsyncSubjectCancellationTests.swift @@ -1,4 +1,5 @@ import Dispatch +import AsyncAlgorithms @testable import AsyncExtensions import XCTest diff --git a/Tests/AsyncSubjets/AsyncSubjectQueuedDeliveryTests.swift b/Tests/AsyncSubjets/AsyncSubjectQueuedDeliveryTests.swift index 073165d..769379e 100644 --- a/Tests/AsyncSubjets/AsyncSubjectQueuedDeliveryTests.swift +++ b/Tests/AsyncSubjets/AsyncSubjectQueuedDeliveryTests.swift @@ -1,4 +1,5 @@ import Dispatch +import AsyncAlgorithms @testable import AsyncExtensions import XCTest diff --git a/Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift b/Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift new file mode 100644 index 0000000..ea7478c --- /dev/null +++ b/Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift @@ -0,0 +1,70 @@ +import AsyncAlgorithms +import AsyncExtensions +import XCTest + +final class AsyncAlgorithmsCompatibilityTests: XCTestCase { + func test_async_extension_comes_from_async_algorithms() async { + let sequence: AsyncSyncSequence<[Int]> = [1, 2, 3].async + var received = [Int]() + for await element in sequence { + received.append(element) + } + XCTAssertEqual(received, [1, 2, 3]) + } + + func test_two_input_zip_comes_from_async_algorithms() async { + let sequence = zip([1, 2].async, ["a", "b"].async) + var received = [Tuple2]() + for await element in sequence { + received.append(Tuple2(element)) + } + XCTAssertEqual(received, [Tuple2((1, "a")), Tuple2((2, "b"))]) + } + + func test_three_input_zip_comes_from_async_algorithms() async { + let sequence = zip([1, 2].async, ["a", "b"].async, [true, false].async) + var received = [Tuple3]() + for await element in sequence { + received.append(Tuple3(element)) + } + XCTAssertEqual(received, [Tuple3((1, "a", true)), Tuple3((2, "b", false))]) + } + + func test_two_input_merge_comes_from_async_algorithms() async { + let sequence = merge([1, 2].async, [3, 4].async) + let appleSequence: AsyncMerge2Sequence, AsyncSyncSequence<[Int]>> = sequence + var received = [Int]() + for await element in appleSequence { + received.append(element) + } + XCTAssertEqual(received.sorted(), [1, 2, 3, 4]) + } + + func test_three_input_merge_comes_from_async_algorithms() async { + let sequence = merge([1, 2].async, [3, 4].async, [5, 6].async) + let appleSequence: AsyncMerge3Sequence, AsyncSyncSequence<[Int]>, AsyncSyncSequence<[Int]>> = sequence + var received = [Int]() + for await element in appleSequence { + received.append(element) + } + XCTAssertEqual(received.sorted(), [1, 2, 3, 4, 5, 6]) + } + + func test_four_input_zip_remains_available_from_async_extensions() async { + let sequence = zip([1, 2].async, [10, 20].async, [100, 200].async, [1000, 2000].async) + var received = [[Int]]() + for await element in sequence { + received.append(element) + } + XCTAssertEqual(received, [[1, 10, 100, 1000], [2, 20, 200, 2000]]) + } + + func test_four_input_merge_remains_available_from_async_extensions() async { + let sequence = merge([1, 2].async, [3, 4].async, [5, 6].async, [7, 8].async) + var received = [Int]() + for await element in sequence { + received.append(element) + } + XCTAssertEqual(received.sorted(), [1, 2, 3, 4, 5, 6, 7, 8]) + } +} diff --git a/Tests/Combiners/Merge/AsyncMergeSequenceTests.swift b/Tests/Combiners/Merge/AsyncMergeSequenceTests.swift index 995f8d4..03aa76c 100644 --- a/Tests/Combiners/Merge/AsyncMergeSequenceTests.swift +++ b/Tests/Combiners/Merge/AsyncMergeSequenceTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 01/01/2022. // +import AsyncAlgorithms import AsyncExtensions import XCTest @@ -81,7 +82,7 @@ final class AsyncMergeSequenceTests: XCTestCase { let asyncSequence1 = TimedAsyncSequence(intervalInMills: [0, 1000, 1000], sequence: ["a", "b", "c"]) let asyncSequence2 = TimedAsyncSequence(intervalInMills: [500, 1000], sequence: ["d", "e"]) - let sut = merge(asyncSequence1, asyncSequence2) + let sut = AsyncExtensions.merge(asyncSequence1, asyncSequence2) var receivedElements = [String]() var iterator = sut.makeAsyncIterator() @@ -104,7 +105,7 @@ final class AsyncMergeSequenceTests: XCTestCase { let expectedElements = asyncSequence1 + asyncSequence2 + asyncSequence3 + asyncSequence4 - let sut = merge(asyncSequence1.async, asyncSequence2.async, asyncSequence3.async, asyncSequence4.async) + let sut = AsyncExtensions.merge(asyncSequence1.async, asyncSequence2.async, asyncSequence3.async, asyncSequence4.async) var receivedElements = [Int]() var iterator = sut.makeAsyncIterator() @@ -132,32 +133,40 @@ final class AsyncMergeSequenceTests: XCTestCase { let stream2 = AsyncPassthroughSubject() let stream3 = AsyncPassthroughSubject() - let sut = merge(stream1, stream2, stream3) + let sut = AsyncExtensions.merge( + stream1.eraseToAnyAsyncSequence(), + stream2.eraseToAnyAsyncSequence(), + stream3.eraseToAnyAsyncSequence() + ) Task { var receivedElements = [Int]() - for await element in sut { - receivedElements.append(element) - if element == 1 { - canSend2Expectation.fulfill() - } - if element == 2 { - canSend3Expectation.fulfill() - } - if element == 3 { - canSend4Expectation.fulfill() - } - if element == 4 { - canSend5Expectation.fulfill() - } - if element == 5 { - canSend6Expectation.fulfill() - } + do { + for try await element in sut { + receivedElements.append(element) + if element == 1 { + canSend2Expectation.fulfill() + } + if element == 2 { + canSend3Expectation.fulfill() + } + if element == 3 { + canSend4Expectation.fulfill() + } + if element == 4 { + canSend5Expectation.fulfill() + } + if element == 5 { + canSend6Expectation.fulfill() + } - if element == 6 { - canSendFinishExpectation.fulfill() + if element == 6 { + canSendFinishExpectation.fulfill() + } } + } catch { + XCTFail("The merged streams should not fail: \(error)") } XCTAssertEqual(receivedElements, [1, 2, 3, 4, 5, 6]) mergedSequenceIsFinisedExpectation.fulfill() @@ -195,7 +204,7 @@ final class AsyncMergeSequenceTests: XCTestCase { let asyncSequence2 = AsyncEmptySequence() let asyncSequence3 = AsyncEmptySequence() - let sut = merge(asyncSequence1, asyncSequence2, asyncSequence3) + let sut = AsyncExtensions.merge(asyncSequence1, asyncSequence2, asyncSequence3) for await element in sut { receivedResult.append(element) @@ -204,16 +213,16 @@ final class AsyncMergeSequenceTests: XCTestCase { XCTAssertTrue(receivedResult.isEmpty) } - func testMerge_returns_original_sequence_when_one_sequence_is_empty() async { + func testMerge_returns_original_sequence_when_one_sequence_is_empty() async throws { let expectedResult = [1, 2, 3] var receivedResult = [Int]() let asyncSequence1 = expectedResult.async let asyncSequence2 = AsyncEmptySequence() - let sut = merge(asyncSequence1, asyncSequence2) + let sut = AsyncExtensions.merge(asyncSequence1.eraseToAnyAsyncSequence(), asyncSequence2.eraseToAnyAsyncSequence()) - for await element in sut { + for try await element in sut { receivedResult.append(element) } @@ -228,7 +237,7 @@ final class AsyncMergeSequenceTests: XCTestCase { let stream1 = AsyncThrowingCurrentValueSubject(1) let stream2 = AsyncPassthroughSubject() - let sut = merge(stream1, stream2) + let sut = AsyncExtensions.merge(stream1.eraseToAnyAsyncSequence(), stream2.eraseToAnyAsyncSequence()) Task { var receivedElements = [Int]() @@ -267,7 +276,7 @@ final class AsyncMergeSequenceTests: XCTestCase { let asyncSequence2 = TimedAsyncSequence(intervalInMills: [50, 100, 100, 100], sequence: [6, 7, 8, 9]) let asyncSequence3 = TimedAsyncSequence(intervalInMills: [1, 399], sequence: [10, 11]) - let sut = merge(asyncSequence1, asyncSequence2, asyncSequence3) + let sut = AsyncExtensions.merge(asyncSequence1, asyncSequence2, asyncSequence3) let task = Task { var firstElement: Int? @@ -297,12 +306,12 @@ final class AsyncMergeSequenceTests: XCTestCase { let asyncSequence1 = AsyncCurrentValueSubject(1) let asyncSequence2 = AsyncPassthroughSubject() - let sut = merge(asyncSequence1, asyncSequence2) + let sut = AsyncExtensions.merge(asyncSequence1.eraseToAnyAsyncSequence(), asyncSequence2.eraseToAnyAsyncSequence()) let task = Task { var iterator = sut.makeAsyncIterator() canIterateExpectation.fulfill() - while let _ = await iterator.next() { + while let _ = try await iterator.next() { firstElementHasBeenReceivedExpectation.fulfill() } hasCancelExceptation.fulfill() @@ -345,7 +354,7 @@ final class AsyncMergeSequenceTests: XCTestCase { ) let failingBase = TimedAsyncSequence(intervalInMills: [0, 0], sequence: [1, 2], indexOfError: 1) - let sut = merge(failingBase, blockingBase) + let sut = AsyncExtensions.merge(failingBase.eraseToAnyAsyncSequence(), blockingBase.eraseToAnyAsyncSequence()) var iterator = sut.makeAsyncIterator() do { diff --git a/Tests/Combiners/WithLatestFrom/AsyncWithLatestFrom2SequenceTests.swift b/Tests/Combiners/WithLatestFrom/AsyncWithLatestFrom2SequenceTests.swift index 915e042..557082e 100644 --- a/Tests/Combiners/WithLatestFrom/AsyncWithLatestFrom2SequenceTests.swift +++ b/Tests/Combiners/WithLatestFrom/AsyncWithLatestFrom2SequenceTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 11/09/2022. // +import AsyncAlgorithms @testable import AsyncExtensions import XCTest diff --git a/Tests/Combiners/WithLatestFrom/AsyncWithLatestFromSequenceTests.swift b/Tests/Combiners/WithLatestFrom/AsyncWithLatestFromSequenceTests.swift index 732673f..9ae323f 100644 --- a/Tests/Combiners/WithLatestFrom/AsyncWithLatestFromSequenceTests.swift +++ b/Tests/Combiners/WithLatestFrom/AsyncWithLatestFromSequenceTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 01/04/2022. // +import AsyncAlgorithms @testable import AsyncExtensions import XCTest diff --git a/Tests/Combiners/Zip/AsyncZipSequenceTests.swift b/Tests/Combiners/Zip/AsyncZipSequenceTests.swift index 68eafaa..5243e23 100644 --- a/Tests/Combiners/Zip/AsyncZipSequenceTests.swift +++ b/Tests/Combiners/Zip/AsyncZipSequenceTests.swift @@ -31,307 +31,6 @@ private struct TimedAsyncSequence: AsyncSequence, AsyncIteratorProtocol } final class AsyncZipSequenceTests: XCTestCase { - func testZip2_respects_chronology_and_ends_when_first_sequence_ends() async { - let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4]) - let asyncSeq2 = TimedAsyncSequence(intervalInMills: 10, sequence: ["6", "7", "8", "9", "10"]) - - let sut = zip(asyncSeq1, asyncSeq2) - - var receivedElements = [(Int, String)]() - - for await element in sut { - receivedElements.append(element) - } - - XCTAssertEqual(receivedElements.map { $0.0 }, [1, 2, 3, 4]) - XCTAssertEqual(receivedElements.map { $0.1 }, ["6", "7", "8", "9"]) - } - - func testZip2_respects_chronology_and_ends_when_second_sequence_ends() async { - let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4, 5]) - let asyncSeq2 = TimedAsyncSequence(intervalInMills: 10, sequence: ["6", "7", "8"]) - - let sut = zip(asyncSeq1, asyncSeq2) - - var receivedElements = [(Int, String)]() - - for await element in sut { - receivedElements.append(element) - } - - XCTAssertEqual(receivedElements.map { $0.0 }, [1, 2, 3]) - XCTAssertEqual(receivedElements.map { $0.1 }, ["6", "7", "8"]) - } - - func testZip2_respects_returns_nil_pastEnd() async { - let asyncSeq1 = AsyncLazySequence([1, 2, 3]) - let asyncSeq2 = AsyncLazySequence(["1", "2", "3"]) - - let sut = zip(asyncSeq1, asyncSeq2) - - var receivedElements = [(Int, String)]() - let iterator = sut.makeAsyncIterator() - - while let element = await iterator.next() { - receivedElements.append(element) - } - - XCTAssertEqual(receivedElements.map { $0.0 }, [1, 2, 3]) - XCTAssertEqual(receivedElements.map { $0.1 }, ["1", "2", "3"]) - - let pastEnd = await iterator.next() - XCTAssertNil(pastEnd) - } - - func testZip2_propagates_error_when_first_fails() async throws { - let mockError = MockError(code: Int.random(in: 0...100)) - - let asyncSeq1 = AsyncFailSequence( mockError) - let asyncSeq2 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4]) - - let sut = zip(asyncSeq1, asyncSeq2) - let iterator = sut.makeAsyncIterator() - - do { - while let element = try await iterator.next() { - print(element) - } - XCTFail("The zipped sequence should fail") - } catch { - XCTAssertEqual(error as? MockError, mockError) - } - - let pastFail = try await iterator.next() - XCTAssertNil(pastFail) - } - - func testZip2_propagates_error_when_second_fails() async throws { - let mockError = MockError(code: Int.random(in: 0...100)) - - let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4]) - let asyncSeq2 = AsyncFailSequence( mockError) - - let sut = zip(asyncSeq1, asyncSeq2) - let iterator = sut.makeAsyncIterator() - - do { - while let _ = try await iterator.next() {} - XCTFail("The zipped sequence should fail") - } catch { - XCTAssertEqual(error as? MockError, mockError) - } - - let pastFail = try await iterator.next() - XCTAssertNil(pastFail) - } - - func testZip2_finishes_when_task_is_cancelled() { - let canCancelExpectation = expectation(description: "The first element has been emitted") - let hasCancelExceptation = expectation(description: "The task has been cancelled") - let taskHasFinishedExpectation = expectation(description: "The task has finished") - - let asyncSeq1 = AsyncLazySequence([1, 2, 3]) - let asyncSeq2 = AsyncLazySequence(["1", "2", "3"]) - - let sut = zip(asyncSeq1, asyncSeq2) - - let task = Task { - var firstElement: (Int, String)? - for try await element in sut { - firstElement = element - canCancelExpectation.fulfill() - await fulfillment(of: [hasCancelExceptation], timeout: 5) - } - XCTAssertEqual(firstElement!.0, 1) - XCTAssertEqual(firstElement!.1, "1") - taskHasFinishedExpectation.fulfill() - } - - wait(for: [canCancelExpectation], timeout: 5) // one element has been emitted, we can cancel the task - - task.cancel() - - hasCancelExceptation.fulfill() // we can release the lock in the for loop - - wait(for: [taskHasFinishedExpectation], timeout: 5) // task has been cancelled and has finished - } -} - -extension AsyncZipSequenceTests { - func testZip3_respects_chronology_and_ends_when_first_sequence_ends() async throws { - let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3]) - let asyncSeq2 = TimedAsyncSequence(intervalInMills: 10, sequence: ["6", "7", "8", "9", "10"]) - let asyncSeq3 = TimedAsyncSequence(intervalInMills: 30, sequence: [true, false, true, false, true]) - - let sut = zip(asyncSeq1, asyncSeq2, asyncSeq3) - - var receivedElements = [(Int, String, Bool)]() - - for try await element in sut { - receivedElements.append(element) - } - - XCTAssertEqual(receivedElements.map { $0.0 }, [1, 2, 3]) - XCTAssertEqual(receivedElements.map { $0.1 }, ["6", "7", "8"]) - XCTAssertEqual(receivedElements.map { $0.2 }, [true, false, true]) - } - - func testZip3_respects_chronology_and_ends_when_second_sequence_ends() async throws { - let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4, 5]) - let asyncSeq2 = TimedAsyncSequence(intervalInMills: 10, sequence: ["6", "7", "8"]) - let asyncSeq3 = TimedAsyncSequence(intervalInMills: 30, sequence: [true, false, true, false, true]) - - let sut = zip(asyncSeq1, asyncSeq2, asyncSeq3) - - var receivedElements = [(Int, String, Bool)]() - - for try await element in sut { - receivedElements.append(element) - } - - XCTAssertEqual(receivedElements.map { $0.0 }, [1, 2, 3]) - XCTAssertEqual(receivedElements.map { $0.1 }, ["6", "7", "8"]) - XCTAssertEqual(receivedElements.map { $0.2 }, [true, false, true]) - } - - func testZip3_respects_chronology_and_ends_when_third_sequence_ends() async throws { - let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4, 5]) - let asyncSeq2 = TimedAsyncSequence(intervalInMills: 10, sequence: ["6", "7", "8", "9", "10"]) - let asyncSeq3 = TimedAsyncSequence(intervalInMills: 30, sequence: [true, false, true]) - - let sut = zip(asyncSeq1, asyncSeq2, asyncSeq3) - - var receivedElements = [(Int, String, Bool)]() - - for try await element in sut { - receivedElements.append(element) - } - - XCTAssertEqual(receivedElements.map { $0.0 }, [1, 2, 3]) - XCTAssertEqual(receivedElements.map { $0.1 }, ["6", "7", "8"]) - XCTAssertEqual(receivedElements.map { $0.2 }, [true, false, true]) - } - - func testZip3_respects_returns_nil_pastEnd() async { - let asyncSeq1 = AsyncLazySequence([1, 2, 3]) - let asyncSeq2 = AsyncLazySequence(["1", "2", "3"]) - let asyncSeq3 = AsyncLazySequence([true, false, true]) - - let sut = zip(asyncSeq1, asyncSeq2, asyncSeq3) - - var receivedElements = [(Int, String, Bool)]() - let iterator = sut.makeAsyncIterator() - - while let element = await iterator.next() { - receivedElements.append(element) - } - - XCTAssertEqual(receivedElements.map { $0.0 }, [1, 2, 3]) - XCTAssertEqual(receivedElements.map { $0.1 }, ["1", "2", "3"]) - XCTAssertEqual(receivedElements.map { $0.2 }, [true, false, true]) - - let pastEnd = await iterator.next() - XCTAssertNil(pastEnd) - } - - func testZip3_propagates_error_when_first_fails() async throws { - let mockError = MockError(code: Int.random(in: 0...100)) - - let asyncSeq1 = AsyncFailSequence( mockError) - let asyncSeq2 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4]) - let asyncSeq3 = TimedAsyncSequence(intervalInMills: 5, sequence: ["1", "2", "3", "4"]) - - let sut = zip(asyncSeq1, asyncSeq2, asyncSeq3) - let iterator = sut.makeAsyncIterator() - - do { - while let _ = try await iterator.next() {} - XCTFail("The zipped sequence should fail") - } catch { - XCTAssertEqual(error as? MockError, mockError) - } - - let pastFail = try await iterator.next() - XCTAssertNil(pastFail) - } - - func testZip3_propagates_error_when_second_fails() async throws { - let mockError = MockError(code: Int.random(in: 0...100)) - - let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4]) - let asyncSeq2 = AsyncFailSequence( mockError) - let asyncSeq3 = TimedAsyncSequence(intervalInMills: 5, sequence: ["1", "2", "3", "4"]) - - let sut = zip(asyncSeq1, asyncSeq2, asyncSeq3) - let iterator = sut.makeAsyncIterator() - - do { - while let _ = try await iterator.next() {} - XCTFail("The zipped sequence should fail") - } catch { - XCTAssertEqual(error as? MockError, mockError) - } - - let pastFail = try await iterator.next() - XCTAssertNil(pastFail) - } - - func testZip3_propagates_error_when_third_fails() async throws { - let mockError = MockError(code: Int.random(in: 0...100)) - - let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4]) - let asyncSeq2 = TimedAsyncSequence(intervalInMills: 5, sequence: ["1", "2", "3", "4"]) - let asyncSeq3 = AsyncFailSequence( mockError) - - let sut = zip(asyncSeq1, asyncSeq2, asyncSeq3) - let iterator = sut.makeAsyncIterator() - - do { - while let _ = try await iterator.next() {} - XCTFail("The zipped sequence should fail") - } catch { - XCTAssertEqual(error as? MockError, mockError) - } - - let pastFail = try await iterator.next() - XCTAssertNil(pastFail) - } - - func testZip3_finishes_when_task_is_cancelled() { - let canCancelExpectation = expectation(description: "The first element has been emitted") - let hasCancelExceptation = expectation(description: "The task has been cancelled") - let taskHasFinishedExpectation = expectation(description: "The task has finished") - - let asyncSeq1 = AsyncLazySequence([1, 2, 3]) - let asyncSeq2 = AsyncLazySequence(["1", "2", "3"]) - let asyncSeq3 = AsyncLazySequence([true, false, true]) - - let sut = zip(asyncSeq1, asyncSeq2, asyncSeq3) - - let task = Task { - var firstElement: (Int, String, Bool)? - for try await element in sut { - firstElement = element - canCancelExpectation.fulfill() - await fulfillment(of: [hasCancelExceptation], timeout: 5) - } - XCTAssertEqual(firstElement!.0, 1) // the AsyncSequence is cancelled having only emitted the first element - XCTAssertEqual(firstElement!.1, "1") - XCTAssertEqual(firstElement!.2, true) - taskHasFinishedExpectation.fulfill() - } - - wait(for: [canCancelExpectation], timeout: 5) // one element has been emitted, we can cancel the task - - task.cancel() - - hasCancelExceptation.fulfill() // we can release the lock in the for loop - - wait(for: [taskHasFinishedExpectation], timeout: 5) // task has been cancelled and has finished - } -} - -extension AsyncZipSequenceTests { func testZip_respects_chronology_and_ends_when_any_sequence_ends() async { let asyncSeq1 = TimedAsyncSequence(intervalInMills: 50, sequence: [1, 2, 3, 4, 5]) let asyncSeq2 = TimedAsyncSequence(intervalInMills: 10, sequence: [1, 2, 3]) diff --git a/Tests/Operators/AsyncHandleEventsSequenceTests.swift b/Tests/Operators/AsyncHandleEventsSequenceTests.swift index eedd8d4..fe08908 100644 --- a/Tests/Operators/AsyncHandleEventsSequenceTests.swift +++ b/Tests/Operators/AsyncHandleEventsSequenceTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 01/01/2022. // +import AsyncAlgorithms @testable import AsyncExtensions import XCTest diff --git a/Tests/Operators/AsyncScanSequenceTests.swift b/Tests/Operators/AsyncScanSequenceTests.swift index 8eeb269..d9cf933 100644 --- a/Tests/Operators/AsyncScanSequenceTests.swift +++ b/Tests/Operators/AsyncScanSequenceTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 31/12/2021. // +import AsyncAlgorithms import AsyncExtensions import XCTest diff --git a/Tests/Operators/AsyncSequence+EraseToAnyAsyncSequenceTests.swift b/Tests/Operators/AsyncSequence+EraseToAnyAsyncSequenceTests.swift index 7b68f63..cdd7849 100644 --- a/Tests/Operators/AsyncSequence+EraseToAnyAsyncSequenceTests.swift +++ b/Tests/Operators/AsyncSequence+EraseToAnyAsyncSequenceTests.swift @@ -5,6 +5,7 @@ // Created by Thibault Wittemberg on 04/01/2022. // +import AsyncAlgorithms import AsyncExtensions import XCTest diff --git a/Tests/Operators/AsyncSwitchToLatestCancellationTests.swift b/Tests/Operators/AsyncSwitchToLatestCancellationTests.swift index 4320d94..fe6a252 100644 --- a/Tests/Operators/AsyncSwitchToLatestCancellationTests.swift +++ b/Tests/Operators/AsyncSwitchToLatestCancellationTests.swift @@ -1,3 +1,4 @@ +import AsyncAlgorithms @testable import AsyncExtensions import XCTest From 180ab7c05ec78b9c9436e0be8e8751bbad4853a7 Mon Sep 17 00:00:00 2001 From: Thibault Wittemberg Date: Sat, 3 Oct 2026 12:51:01 +0200 Subject: [PATCH 4/4] Disambiguate buffered timer from Swift Async Algorithms --- CHANGELOG.md | 1 + README.md | 4 +++- ...merSequence.swift => AsyncBufferedTimerSequence.swift} | 8 ++++---- Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift | 6 ++++++ ...eTests.swift => AsyncBufferedTimerSequenceTests.swift} | 6 +++--- 5 files changed, 17 insertions(+), 8 deletions(-) rename Sources/Creators/{AsyncTimerSequence.swift => AsyncBufferedTimerSequence.swift} (89%) rename Tests/Creators/{AsyncTimerSequenceTests.swift => AsyncBufferedTimerSequenceTests.swift} (78%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e750d8..230972e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ **Unreleased:** - Breaking: use Swift Async Algorithms for `Sequence.async` and two- or three-input `zip`/`merge`, removing import ambiguities. Add the `AsyncAlgorithms` product dependency and import when migrating these APIs. Preserve variadic `zip`/`merge` and the explicit `AsyncLazySequence` constructor. +- Breaking: rename the buffered Date timer from `AsyncTimerSequence` to `AsyncBufferedTimerSequence` to avoid ambiguity with Apple's clock-based timer. - SwiftPM: require Swift 5.8 or later for the Swift Async Algorithms test dependency. - SwitchToLatest: finish cancelled collection while the latest channel or outer sequence remains open, and discard late producer results (https://github.com/sideeffect-io/AsyncExtensions/issues/53). diff --git a/README.md b/README.md index 70c2512..8fa7130 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,8 @@ Variadic inputs share one concrete sequence type. Use `eraseToAnyAsyncSequence() The `AsyncLazySequence(sequence)` constructor remains available when you only need AsyncExtensions. The `.async` extension is supplied exclusively by AsyncAlgorithms. +Rename uses of AsyncExtensions' `AsyncTimerSequence` to `AsyncBufferedTimerSequence`. It retains the buffered `Date` values and `DispatchTimeInterval` initializer. The unqualified `AsyncTimerSequence` name now refers to Apple's clock-based timer when both modules are imported. + ## Features ### Channels @@ -104,7 +106,7 @@ or replay consumer receives the latest stored state, followed by subsequent send * [AsyncJustSequence](./Sources/Creators/AsyncJustSequence.swift): Creates an `AsyncSequence` that emits an element an finishes * [AsyncThrowingJustSequence](./Sources/Creators/AsyncThrowingJustSequence.swift): Creates an `AsyncSequence` that emits an elements and finishes bases on a throwing closure * [AsyncLazySequence](./Sources/Creators/AsyncLazySequence.swift): Creates an async sequence from an explicit synchronous sequence -* [AsyncTimerSequence](./Sources/Creators/AsyncTimerSequence.swift): Creates an `AsyncSequence` that emits a date value periodically +* [AsyncBufferedTimerSequence](./Sources/Creators/AsyncBufferedTimerSequence.swift): Creates an `AsyncSequence` that buffers date values emitted periodically * [AsyncStream Pipe](./Sources/Creators/AsyncStream+Pipe.swift): Creates an AsyncStream and returns a tuple standing for its inputs and outputs ### Operators diff --git a/Sources/Creators/AsyncTimerSequence.swift b/Sources/Creators/AsyncBufferedTimerSequence.swift similarity index 89% rename from Sources/Creators/AsyncTimerSequence.swift rename to Sources/Creators/AsyncBufferedTimerSequence.swift index a0d6146..2bef22b 100644 --- a/Sources/Creators/AsyncTimerSequence.swift +++ b/Sources/Creators/AsyncBufferedTimerSequence.swift @@ -1,5 +1,5 @@ // -// AsyncSequences+Timer.swift +// AsyncBufferedTimerSequence.swift // // // Created by Thibault Wittemberg on 04/03/2022. @@ -20,11 +20,11 @@ private extension DispatchTimeInterval { } } -/// `AsyncTimerSequence`is an async sequence that repeatedly emits the current date on the given interval, with the given priority. +/// `AsyncBufferedTimerSequence` is an async sequence that repeatedly emits the current date on the given interval, with the given priority. /// The Dates will be buffered until the consumers are available to process them. /// /// ``` -/// let timer = AsyncTimerSequence(priority: .high, every: .seconds(1)) +/// let timer = AsyncBufferedTimerSequence(priority: .high, every: .seconds(1)) /// /// Task { /// for try await element in timer { @@ -39,7 +39,7 @@ private extension DispatchTimeInterval { /// // 2022-03-06 19:31:25 +0000 /// // 2022-03-06 19:31:26 +0000 /// ``` -public struct AsyncTimerSequence: AsyncSequence { +public struct AsyncBufferedTimerSequence: AsyncSequence { public typealias Element = Date public typealias AsyncIterator = Iterator diff --git a/Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift b/Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift index ea7478c..0abc5cc 100644 --- a/Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift +++ b/Tests/Combiners/AsyncAlgorithmsCompatibilityTests.swift @@ -67,4 +67,10 @@ final class AsyncAlgorithmsCompatibilityTests: XCTestCase { } XCTAssertEqual(received.sorted(), [1, 2, 3, 4, 5, 6, 7, 8]) } + + @available(macOS 13.0, iOS 16.0, tvOS 16.0, watchOS 9.0, *) + func test_buffered_and_clock_timer_types_coexist() { + let _: AsyncBufferedTimerSequence = AsyncBufferedTimerSequence(every: .seconds(1)) + let _: AsyncTimerSequence = .repeating(every: .seconds(1)) + } } diff --git a/Tests/Creators/AsyncTimerSequenceTests.swift b/Tests/Creators/AsyncBufferedTimerSequenceTests.swift similarity index 78% rename from Tests/Creators/AsyncTimerSequenceTests.swift rename to Tests/Creators/AsyncBufferedTimerSequenceTests.swift index ca9ff5c..af23877 100644 --- a/Tests/Creators/AsyncTimerSequenceTests.swift +++ b/Tests/Creators/AsyncBufferedTimerSequenceTests.swift @@ -1,5 +1,5 @@ // -// AsyncTimerSequenceTests.swift +// AsyncBufferedTimerSequenceTests.swift // // // Created by Thibault Wittemberg on 06/03/2022. @@ -8,12 +8,12 @@ import AsyncExtensions import XCTest -final class AsyncTimerSequenceTests: XCTestCase { +final class AsyncBufferedTimerSequenceTests: XCTestCase { func testTimer_finishes_when_task_is_cancelled() { let canCancelExpectation = expectation(description: "the timer can be cancelled") let asyncSequenceHasFinishedExpectation = expectation(description: "The async sequence has finished") - let sut = AsyncTimerSequence(priority: .userInitiated, every: .milliseconds(100)) + let sut = AsyncBufferedTimerSequence(priority: .userInitiated, every: .milliseconds(100)) let task = Task { var index = 1