Fix forked Workflow invoking didComplete multiple times

CHANGELOG.md

@@ -3,3 +3,7 @@
 ### Version 0.1.0
 
 * Initial release
+
+### Version 0.9.2
+
+* Fix forked Workflow invoking didComplete multiple times

RIBs.podspec

@@ -1,6 +1,6 @@
 Pod::Spec.new do |s|
   s.name             = 'RIBs'
-  s.version          = '0.9.1'
+  s.version          = '0.9.2'
   s.summary          = 'Uber\'s cross-platform mobile architecture.'
   s.description      = <<-DESC
 RIBs is the cross-platform architecture behind many mobile apps at Uber. This architecture framework is designed for mobile apps with a large number of engineers and nested states.

ios/RIBs/Classes/Workflow/Workflow.swift

@@ -77,10 +77,25 @@ open class Workflow<ActionableItemType> {
         return compositeDisposable
     }
 
-    // MARK: - Private Interface
+    // MARK: - Private
 
     private let subject = PublishSubject<(ActionableItemType, ())>()
+    private var didInvokeComplete = false
+
+    /// The composite disposable that contains all subscriptions including the original workflow
+    /// as well as all the forked ones.
     fileprivate let compositeDisposable = CompositeDisposable()
+
+    fileprivate func didCompleteIfNotYet() {
+        // Since a workflow may be forked to produce multiple subscribed Rx chains, we should
+        // ensure the didComplete method is only invoked once per Workflow instance. See `Step.commit`
+        // on why the side-effects must be added at the end of the Rx chains.
+        guard !didInvokeComplete else {
+            return
+        }
+        didInvokeComplete = true
+        didComplete()
+    }
 }
 
 /// Defines a single step in a `Workflow`.
@@ -145,8 +160,10 @@ open class Step<WorkflowActionableItemType, ActionableItemType, ValueType> {
     /// - returns: The committed `Workflow`.
     @discardableResult
     public final func commit() -> Workflow<WorkflowActionableItemType> {
+        // Side-effects must be chained at the last observable sequence, since errors and complete
+        // events can be emitted by any observables on any steps of the workflow.
         let disposable = observable
-            .do(onError: workflow.didReceiveError, onCompleted: workflow.didComplete)
+            .do(onError: workflow.didReceiveError, onCompleted: workflow.didCompleteIfNotYet)
             .subscribe()
         _ = workflow.compositeDisposable.insert(disposable)
         return workflow

ios/RIBsTests/Workflow/WorkflowTests.swift

@@ -20,9 +20,7 @@ import RxSwift
 
 final class WorkerflowTests: XCTestCase {
 
-    // MARK: - Tests
-
-    func test_nestedStepDoNotRepeat() {
+    func test_nestedStepsDoNotRepeat() {
         var outerStep1RunCount = 0
         var outerStep2RunCount = 0
         var outerStep3RunCount = 0
@@ -79,7 +77,7 @@ final class WorkerflowTests: XCTestCase {
         XCTAssertEqual(innerStep3RunCount, 1, "Inner step 3 should not have been run more than once")
     }
 
-    func test_workflowRecivesError() {
+    func test_workflowReceivesError() {
         let workflow = TestWorkflow()
 
         let emptyObservable = Observable.just(((), ()))
@@ -126,6 +124,37 @@ final class WorkerflowTests: XCTestCase {
         XCTAssertEqual(0, workflow.errorCallCount)
     }
 
+    func test_workflowDidFork() {
+        let workflow = TestWorkflow()
+
+        let emptyObservable = Observable.just(((), ()))
+        _ = workflow
+            .onStep { _ -> Observable<((), ())> in
+                return emptyObservable
+            }
+            .onStep { _, _ -> Observable<((), ())> in
+                return emptyObservable
+            }
+            .onStep { _, _ -> Observable<((), ())> in
+                return emptyObservable
+            }
+            .onStep { _, _ -> Observable<((), ())> in
+                let forkedStep: Step<(), (), ()>? = emptyObservable.fork(workflow)
+                forkedStep?
+                    .onStep { _, _ -> Observable<((), ())> in
+                        return emptyObservable
+                    }
+                    .commit()
+                return emptyObservable
+            }
+            .commit()
+            .subscribe(())
+
+        XCTAssertEqual(1, workflow.completeCallCount)
+        XCTAssertEqual(1, workflow.forkCallCount)
+        XCTAssertEqual(0, workflow.errorCallCount)
+    }
+
     func test_fork_verifySingleInvocationAtRoot() {
         let workflow = TestWorkflow()
 
@@ -135,7 +164,7 @@ final class WorkerflowTests: XCTestCase {
             .onStep { _ -> Observable<((), ())> in
                 rootCallCount += 1
                 return emptyObservable
-            }
+        }
 
         let firstFork: Step<(), (), ()>? = rootStep.asObservable().fork(workflow)
         _ = firstFork?
@@ -163,29 +192,20 @@ private enum WorkflowTestError: Error {
     case error
 }
 
-private final class TestWorkflow: Workflow<()> {
-
-    private(set) var completeCallCount: Int = 0
-    private(set) var errorCallCount: Int = 0
-    private(set) var forkCallCount: Int = 0
-
-    // MARK: - Overrides
+private class TestWorkflow: Workflow<()> {
+    var completeCallCount = 0
+    var errorCallCount = 0
+    var forkCallCount = 0
 
     override func didComplete() {
-        super.didComplete()
-
         completeCallCount += 1
     }
 
     override func didFork() {
-        super.didFork()
-
-        errorCallCount += 1
+        forkCallCount += 1
     }
 
     override func didReceiveError(_ error: Error) {
-        super.didReceiveError(error)
-
         errorCallCount += 1
     }
 }