Skip to content

Commit 17143c3

Browse files
haeseokleehaeseoklee
authored
Add async stream interfaces (#51)
Add Swift Concurrency interop on top of the existing RxSwift RIBs APIs, keeping Rx as the backbone: - `Single.fromAsync` / `Observable.fromAsync` — wrap an async call as a cold Observable that cancels its Task on dispose - `Task` lifecycle extensions — `cancelOnDeactivate(interactor:)` / `cancelOnStop(_:)` / `cancel(with:)` - `onAsyncStep` (root and step) — async Workflow steps built on `fromAsync` Closes #6
1 parent 83747b1 commit 17143c3

8 files changed

Lines changed: 678 additions & 1 deletion

File tree

‎RIBs.xcodeproj/project.pbxproj‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,12 @@
3333
AFB7D4031FC81C8F00045D2B /* Foundation+ExtensionsTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = AF82F6731FC81B5F006DF7BC /* Foundation+ExtensionsTests.swift */; };
3434
AFB7D4051FC81D6100045D2B /* LaunchRouterTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = AFB7D4041FC81D6100045D2B /* LaunchRouterTests.swift */; };
3535
BF5FC0F122808377004235F1 /* RxRelay.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = BF5FC0F022808377004235F1 /* RxRelay.framework */; };
36+
C0A100012B10000100A10001 /* Observable+Async.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A100032B10000100A10001 /* Observable+Async.swift */; };
37+
C0A100082B10000200A10001 /* Task+RIBs.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A1000A2B10000200A10001 /* Task+RIBs.swift */; };
38+
C0A101022B10001000A10001 /* ObservableConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A101012B10001000A10001 /* ObservableConcurrencyTests.swift */; };
39+
C0A101042B10001000A10001 /* InteractorConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A101032B10001000A10001 /* InteractorConcurrencyTests.swift */; };
40+
C0A101062B10001000A10001 /* WorkerConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A101052B10001000A10001 /* WorkerConcurrencyTests.swift */; };
41+
C0A1010C2B10001000A10001 /* WorkflowConcurrencyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A1010B2B10001000A10001 /* WorkflowConcurrencyTests.swift */; };
3642
/* End PBXBuildFile section */
3743

3844
/* Begin PBXContainerItemProxy section */
@@ -75,6 +81,12 @@
7581
AF9966B11FC40D7E00CAEAA2 /* RxSwift.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = RxSwift.framework; path = ../Carthage/Build/iOS/RxSwift.framework; sourceTree = "<group>"; };
7682
AFB7D4041FC81D6100045D2B /* LaunchRouterTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = LaunchRouterTests.swift; sourceTree = "<group>"; };
7783
BF5FC0F022808377004235F1 /* RxRelay.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = RxRelay.framework; path = ../Carthage/Build/iOS/RxRelay.framework; sourceTree = "<group>"; };
84+
C0A100032B10000100A10001 /* Observable+Async.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "Observable+Async.swift"; sourceTree = "<group>"; };
85+
C0A101012B10001000A10001 /* ObservableConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ObservableConcurrencyTests.swift; sourceTree = "<group>"; };
86+
C0A101032B10001000A10001 /* InteractorConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = InteractorConcurrencyTests.swift; sourceTree = "<group>"; };
87+
C0A101052B10001000A10001 /* WorkerConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = WorkerConcurrencyTests.swift; sourceTree = "<group>"; };
88+
C0A1010B2B10001000A10001 /* WorkflowConcurrencyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = WorkflowConcurrencyTests.swift; sourceTree = "<group>"; };
89+
C0A1000A2B10000200A10001 /* Task+RIBs.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "Task+RIBs.swift"; sourceTree = "<group>"; };
7890
E8E789432378AD000043E59E /* Package.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; name = Package.swift; path = ../Package.swift; sourceTree = SOURCE_ROOT; };
7991
/* End PBXFileReference section */
8092

