diff --git a/RIBs.xcodeproj/project.pbxproj b/RIBs.xcodeproj/project.pbxproj index 049e88a..e66a516 100644 --- a/RIBs.xcodeproj/project.pbxproj +++ b/RIBs.xcodeproj/project.pbxproj @@ -33,6 +33,12 @@ AFB7D4031FC81C8F00045D2B /* Foundation+ExtensionsTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = AF82F6731FC81B5F006DF7BC /* Foundation+ExtensionsTests.swift */; }; AFB7D4051FC81D6100045D2B /* LaunchRouterTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = AFB7D4041FC81D6100045D2B /* LaunchRouterTests.swift */; }; BF5FC0F122808377004235F1 /* RxRelay.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = BF5FC0F022808377004235F1 /* RxRelay.framework */; }; + C0A100012B10000100A10001 /* Observable+Async.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A100032B10000100A10001 /* Observable+Async.swift */; }; + C0A100082B10000200A10001 /* Task+RIBs.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A1000A2B10000200A10001 /* Task+RIBs.swift */; }; + C0A101022B10001000A10001 /* ObservableConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A101012B10001000A10001 /* ObservableConcurrencyTests.swift */; }; + C0A101042B10001000A10001 /* InteractorConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A101032B10001000A10001 /* InteractorConcurrencyTests.swift */; }; + C0A101062B10001000A10001 /* WorkerConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A101052B10001000A10001 /* WorkerConcurrencyTests.swift */; }; + C0A1010C2B10001000A10001 /* WorkflowConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A1010B2B10001000A10001 /* WorkflowConcurrencyTests.swift */; }; /* End PBXBuildFile section */ /* Begin PBXContainerItemProxy section */ @@ -75,6 +81,12 @@ AF9966B11FC40D7E00CAEAA2 /* RxSwift.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = RxSwift.framework; path = ../Carthage/Build/iOS/RxSwift.framework; sourceTree = ""; }; AFB7D4041FC81D6100045D2B /* LaunchRouterTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = LaunchRouterTests.swift; sourceTree = ""; }; BF5FC0F022808377004235F1 /* RxRelay.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = RxRelay.framework; path = ../Carthage/Build/iOS/RxRelay.framework; sourceTree = ""; }; + C0A100032B10000100A10001 /* Observable+Async.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "Observable+Async.swift"; sourceTree = ""; }; + C0A101012B10001000A10001 /* ObservableConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ObservableConcurrencyTests.swift; sourceTree = ""; }; + C0A101032B10001000A10001 /* InteractorConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = InteractorConcurrencyTests.swift; sourceTree = ""; }; + C0A101052B10001000A10001 /* WorkerConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = WorkerConcurrencyTests.swift; sourceTree = ""; }; + C0A1010B2B10001000A10001 /* WorkflowConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = WorkflowConcurrencyTests.swift; sourceTree = ""; }; + C0A1000A2B10000200A10001 /* Task+RIBs.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "Task+RIBs.swift"; sourceTree = ""; }; E8E789432378AD000043E59E /* Package.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; name = Package.swift; path = ../Package.swift; sourceTree = SOURCE_ROOT; }; /* End PBXFileReference section */ @@ -110,6 +122,7 @@ 413177271F8EEFEF005F08F0 /* Router.swift */, 413177241F8EEFEF005F08F0 /* ViewableRouter.swift */, 413177281F8EEFEF005F08F0 /* ViewControllable.swift */, + C0A100002B10000100A10001 /* Concurrency */, 418C17551F97DB0E003C03F7 /* DI */, 4131773B1F8EF981005F08F0 /* Extensions */, 413177361F8EF70A005F08F0 /* LeakDetector */, @@ -119,6 +132,15 @@ path = Classes; sourceTree = ""; }; + C0A100002B10000100A10001 /* Concurrency */ = { + isa = PBXGroup; + children = ( + C0A100032B10000100A10001 /* Observable+Async.swift */, + C0A1000A2B10000200A10001 /* Task+RIBs.swift */, + ); + path = Concurrency; + sourceTree = ""; + }; 413177201F8EEFEF005F08F0 /* Workflow */ = { isa = PBXGroup; children = ( @@ -195,6 +217,7 @@ isa = PBXGroup; children = ( 8B9882F21F86E1CF00ABE009 /* Info.plist */, + C0A1000B2B10000300A10001 /* Concurrency */, AFB7D4041FC81D6100045D2B /* LaunchRouterTests.swift */, AF90B40B1FBA157700920384 /* Mocks.swift */, AF90B4091FBA14DB00920384 /* RouterTests.swift */, @@ -206,6 +229,17 @@ path = RIBsTests; sourceTree = ""; }; + C0A1000B2B10000300A10001 /* Concurrency */ = { + isa = PBXGroup; + children = ( + C0A101032B10001000A10001 /* InteractorConcurrencyTests.swift */, + C0A101012B10001000A10001 /* ObservableConcurrencyTests.swift */, + C0A101052B10001000A10001 /* WorkerConcurrencyTests.swift */, + C0A1010B2B10001000A10001 /* WorkflowConcurrencyTests.swift */, + ); + path = Concurrency; + sourceTree = ""; + }; AF5101421FBBA64A009C0DB3 /* Frameworks */ = { isa = PBXGroup; children = ( @@ -392,6 +426,8 @@ 4131772D1F8EF5FF005F08F0 /* Interactor.swift in Sources */, 413177321F8EF5FF005F08F0 /* ViewableRouter.swift in Sources */, 413177351F8EF605005F08F0 /* Workflow.swift in Sources */, + C0A100012B10000100A10001 /* Observable+Async.swift in Sources */, + C0A100082B10000200A10001 /* Task+RIBs.swift in Sources */, 4131772F1F8EF5FF005F08F0 /* PresentableInteractor.swift in Sources */, 4131772C1F8EF5FF005F08F0 /* Builder.swift in Sources */, 413177331F8EF5FF005F08F0 /* ViewControllable.swift in Sources */, @@ -404,6 +440,10 @@ isa = PBXSourcesBuildPhase; buildActionMask = 2147483647; files = ( + C0A101042B10001000A10001 /* InteractorConcurrencyTests.swift in Sources */, + C0A101022B10001000A10001 /* ObservableConcurrencyTests.swift in Sources */, + C0A101062B10001000A10001 /* WorkerConcurrencyTests.swift in Sources */, + C0A1010C2B10001000A10001 /* WorkflowConcurrencyTests.swift in Sources */, AFB7D4031FC81C8F00045D2B /* Foundation+ExtensionsTests.swift in Sources */, AF90B4111FBA185E00920384 /* ComponentTests.swift in Sources */, AFB7D4051FC81D6100045D2B /* LaunchRouterTests.swift in Sources */, diff --git a/RIBs/Classes/Concurrency/Observable+Async.swift b/RIBs/Classes/Concurrency/Observable+Async.swift new file mode 100644 index 0000000..da073b0 --- /dev/null +++ b/RIBs/Classes/Concurrency/Observable+Async.swift @@ -0,0 +1,48 @@ +// +// Copyright (c) 2017. Uber Technologies +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +#if swift(>=5.6) && canImport(_Concurrency) + +import RxSwift + +public extension PrimitiveSequenceType where Trait == SingleTrait { + + /// Create a cold single that runs the given async work on subscription. + static func fromAsync(_ work: @escaping () async throws -> Element) -> Single { + return .create { observer in + let task = Task { + do { + observer(.success(try await work())) + } catch { + observer(.failure(error)) + } + } + + return Disposables.create { + task.cancel() + } + } + } +} + +public extension ObservableType { + + /// Create a cold observable that runs the given async work on subscription. + static func fromAsync(_ work: @escaping () async throws -> Element) -> Observable { + return Single.fromAsync(work).asObservable() + } +} +#endif diff --git a/RIBs/Classes/Concurrency/Task+RIBs.swift b/RIBs/Classes/Concurrency/Task+RIBs.swift new file mode 100644 index 0000000..b0bb90d --- /dev/null +++ b/RIBs/Classes/Concurrency/Task+RIBs.swift @@ -0,0 +1,53 @@ +// +// Copyright (c) 2017. Uber Technologies +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +#if swift(>=5.6) && canImport(_Concurrency) + +import RxSwift + +public extension Task { + + /// Cancel this task when the given interactor deactivates. + @discardableResult + func cancelOnDeactivate(interactor: Interactor) -> Task { + Disposables.create { + self.cancel() + } + .disposeOnDeactivate(interactor: interactor) + return self + } + + /// Cancel this task when the given worker stops. + @discardableResult + func cancelOnStop(_ worker: Worker) -> Task { + Disposables.create { + self.cancel() + } + .disposeOnStop(worker) + return self + } + + /// Cancel this task when the given workflow is disposed. + @discardableResult + func cancel(with workflow: Workflow) -> Task { + Disposables.create { + self.cancel() + } + .disposeWith(workflow: workflow) + return self + } +} +#endif diff --git a/RIBs/Classes/Workflow/Workflow.swift b/RIBs/Classes/Workflow/Workflow.swift index 472ffa3..5d69b85 100644 --- a/RIBs/Classes/Workflow/Workflow.swift +++ b/RIBs/Classes/Workflow/Workflow.swift @@ -63,6 +63,19 @@ open class Workflow { } } + /// Execute the given async closure as the root step. + /// + /// - parameter onStep: The async closure to execute for the root step. + /// - returns: The next step. + public final func onAsyncStep(_ onStep: @escaping (ActionableItemType) async throws -> (NextActionableItemType, NextValueType)) -> Step { + return self.onStep { actionableItem in + Single.fromAsync { + try await onStep(actionableItem) + } + .asObservable() + } + } + /// Subscribe and start the `Workflow` sequence. /// /// - parameter actionableItem: The initial actionable item for the first step. @@ -146,6 +159,19 @@ open class Step { return Step(workflow: workflow, observable: confinedNextStep) } + /// Executes the given async closure for this step. + /// + /// - parameter onStep: The async closure to execute for the `Step`. + /// - returns: The next step. + public final func onAsyncStep(_ onStep: @escaping (ActionableItemType, ValueType) async throws -> (NextActionableItemType, NextValueType)) -> Step { + return self.onStep { actionableItem, value in + Single.fromAsync { + try await onStep(actionableItem, value) + } + .asObservable() + } + } + /// Executes the given closure when the `Step` produces an error. /// /// - parameter onError: The closure to execute when an error occurs. @@ -175,9 +201,10 @@ open class Step { public final func asObservable() -> Observable<(ActionableItemType, ValueType)> { return observable } + } -/// `Workflow` related obervable extensions. + public extension ObservableType { /// Fork the step from this obervable. diff --git a/RIBsTests/Concurrency/InteractorConcurrencyTests.swift b/RIBsTests/Concurrency/InteractorConcurrencyTests.swift new file mode 100644 index 0000000..c69bfbe --- /dev/null +++ b/RIBsTests/Concurrency/InteractorConcurrencyTests.swift @@ -0,0 +1,81 @@ +// +// Copyright (c) 2017. Uber Technologies +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import XCTest +@testable import RIBs + +final class InteractorConcurrencyTests: XCTestCase { + + func test_taskCancelOnDeactivate_cancelsTaskWhenInteractorDeactivates() async { + let interactor = Interactor() + interactor.activate() + let taskCancelled = expectation(description: "Task cancelled") + + let task = Task { + while !Task.isCancelled { + await Task.yield() + } + taskCancelled.fulfill() + } + .cancelOnDeactivate(interactor: interactor) + XCTAssertFalse(task.isCancelled) + + interactor.deactivate() + await fulfillment(of: [taskCancelled], timeout: 1) + + XCTAssertTrue(task.isCancelled) + } + + func test_taskCancelOnDeactivate_cancelsImmediatelyWhenInteractorIsInactive() async { + let interactor = Interactor() + let taskCancelled = expectation(description: "Task cancelled") + + let task = Task { + while !Task.isCancelled { + await Task.yield() + } + taskCancelled.fulfill() + } + .cancelOnDeactivate(interactor: interactor) + + await fulfillment(of: [taskCancelled], timeout: 1) + XCTAssertTrue(task.isCancelled) + } + + func test_throwingTaskCancelOnDeactivate_cancelsTaskWhenInteractorDeactivates() async { + let interactor = Interactor() + interactor.activate() + + let task = Task { + try await Task.sleep(nanoseconds: 10_000_000_000) + } + .cancelOnDeactivate(interactor: interactor) + XCTAssertFalse(task.isCancelled) + + interactor.deactivate() + + do { + try await task.value + XCTFail("Expected task to throw CancellationError") + } catch is CancellationError { + // Expected. + } catch { + XCTFail("Unexpected error: \(error)") + } + + XCTAssertTrue(task.isCancelled) + } +} diff --git a/RIBsTests/Concurrency/ObservableConcurrencyTests.swift b/RIBsTests/Concurrency/ObservableConcurrencyTests.swift new file mode 100644 index 0000000..7024c7f --- /dev/null +++ b/RIBsTests/Concurrency/ObservableConcurrencyTests.swift @@ -0,0 +1,108 @@ +// +// Copyright (c) 2017. Uber Technologies +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import RxSwift +import XCTest +@testable import RIBs + +final class ObservableConcurrencyTests: XCTestCase { + + func test_singleFromAsync_emitsSuccess() async { + let success = expectation(description: "Single emitted success") + + _ = Single.fromAsync { + return 1 + } + .subscribe( + onSuccess: { value in + XCTAssertEqual(value, 1) + success.fulfill() + }, + onFailure: { error in + XCTFail("Unexpected error: \(error)") + } + ) + + await fulfillment(of: [success], timeout: 1) + } + + func test_singleFromAsync_emitsFailure() async { + let failure = expectation(description: "Single emitted failure") + + _ = Single.fromAsync { + throw ObservableConcurrencyTestError.error + } + .subscribe( + onSuccess: { value in + XCTFail("Unexpected value: \(value)") + }, + onFailure: { error in + guard case ObservableConcurrencyTestError.error = error else { + XCTFail("Unexpected error: \(error)") + return + } + failure.fulfill() + } + ) + + await fulfillment(of: [failure], timeout: 1) + } + + func test_singleFromAsync_cancelsTaskWhenDisposed() async { + let taskStarted = expectation(description: "Task started") + let taskCancelled = expectation(description: "Task cancelled") + let disposable = Single.fromAsync { + taskStarted.fulfill() + while !Task.isCancelled { + await Task.yield() + } + taskCancelled.fulfill() + throw CancellationError() + } + .subscribe() + + await fulfillment(of: [taskStarted], timeout: 1) + disposable.dispose() + await fulfillment(of: [taskCancelled], timeout: 1) + } + + func test_observableFromAsync_emitsValueAndCompletes() async { + let completed = expectation(description: "Observable completed") + var receivedValues: [Int] = [] + + _ = Observable.fromAsync { + return 1 + } + .subscribe( + onNext: { value in + receivedValues.append(value) + }, + onError: { error in + XCTFail("Unexpected error: \(error)") + }, + onCompleted: { + completed.fulfill() + } + ) + + await fulfillment(of: [completed], timeout: 1) + XCTAssertEqual(receivedValues, [1]) + } +} + +private enum ObservableConcurrencyTestError: Error { + case error +} diff --git a/RIBsTests/Concurrency/WorkerConcurrencyTests.swift b/RIBsTests/Concurrency/WorkerConcurrencyTests.swift new file mode 100644 index 0000000..cee5d3d --- /dev/null +++ b/RIBsTests/Concurrency/WorkerConcurrencyTests.swift @@ -0,0 +1,85 @@ +// +// Copyright (c) 2017. Uber Technologies +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import XCTest +@testable import RIBs + +final class WorkerConcurrencyTests: XCTestCase { + + func test_taskCancelOnStop_cancelsTaskWhenWorkerStops() async { + let interactor = Interactor() + let worker = Worker() + let taskCancelled = expectation(description: "Task cancelled") + interactor.activate() + worker.start(interactor) + + let task = Task { + while !Task.isCancelled { + await Task.yield() + } + taskCancelled.fulfill() + } + .cancelOnStop(worker) + XCTAssertFalse(task.isCancelled) + + worker.stop() + await fulfillment(of: [taskCancelled], timeout: 1) + + XCTAssertTrue(task.isCancelled) + } + + func test_taskCancelOnStop_cancelsImmediatelyWhenWorkerIsStopped() async { + let worker = Worker() + let taskCancelled = expectation(description: "Task cancelled") + + let task = Task { + while !Task.isCancelled { + await Task.yield() + } + taskCancelled.fulfill() + } + .cancelOnStop(worker) + + await fulfillment(of: [taskCancelled], timeout: 1) + XCTAssertTrue(task.isCancelled) + } + + func test_throwingTaskCancelOnStop_cancelsTaskWhenWorkerStops() async { + let interactor = Interactor() + let worker = Worker() + interactor.activate() + worker.start(interactor) + + let task = Task { + try await Task.sleep(nanoseconds: 10_000_000_000) + } + .cancelOnStop(worker) + XCTAssertFalse(task.isCancelled) + + worker.stop() + + do { + try await task.value + XCTFail("Expected task to throw CancellationError") + } catch is CancellationError { + // Expected. + } catch { + XCTFail("Unexpected error: \(error)") + } + + XCTAssertTrue(task.isCancelled) + } +} diff --git a/RIBsTests/Concurrency/WorkflowConcurrencyTests.swift b/RIBsTests/Concurrency/WorkflowConcurrencyTests.swift new file mode 100644 index 0000000..e072fdd --- /dev/null +++ b/RIBsTests/Concurrency/WorkflowConcurrencyTests.swift @@ -0,0 +1,235 @@ +// +// Copyright (c) 2017. Uber Technologies +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import RxSwift +import XCTest +@testable import RIBs + +final class WorkflowConcurrencyTests: XCTestCase { + + func test_workflowTaskCancelWithWorkflow_cancelsWhenWorkflowDisposableIsDisposed() async { + let workflow = Workflow<()>() + let taskCancelled = expectation(description: "Task cancelled") + let task = Task { + while !Task.isCancelled { + await Task.yield() + } + taskCancelled.fulfill() + } + .cancel(with: workflow) + let disposable = workflow + .onStep { _ in + Observable.just(((), ())) + } + .commit() + .subscribe(()) + + XCTAssertFalse(task.isCancelled) + disposable.dispose() + await fulfillment(of: [taskCancelled], timeout: 1) + + XCTAssertTrue(task.isCancelled) + } + + func test_throwingTaskCancelWithWorkflow_cancelsWhenWorkflowDisposableIsDisposed() async { + let workflow = Workflow<()>() + let task = Task { + try await Task.sleep(nanoseconds: 10_000_000_000) + } + .cancel(with: workflow) + let disposable = workflow + .onStep { _ in + Observable.just(((), ())) + } + .commit() + .subscribe(()) + + XCTAssertFalse(task.isCancelled) + disposable.dispose() + + do { + try await task.value + XCTFail("Expected task to throw CancellationError") + } catch is CancellationError { + // Expected. + } catch { + XCTFail("Unexpected error: \(error)") + } + + XCTAssertTrue(task.isCancelled) + } + + func test_workflowOnAsyncStep_emitsAsyncResultAndInvokesWorkflowDidComplete() async { + let workflow = WorkflowConcurrencyTestWorkflow() + let stepRan = expectation(description: "Root async step ran") + workflow.onComplete = { + stepRan.fulfill() + } + + _ = workflow + .onAsyncStep { actionableItem in + return (actionableItem.count, actionableItem) + } + .onStep { length, value in + XCTAssertEqual(length, 4) + XCTAssertEqual(value, "test") + return Observable.just(((), ())) + } + .commit() + .subscribe("test") + + await fulfillment(of: [stepRan], timeout: 1) + XCTAssertEqual(workflow.completeCallCount, 1) + XCTAssertEqual(workflow.errorCallCount, 0) + } + + func test_workflowOnAsyncStep_invokesWorkflowDidReceiveError() async { + let workflow = WorkflowConcurrencyTestWorkflow() + let receivedError = expectation(description: "Workflow received error") + workflow.onError = { + receivedError.fulfill() + } + + _ = workflow + .onAsyncStep { _ -> (Int, Int) in + throw WorkflowConcurrencyTestError.error + } + .commit() + .subscribe(1) + + await fulfillment(of: [receivedError], timeout: 1) + XCTAssertEqual(workflow.completeCallCount, 0) + XCTAssertEqual(workflow.errorCallCount, 1) + } + + func test_workflowOnAsyncStep_cancelsTaskWhenWorkflowDisposableIsDisposed() async { + let workflow = Workflow<()>() + let taskStarted = expectation(description: "Root async step task started") + let taskCancelled = expectation(description: "Root async step task cancelled") + let disposable = workflow + .onAsyncStep { _ -> ((), ()) in + taskStarted.fulfill() + while !Task.isCancelled { + await Task.yield() + } + taskCancelled.fulfill() + return ((), ()) + } + .commit() + .subscribe(()) + + await fulfillment(of: [taskStarted], timeout: 1) + disposable.dispose() + await fulfillment(of: [taskCancelled], timeout: 1) + } + + func test_onAsyncStep_emitsAsyncResult() async { + let workflow = Workflow() + let stepRan = expectation(description: "Async step ran") + + _ = workflow + .onStep { actionableItem in + Observable.just((actionableItem, actionableItem)) + } + .onAsyncStep { actionableItem, value in + return (actionableItem + 1, value + 2) + } + .onStep { actionableItem, value in + XCTAssertEqual(actionableItem, 2) + XCTAssertEqual(value, 3) + stepRan.fulfill() + return Observable.just(((), ())) + } + .commit() + .subscribe(1) + + await fulfillment(of: [stepRan], timeout: 1) + } + + func test_onAsyncStep_invokesWorkflowDidReceiveError() async { + let workflow = WorkflowConcurrencyTestWorkflow() + let receivedError = expectation(description: "Workflow received error") + workflow.onError = { + receivedError.fulfill() + } + + _ = workflow + .onStep { actionableItem in + Observable.just((actionableItem, actionableItem)) + } + .onAsyncStep { _, _ -> (Int, Int) in + throw WorkflowConcurrencyTestError.error + } + .commit() + .subscribe(1) + + await fulfillment(of: [receivedError], timeout: 1) + XCTAssertEqual(workflow.completeCallCount, 0) + XCTAssertEqual(workflow.errorCallCount, 1) + } + + func test_onAsyncStep_cancelsTaskWhenWorkflowDisposableIsDisposed() async { + let workflow = Workflow<()>() + let taskStarted = expectation(description: "Async step task started") + let taskCancelled = expectation(description: "Async step task cancelled") + let disposable = workflow + .onStep { _ in + Observable.just(((), ())) + } + .onAsyncStep { _, _ -> ((), ()) in + taskStarted.fulfill() + while !Task.isCancelled { + await Task.yield() + } + taskCancelled.fulfill() + return ((), ()) + } + .commit() + .subscribe(()) + + await fulfillment(of: [taskStarted], timeout: 1) + disposable.dispose() + await fulfillment(of: [taskCancelled], timeout: 1) + } +} + +private enum WorkflowConcurrencyTestError: Error { + case error +} + +private final class WorkflowConcurrencyTestWorkflow: Workflow { + var completeCallCount = 0 + var errorCallCount = 0 + var forkCallCount = 0 + var receivedError: Error? + var onComplete: (() -> Void)? + var onError: (() -> Void)? + + override func didComplete() { + completeCallCount += 1 + onComplete?() + } + + override func didFork() { + forkCallCount += 1 + } + + override func didReceiveError(_ error: Error) { + errorCallCount += 1 + receivedError = error + onError?() + } +}