Add async stream interfaces

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 = "<group>"; };
 		AFB7D4041FC81D6100045D2B /* LaunchRouterTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = LaunchRouterTests.swift; sourceTree = "<group>"; };
 		BF5FC0F022808377004235F1 /* RxRelay.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = RxRelay.framework; path = ../Carthage/Build/iOS/RxRelay.framework; sourceTree = "<group>"; };
+		C0A100032B10000100A10001 /* Observable+Async.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "Observable+Async.swift"; sourceTree = "<group>"; };
+		C0A101012B10001000A10001 /* ObservableConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ObservableConcurrencyTests.swift; sourceTree = "<group>"; };
+		C0A101032B10001000A10001 /* InteractorConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = InteractorConcurrencyTests.swift; sourceTree = "<group>"; };
+		C0A101052B10001000A10001 /* WorkerConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = WorkerConcurrencyTests.swift; sourceTree = "<group>"; };
+		C0A1010B2B10001000A10001 /* WorkflowConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = WorkflowConcurrencyTests.swift; sourceTree = "<group>"; };
+		C0A1000A2B10000200A10001 /* Task+RIBs.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "Task+RIBs.swift"; sourceTree = "<group>"; };
 		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 = "<group>";
 		};
+		C0A100002B10000100A10001 /* Concurrency */ = {
+			isa = PBXGroup;
+			children = (
+				C0A100032B10000100A10001 /* Observable+Async.swift */,
+				C0A1000A2B10000200A10001 /* Task+RIBs.swift */,
+			);
+			path = Concurrency;
+			sourceTree = "<group>";
+		};
 		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 = "<group>";
 		};
+		C0A1000B2B10000300A10001 /* Concurrency */ = {
+			isa = PBXGroup;
+			children = (
+				C0A101032B10001000A10001 /* InteractorConcurrencyTests.swift */,
+				C0A101012B10001000A10001 /* ObservableConcurrencyTests.swift */,
+				C0A101052B10001000A10001 /* WorkerConcurrencyTests.swift */,
+				C0A1010B2B10001000A10001 /* WorkflowConcurrencyTests.swift */,
+			);
+			path = Concurrency;
+			sourceTree = "<group>";
+		};
 		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 */,

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<Element> {
+        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<Element> {
+        return Single<Element>.fromAsync(work).asObservable()
+    }
+}
+#endif

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<Success, Failure> {
+        Disposables.create {
+            self.cancel()
+        }
+        .disposeOnDeactivate(interactor: interactor)
+        return self
+    }
+
+    /// Cancel this task when the given worker stops.
+    @discardableResult
+    func cancelOnStop(_ worker: Worker) -> Task<Success, Failure> {
+        Disposables.create {
+            self.cancel()
+        }
+        .disposeOnStop(worker)
+        return self
+    }
+
+    /// Cancel this task when the given workflow is disposed.
+    @discardableResult
+    func cancel<ActionableItemType>(with workflow: Workflow<ActionableItemType>) -> Task<Success, Failure> {
+        Disposables.create {
+            self.cancel()
+        }
+        .disposeWith(workflow: workflow)
+        return self
+    }
+}
+#endif

RIBs/Classes/Workflow/Workflow.swift

@@ -63,6 +63,19 @@ open class Workflow<ActionableItemType> {
             }
     }
 
+    /// 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<NextActionableItemType, NextValueType>(_ onStep: @escaping (ActionableItemType) async throws -> (NextActionableItemType, NextValueType)) -> Step<ActionableItemType, NextActionableItemType, NextValueType> {
+        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<WorkflowActionableItemType, ActionableItemType, ValueType> {
         return Step<WorkflowActionableItemType, NextActionableItemType, NextValueType>(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<NextActionableItemType, NextValueType>(_ onStep: @escaping (ActionableItemType, ValueType) async throws -> (NextActionableItemType, NextValueType)) -> Step<WorkflowActionableItemType, NextActionableItemType, NextValueType> {
+        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<WorkflowActionableItemType, ActionableItemType, ValueType> {
     public final func asObservable() -> Observable<(ActionableItemType, ValueType)> {
         return observable
     }
+
 }
 
-/// `Workflow` related obervable extensions.
+
 public extension ObservableType {
 
     /// Fork the step from this obervable.

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)
+    }
+}