Merge pull request #131 from ReactorKit/scheduler-mutate-reduce

Ensure that mutate and reduce are executed in the same serial scheduler
ReactorKit · Oct 24, 2019 · f95a197 · f95a197
2 parents 5c49334 + dc8f2fe
commit f95a197
Show file tree

Hide file tree

Showing 3 changed files with 58 additions and 13 deletions.
diff --git a/Sources/ReactorKit/Reactor.swift b/Sources/ReactorKit/Reactor.swift
@@ -128,16 +128,15 @@ extension Reactor {
   }
 
   public func createStateStream() -> Observable<State> {
-    let action = self._action.asObservable()
+    let action = self._action.observeOn(self.scheduler)
     let transformedAction = self.transform(action: action)
     let mutation = transformedAction
       .flatMap { [weak self] action -> Observable<Mutation> in
         guard let `self` = self else { return .empty() }
         return self.mutate(action: action).catchError { _ in .empty() }
       }
     let transformedMutation = self.transform(mutation: mutation)
-    let state = transformedMutation
-      .observeOn(self.scheduler)
+    let state = transformedMutation         
       .scan(self.initialState) { [weak self] state, mutation -> State in
         guard let `self` = self else { return state }
         return self.reduce(state: state, mutation: mutation)

diff --git a/Tests/ReactorKitTests/ReactorSchedulerTests.swift b/Tests/ReactorKitTests/ReactorSchedulerTests.swift
@@ -57,10 +57,6 @@ final class ReactorSchedulerTests: XCTestCase {
       let initialState: State = State()
       let scheduler: ImmediateSchedulerType = SerialDispatchQueueScheduler(qos: .default)
 
-      func mutate(action: Action) -> Observable<Mutation> {
-        return Observable.just(Void()).observeOn(MainScheduler.instance)
-      }
-
       func reduce(state: State, mutation: Mutation) -> State {
         var newState = state
         let currentThread = Thread.current

diff --git a/Tests/ReactorKitTests/ReactorTests.swift b/Tests/ReactorKitTests/ReactorTests.swift
@@ -14,10 +14,60 @@ final class ReactorTests: XCTestCase {
     test.assert(reactor.state) { events in
       XCTAssertEqual(events.elements.count, 2)
       XCTAssertEqual(events.elements[0], ["transformedState"]) // initial state
-      XCTAssertEqual(events.elements[1], ["action", "transformedAction", "mutation", "transformedMutation", "transformedState"])
+      XCTAssertEqual(events.elements[1], ["action", "transformedAction", "mutation", "transformedMutation", "reduce", "transformedState"])
     }
   }
 
+  func testReduceIsExecutedRightAfterMutation() {
+    final class MyReactor: Reactor {
+      enum Action {
+        case append([String])
+      }
+
+      enum Mutation {
+        case setCharacters([String])
+      }
+
+      struct State {
+        var characters: [String] = []
+      }
+
+      let initialState = State()
+
+      func mutate(action: Action) -> Observable<Mutation> {
+        switch action {
+        case let .append(characters):
+          let sources: [Observable<Mutation>] = characters.map { character in
+            Observable<Mutation>.create { [weak self] observer in
+              if let self = self {
+                let newCharacters = self.currentState.characters + [character]
+                observer.onNext(Mutation.setCharacters(newCharacters))
+              }
+              observer.onCompleted()
+              return Disposables.create()
+            }
+          }
+          return Observable.concat(sources)
+        }
+      }
+
+      func reduce(state: State, mutation: Mutation) -> State {
+        var newState = state
+        switch mutation {
+        case let .setCharacters(newCharacters):
+          newState.characters = newCharacters
+        }
+        return newState
+      }
+    }
+
+    let reactor = MyReactor()
+    reactor.action.onNext(.append(["a", "b"]))
+    reactor.action.onNext(.append(["c"]))
+    reactor.action.onNext(.append(["d", "e", "f"]))
+    XCTAssertEqual(reactor.currentState.characters, ["a", "b", "c", "d", "e", "f"])
+  }
+
   func testStateReplayCurrentState() {
     let test = RxExpect()
     let reactor = test.retain(CounterReactor())
@@ -34,13 +84,13 @@ final class ReactorTests: XCTestCase {
     let reactor = TestReactor()
     _ = reactor.state
     reactor.action.onNext(["action"])
-    XCTAssertEqual(reactor.currentState, ["action", "transformedAction", "mutation", "transformedMutation", "transformedState"])
+    XCTAssertEqual(reactor.currentState, ["action", "transformedAction", "mutation", "transformedMutation", "reduce", "transformedState"])
   }
 
   func testCurrentState_stateIsCreatedWhenAccessAction() {
     let reactor = TestReactor()
     reactor.action.onNext(["action"])
-    XCTAssertEqual(reactor.currentState, ["action", "transformedAction", "mutation", "transformedMutation", "transformedState"])
+    XCTAssertEqual(reactor.currentState, ["action", "transformedAction", "mutation", "transformedMutation", "reduce", "transformedState"])
   }
 
   func testStreamIgnoresErrorFromAction() {
@@ -281,12 +331,12 @@ private final class TestReactor: Reactor {
     return mutation.map { $0 + ["transformedMutation"] }
   }
 
-  // 4. [] + ["action", "transformedAction", "mutation", "transformedMutation"]
+  // 4. [] + ["action", "transformedAction", "mutation", "transformedMutation"] + ["reduce"]
   func reduce(state: State, mutation: Mutation) -> State {
-    return state + mutation
+    return state + mutation + ["reduce"]
   }
 
-  // 5. ["action", "transformedAction", "mutation", "transformedMutation"] + ["transformedState"]
+  // 5. ["action", "transformedAction", "mutation", "transformedMutation", "reduce"] + ["transformedState"]
   func transform(state: Observable<State>) -> Observable<State> {
     return state.map { $0 + ["transformedState"] }
   }