Improve concurrent state mutation

Author: twittemb

RxReduce/Action.swift

@@ -35,6 +35,6 @@ extension Array: Action where Element == Action {
 // MARK: - Default implementation of toAsync: Observable<Action> is also an Action
 extension Observable: Action where Element == Action {
     public func toAsync () -> Observable<Action> {
-        return self.map { $0 as Action }
+        return self
     }
 }

RxReduce/Store.swift

@@ -20,10 +20,7 @@ public final class Store<State: Equatable> {
     private let stateSubject: BehaviorRelay<State>
     private var neededReducersPerSubState = [String: Int]()
     private var reducers = ContiguousArray<Reducer>()
-    private let serialDispatchScheduler: SerialDispatchQueueScheduler = {
-        let serialQueue = DispatchQueue(label: "com.rxswiftcommunity.rxreduce.serialqueue")
-        return SerialDispatchQueueScheduler.init(queue: serialQueue, internalSerialQueueName: "com.rxswiftcommunity.rxreduce.serialscheduler")
-    }()
+    private let serialQueue = DispatchQueue(label: "com.rxswiftcommunity.rxreduce.serialqueue")
 
     /// The global State is exposed via an Observable, just like some kind of "middleware".
     /// This global State will trigger a new value after a dispatch(action) has triggered a "onNext" event.
@@ -102,15 +99,18 @@ public final class Store<State: Equatable> {
         return action
             .toAsync()
             .map { [unowned self] (action) -> State in
-                return self.reducers.reduce(self.stateSubject.value, { (currentState, reducer) -> State in
-                    return reducer(currentState, action)
-                })
+
+                self.serialQueue.sync {
+                    let newState = self.reducers.reduce(self.stateSubject.value, { (currentState, reducer) -> State in
+                        return reducer(currentState, action)
+                    })
+
+                    self.stateSubject.accept(newState)
+                }
+
+                return self.stateSubject.value
             }
-            .do(onNext: { [unowned self] (newState) in
-                self.stateSubject.accept(newState)
-            })
             .distinctUntilChanged()
-            .subscribeOn(self.serialDispatchScheduler)
     }
 
     /// Dispatches an Action to the registered Mutators but instead of

RxReduceTests/StoreTests.swift

@@ -137,4 +137,61 @@ final class StoreTests: XCTestCase {
 
         waitForExpectations(timeout: 1)
     }
+
+    func testStoreSchedulers () throws {
+        // create 2 queues: 1 for the observeOn part of the Rx stream, and 1 for the subscriveOn part
+        let subscribeOnScheduler: OperationQueueScheduler = {
+            let queue = OperationQueue()
+            queue.name = "SUBSCRIBE_ON_QUEUE"
+            queue.maxConcurrentOperationCount = 1
+            return OperationQueueScheduler(operationQueue: queue)
+        }()
+        let observeOnScheduler: OperationQueueScheduler = {
+            let queue = OperationQueue()
+            queue.name = "OBSERVE_ON_QUEUE"
+            queue.maxConcurrentOperationCount = 1
+            return OperationQueueScheduler(operationQueue: queue)
+        }()
+
+        // Given
+        let exp = expectation(description: "Queue expectation")
+        exp.expectedFulfillmentCount = 3
+        let actionObservable = Observable<Action>.just (IncreaseAction(increment: 10))
+
+        // When
+        actionObservable
+            .map { action -> DecreaseAction in
+                if let  queueName = OperationQueue.current?.name!,
+                    queueName ==  "SUBSCRIBE_ON_QUEUE" {
+                    exp.fulfill()
+                } else {
+                    XCTFail("Subscribe on Wrong Queue")
+                }
+                return DecreaseAction(decrement: 10)
+            }
+            .flatMap{ (action) -> Observable<CounterState> in
+
+                if let  queueName = OperationQueue.current?.name!,
+                    queueName ==  "SUBSCRIBE_ON_QUEUE" {
+                    exp.fulfill()
+                } else {
+                    XCTFail("Subscribe on Wrong Queue")
+                }
+                return self.store.dispatch(action: action, focusingOn: { $0.counterState })
+            }
+            .subscribeOn(subscribeOnScheduler)
+            .observeOn(observeOnScheduler)
+            .subscribe(onNext: { counterState in
+
+                if let  queueName = OperationQueue.current?.name!,
+                    queueName ==  "OBSERVE_ON_QUEUE" {
+                    exp.fulfill()
+                } else {
+                    XCTFail("Observe on Wrong Queue")
+                }
+            }).disposed(by: self.disposeBag)
+
+        // Then
+        waitForExpectations(timeout: 1)
+    }
 }