@@ -110,6 +122,7 @@
110122
413177271F8EEFEF005F08F0 /* Router.swift */,
111123
413177241F8EEFEF005F08F0 /* ViewableRouter.swift */,
112124
413177281F8EEFEF005F08F0 /* ViewControllable.swift */,
125+
C0A100002B10000100A10001 /* Concurrency */,
113126
418C17551F97DB0E003C03F7 /* DI */,
114127
4131773B1F8EF981005F08F0 /* Extensions */,
115128
413177361F8EF70A005F08F0 /* LeakDetector */,
@@ -119,6 +132,15 @@
119132
path = Classes;
120133
sourceTree = "<group>";
121134
};
135+
C0A100002B10000100A10001 /* Concurrency */ = {
136+
isa = PBXGroup;
137+
children = (
138+
C0A100032B10000100A10001 /* Observable+Async.swift */,
139+
C0A1000A2B10000200A10001 /* Task+RIBs.swift */,
140+
);
141+
path = Concurrency;
142+
sourceTree = "<group>";
143+
};
122144
413177201F8EEFEF005F08F0 /* Workflow */ = {
123145
isa = PBXGroup;
124146
children = (
@@ -195,6 +217,7 @@
195217
isa = PBXGroup;
196218
children = (
197219
8B9882F21F86E1CF00ABE009 /* Info.plist */,
220+
C0A1000B2B10000300A10001 /* Concurrency */,
198221
AFB7D4041FC81D6100045D2B /* LaunchRouterTests.swift */,
199222
AF90B40B1FBA157700920384 /* Mocks.swift */,
200223
AF90B4091FBA14DB00920384 /* RouterTests.swift */,
@@ -206,6 +229,17 @@
206229
path = RIBsTests;
207230
sourceTree = "<group>";
208231
};
232+
C0A1000B2B10000300A10001 /* Concurrency */ = {
233+
isa = PBXGroup;
234+
children = (
235+
C0A101032B10001000A10001 /* InteractorConcurrencyTests.swift */,
236+
C0A101012B10001000A10001 /* ObservableConcurrencyTests.swift */,
237+
C0A101052B10001000A10001 /* WorkerConcurrencyTests.swift */,
238+
C0A1010B2B10001000A10001 /* WorkflowConcurrencyTests.swift */,
239+
);
240+
path = Concurrency;
241+
sourceTree = "<group>";
242+
};
209243
AF5101421FBBA64A009C0DB3 /* Frameworks */ = {
210244
isa = PBXGroup;
211245
children = (
@@ -392,6 +426,8 @@
392426
4131772D1F8EF5FF005F08F0 /* Interactor.swift in Sources */,
393427
413177321F8EF5FF005F08F0 /* ViewableRouter.swift in Sources */,
394428
413177351F8EF605005F08F0 /* Workflow.swift in Sources */,
429+
C0A100012B10000100A10001 /* Observable+Async.swift in Sources */,
430+
C0A100082B10000200A10001 /* Task+RIBs.swift in Sources */,
395431
4131772F1F8EF5FF005F08F0 /* PresentableInteractor.swift in Sources */,
396432
4131772C1F8EF5FF005F08F0 /* Builder.swift in Sources */,
397433
413177331F8EF5FF005F08F0 /* ViewControllable.swift in Sources */,
@@ -404,6 +440,10 @@
404440
isa = PBXSourcesBuildPhase;
405441
buildActionMask = 2147483647;
406442
files = (
443+
C0A101042B10001000A10001 /* InteractorConcurrencyTests.swift in Sources */,
444+
C0A101022B10001000A10001 /* ObservableConcurrencyTests.swift in Sources */,
445+
C0A101062B10001000A10001 /* WorkerConcurrencyTests.swift in Sources */,
446+
C0A1010C2B10001000A10001 /* WorkflowConcurrencyTests.swift in Sources */,
407447
AFB7D4031FC81C8F00045D2B /* Foundation+ExtensionsTests.swift in Sources */,
408448
AF90B4111FBA185E00920384 /* ComponentTests.swift in Sources */,
409449
AFB7D4051FC81D6100045D2B /* LaunchRouterTests.swift in Sources */,

‎RIBs/Classes/Concurrency/Observable+Async.swift‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
//
2+
// Copyright (c) 2017. Uber Technologies
3+
//
4+
// Licensed under the Apache License, Version 2.0 (the "License");
5+
// you may not use this file except in compliance with the License.
6+
// You may obtain a copy of the License at
7+
//
8+
// http://www.apache.org/licenses/LICENSE-2.0
9+
//
10+
// Unless required by applicable law or agreed to in writing, software
11+
// distributed under the License is distributed on an "AS IS" BASIS,
12+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
// See the License for the specific language governing permissions and
14+
// limitations under the License.
15+
//
16+
17+
#if swift(>=5.6) && canImport(_Concurrency)
18+
19+
import RxSwift
20+
21+
public extension PrimitiveSequenceType where Trait == SingleTrait {
22+
23+
/// Create a cold single that runs the given async work on subscription.
24+
static func fromAsync(_ work: @escaping () async throws -> Element) -> Single<Element> {
25+
return .create { observer in
26+
let task = Task {
27+
do {
28+
observer(.success(try await work()))
29+
} catch {
30+
observer(.failure(error))
31+
}
32+
}
33+
34+
return Disposables.create {
35+
task.cancel()
36+
}
37+
}
38+
}
39+
}
40+
41+
public extension ObservableType {
42+
43+
/// Create a cold observable that runs the given async work on subscription.
44+
static func fromAsync(_ work: @escaping () async throws -> Element) -> Observable<Element> {
45+
return Single<Element>.fromAsync(work).asObservable()
46+
}
47+
}
48+
#endif

‎RIBs/Classes/Concurrency/Task+RIBs.swift‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
//
2+
// Copyright (c) 2017. Uber Technologies
3+
//
4+
// Licensed under the Apache License, Version 2.0 (the "License");
5+
// you may not use this file except in compliance with the License.
6+
// You may obtain a copy of the License at
7+
//
8+
// http://www.apache.org/licenses/LICENSE-2.0
9+
//
10+
// Unless required by applicable law or agreed to in writing, software
11+
// distributed under the License is distributed on an "AS IS" BASIS,
12+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
// See the License for the specific language governing permissions and
14+
// limitations under the License.
15+
//
16+
17+
#if swift(>=5.6) && canImport(_Concurrency)
18+
19+
import RxSwift
20+
21+
public extension Task {
22+
23+
/// Cancel this task when the given interactor deactivates.
24+
@discardableResult
25+
func cancelOnDeactivate(interactor: Interactor) -> Task<Success, Failure> {
26+
Disposables.create {
27+
self.cancel()
28+
}
29+
.disposeOnDeactivate(interactor: interactor)
30+
return self
31+
}
32+
33+
/// Cancel this task when the given worker stops.
34+
@discardableResult
35+
func cancelOnStop(_ worker: Worker) -> Task<Success, Failure> {
36+
Disposables.create {
37+
self.cancel()
38+
}
39+
.disposeOnStop(worker)
40+
return self
41+
}
42+
43+
/// Cancel this task when the given workflow is disposed.
44+
@discardableResult
45+
func cancel<ActionableItemType>(with workflow: Workflow<ActionableItemType>) -> Task<Success, Failure> {
46+
Disposables.create {
47+
self.cancel()
48+
}
49+
.disposeWith(workflow: workflow)
50+
return self
51+
}
52+
}
53+
#endif

‎RIBs/Classes/Workflow/Workflow.swift‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,19 @@ open class Workflow<ActionableItemType> {
6363
}
6464
}
6565

66+
/// Execute the given async closure as the root step.
67+
///
68+
/// - parameter onStep: The async closure to execute for the root step.
69+
/// - returns: The next step.
70+
public final func onAsyncStep<NextActionableItemType, NextValueType>(_ onStep: @escaping (ActionableItemType) async throws -> (NextActionableItemType, NextValueType)) -> Step<ActionableItemType, NextActionableItemType, NextValueType> {
71+
return self.onStep { actionableItem in
72+
Single.fromAsync {
73+
try await onStep(actionableItem)
74+
}
75+
.asObservable()
76+
}
77+
}
78+
6679
/// Subscribe and start the `Workflow` sequence.
6780
///
6881
/// - parameter actionableItem: The initial actionable item for the first step.
@@ -146,6 +159,19 @@ open class Step<WorkflowActionableItemType, ActionableItemType, ValueType> {
146159
return Step<WorkflowActionableItemType, NextActionableItemType, NextValueType>(workflow: workflow, observable: confinedNextStep)
147160
}
148161

162+
/// Executes the given async closure for this step.
163+
///
164+
/// - parameter onStep: The async closure to execute for the `Step`.
165+
/// - returns: The next step.
166+
public final func onAsyncStep<NextActionableItemType, NextValueType>(_ onStep: @escaping (ActionableItemType, ValueType) async throws -> (NextActionableItemType, NextValueType)) -> Step<WorkflowActionableItemType, NextActionableItemType, NextValueType> {
167+
return self.onStep { actionableItem, value in
168+
Single.fromAsync {
169+
try await onStep(actionableItem, value)
170+
}
171+
.asObservable()
172+
}
173+
}
174+
149175
/// Executes the given closure when the `Step` produces an error.
150176
///
151177
/// - parameter onError: The closure to execute when an error occurs.
@@ -175,9 +201,10 @@ open class Step<WorkflowActionableItemType, ActionableItemType, ValueType> {
175201
public final func asObservable() -> Observable<(ActionableItemType, ValueType)> {
176202
return observable
177203
}
204+
178205
}
179206

180-
/// `Workflow` related obervable extensions.
207+
181208
public extension ObservableType {
182209

183210
/// Fork the step from this obervable.

‎RIBsTests/Concurrency/InteractorConcurrencyTests.swift‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
//
2+
// Copyright (c) 2017. Uber Technologies
3+
//
4+
// Licensed under the Apache License, Version 2.0 (the "License");
5+
// you may not use this file except in compliance with the License.
6+
// You may obtain a copy of the License at
7+
//
8+
// http://www.apache.org/licenses/LICENSE-2.0
9+
//
10+
// Unless required by applicable law or agreed to in writing, software
11+
// distributed under the License is distributed on an "AS IS" BASIS,
12+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
// See the License for the specific language governing permissions and
14+
// limitations under the License.
15+
//
16+
17+
import XCTest
18+
@testable import RIBs
19+
20+
final class InteractorConcurrencyTests: XCTestCase {
21+
22+
func test_taskCancelOnDeactivate_cancelsTaskWhenInteractorDeactivates() async {
23+
let interactor = Interactor()
24+
interactor.activate()
25+
let taskCancelled = expectation(description: "Task cancelled")
26+
27+
let task = Task {
28+
while !Task.isCancelled {
29+
await Task.yield()
30+
}
31+
taskCancelled.fulfill()
32+
}
33+
.cancelOnDeactivate(interactor: interactor)
34+
XCTAssertFalse(task.isCancelled)
35+
36+
interactor.deactivate()
37+
await fulfillment(of: [taskCancelled], timeout: 1)
38+
39+
XCTAssertTrue(task.isCancelled)
40+
}
41+
42+
func test_taskCancelOnDeactivate_cancelsImmediatelyWhenInteractorIsInactive() async {
43+
let interactor = Interactor()
44+
let taskCancelled = expectation(description: "Task cancelled")
45+
46+
let task = Task {
47+
while !Task.isCancelled {
48+
await Task.yield()
49+
}
50+
taskCancelled.fulfill()
51+
}
52+
.cancelOnDeactivate(interactor: interactor)
53+
54+
await fulfillment(of: [taskCancelled], timeout: 1)
55+
XCTAssertTrue(task.isCancelled)
56+
}
57+
58+
func test_throwingTaskCancelOnDeactivate_cancelsTaskWhenInteractorDeactivates() async {
59+
let interactor = Interactor()
60+
interactor.activate()
61+
62+
let task = Task {
63+
try await Task.sleep(nanoseconds: 10_000_000_000)
64+
}
65+
.cancelOnDeactivate(interactor: interactor)
66+
XCTAssertFalse(task.isCancelled)
67+
68+
interactor.deactivate()
69+
70+
do {
71+
try await task.value
72+
XCTFail("Expected task to throw CancellationError")
73+
} catch is CancellationError {
74+
// Expected.
75+
} catch {
76+
XCTFail("Unexpected error: \(error)")
77+
}
78+
79+
XCTAssertTrue(task.isCancelled)
80+
}
81+
}

0 commit comments

Comments
 (0)