system: refactor propagate execute(on:)

Author: twittemb

Examples/Examples/CounterApp/System/CounterApp+System.swift

@@ -5,6 +5,7 @@
 //  Created by Thibault Wittemberg on 2021-01-12.
 //
 
+import Dispatch
 import Feedbacks
 
 // define a namespace for this app's system
@@ -55,5 +56,5 @@ extension CounterApp.System {
                 On(CounterApp.Events.Decrease.self, transitionTo: CounterApp.States.Decreasing(counter: state.counter.decrease(), isPaused: false))
             }
         }
-    }
+    }.execute(on: DispatchQueue(label: "Counter Queue"))
 }

Examples/Examples/GiphyApp/Features/GifDetail/System/GifDetail+System.swift

@@ -32,10 +32,7 @@ extension GifDetail.System {
 
             Feedbacks {
                 Feedback(on: GifDetail.States.Loading.self, strategy: .cancelOnNewState, perform: loadSideEffect)
-                    .execute(on: DispatchQueue(label: "Load Gif Queue"))
-
                 Feedback(on: GifDetail.States.TogglingFavorite.self, strategy: .cancelOnNewState, perform: toggleFavoriteSideEffect)
-                    .execute(on: DispatchQueue(label: "Toggle Favorite Queue"))
             }
             .onStateReceived {
                 print("GifDetail: New state has been received: \($0)")
@@ -64,6 +61,6 @@ extension GifDetail.System {
                     On(GifDetail.Events.LoadingHasFailed.self, transitionTo: GifDetail.States.Failed())
                 }
             }
-        }
+        }.execute(on: DispatchQueue(label: "Load Gif Queue"))
     }
 }

Examples/Examples/GiphyApp/Features/GifList/System/GifList+System.swift

@@ -26,7 +26,6 @@ extension GifList.System {
 
             Feedbacks {
                 Feedback(on: GifList.States.Loading.self , strategy: .cancelOnNewState, perform: loadSideEffect)
-                    .execute(on: DispatchQueue(label: "Load Gifs Queue"))
             }
             .onStateReceived {
                 print("GifList: New state has been received: \($0)")
@@ -64,6 +63,6 @@ extension GifList.System {
                     On(GifList.Events.Refresh.self, transitionTo: GifList.States.Loading())
                 }
             }
-        }
+        }.execute(on: DispatchQueue(label: "Load Gifs Queue"))
     }
 }

Sources/Feedbacks/System/Feedback.swift renamed to Sources/Feedbacks/Feedbacks/Feedback.swift

@@ -18,37 +18,26 @@ import Foundation
 public class System {
     let initialState: InitialState
     var feedbacks: Feedbacks
-    public let transitions: Transitions
-    var scheduledStream: (AnyPublisher<Event, Never>) -> AnyPublisher<Event, Never>
+    public private(set) var transitions: Transitions
 
     private var subscriptions = [AnyCancellable]()
 
-    static let defaultQueue = DispatchQueue(label: "Feedbacks.System.\(UUID().uuidString)")
-
     /// Builds a System based on its three components: an initial state, some feedbacks, a state machine
     /// By default, the System will be executed an a serial background queue. This can be altered thanks to the `.execute(on:)` modifier.
     /// - Parameter components: the three components of the System
     public convenience init(@SystemBuilder _ components: () -> (InitialState, Feedbacks, Transitions)) {
         let (initialState, feedbacks, transitions) = System.decode(builder: components)
         self.init(initialState: initialState,
                   feedbacks: feedbacks,
-                  transitions: transitions,
-                  scheduledStream: { (events: AnyPublisher<Event, Never>) in
-                    events
-                        .subscribe(on: System.defaultQueue)
-                        .receive(on: System.defaultQueue)
-                        .eraseToAnyPublisher()
-                  })
+                  transitions: transitions)
     }
 
     init(initialState: InitialState,
          feedbacks: Feedbacks,
-         transitions: Transitions,
-         scheduledStream: @escaping (AnyPublisher<Event, Never>) -> AnyPublisher<Event, Never>) {
+         transitions: Transitions) {
         self.initialState = initialState
         self.feedbacks = feedbacks
         self.transitions = transitions
-        self.scheduledStream = scheduledStream
     }
 
     static func decode(builder system: () -> (InitialState, Feedbacks, Transitions)) -> (InitialState, Feedbacks, Transitions) {
@@ -62,16 +51,14 @@ public extension System {
     /// Once this stream has been subscribed to, the initial state is given as an input to the feedbacks.
     /// Then the feedbacks can publish event that will trigger some transitions, generating a new state, and so on and so forth.
     var stream: AnyPublisher<State, Never> {
-        Deferred<AnyPublisher<State, Never>> { [initialState, feedbacks, transitions, scheduledStream] in
+        Deferred<AnyPublisher<State, Never>> { [initialState, feedbacks, transitions] in
             let currentState = ReplaySubject<State, Never>(bufferSize: 1)
 
             // merging all the effects into one event stream
             let stateInputStream = currentState.eraseToAnyPublisher()
             let eventStream = feedbacks.eventStream(stateInputStream)
-            let scheduledEventStream = scheduledStream(eventStream)
 
-            return scheduledEventStream
-                .scan(initialState.value, transitions.reducer)
+            return transitions.scheduledReducer(initialState.value, eventStream)
                 .prepend(initialState.value)
                 .handleEvents(receiveOutput: currentState.send)
                 .eraseToAnyPublisher()
@@ -85,6 +72,14 @@ public extension System {
         self.stream.sink(receiveValue: { _ in }).store(in: &self.subscriptions)
         return self
     }
+
+    /// Subscribes to the state stream and store the cancellable in the System.
+    /// The subscription will be canceled once the System is deinit.
+    @discardableResult
+    func run<SchedulerType: Scheduler>(subscribeOn scheduler: SchedulerType) -> Self {
+        self.stream.subscribe(on: scheduler).sink(receiveValue: { _ in }).store(in: &self.subscriptions)
+        return self
+    }
 }
 
 // MARK: modifiers
@@ -95,13 +90,8 @@ public extension System {
     /// - Parameter scheduler: the scheduler on which to execute the System
     /// - Returns: The newly scheduled System
     func execute<SchedulerType: Scheduler>(on scheduler: SchedulerType) -> Self {
-        self.scheduledStream = { events in
-            events
-                .subscribe(on: scheduler)
-                .receive(on: scheduler)
-                .eraseToAnyPublisher()
-        }
-
+        self.feedbacks = self.feedbacks.execute(on: scheduler)
+        self.transitions = self.transitions.execute(on: scheduler)
         return self
     }
 

Sources/Feedbacks/System/Feedbacks.swift renamed to Sources/Feedbacks/Feedbacks/Feedbacks.swift

@@ -42,31 +42,22 @@ public class UISystem<PublishedState: State>: System, ObservableObject {
 
         self.init(initialState: initialState,
                   feedbacks: feedbacks,
-                  transitions: transitions,
-                  scheduledStream: { (events: AnyPublisher<Event, Never>) in
-                    events
-                        .subscribe(on: System.defaultQueue)
-                        .receive(on: System.defaultQueue)
-                        .eraseToAnyPublisher()
-                  })
+                  transitions: transitions)
     }
 
     convenience init(system: System) where PublishedState == RawState {
         self.init(initialState: system.initialState,
                   feedbacks: system.feedbacks,
-                  transitions: system.transitions,
-                  scheduledStream: system.scheduledStream)
+                  transitions: system.transitions)
     }
 
     override init(initialState: InitialState,
          feedbacks: Feedbacks,
-         transitions: Transitions,
-         scheduledStream: @escaping (AnyPublisher<Event, Never>) -> AnyPublisher<Event, Never>) where PublishedState == RawState {
+         transitions: Transitions) where PublishedState == RawState {
         self.state = RawState(state: initialState.value)
         super.init(initialState: initialState,
                    feedbacks: feedbacks,
-                   transitions: transitions,
-                   scheduledStream: scheduledStream)
+                   transitions: transitions)
 
         let stateFeedback = Self.makeStatePublishingFeedback(publishingFunction: { [weak self] in
             self?.state = $0
@@ -76,7 +67,9 @@ public class UISystem<PublishedState: State>: System, ObservableObject {
             events.eraseToAnyPublisher()
         }
 
-        self.feedbacks = self.feedbacks.add(feedback: stateFeedback).add(feedback: eventFeedback)
+        self.feedbacks = self.feedbacks
+            .add(feedback: stateFeedback)
+            .add(feedback: eventFeedback)
     }
 
     /// Creates a UISystem based on the 3 components of a System (initial state, feedbacks, state machine) and a View State factory function
@@ -85,7 +78,7 @@ public class UISystem<PublishedState: State>: System, ObservableObject {
     ///   - system: the 3 components of the System
     public convenience init(
         viewStateFactory: @escaping (State) -> PublishedState,
-                            @SystemBuilder _ components: () -> (InitialState, Feedbacks, Transitions)
+        @SystemBuilder _ components: () -> (InitialState, Feedbacks, Transitions)
     ) where PublishedState: ViewState {
         self.init(viewStateFactory: viewStateFactory,
                   on: DispatchQueue(label: "Feedbacks.UISystem.\(UUID().uuidString)"),
@@ -99,21 +92,15 @@ public class UISystem<PublishedState: State>: System, ObservableObject {
     ///   - system: the 3 components of the System
     public convenience init<SchedulerType: Scheduler>(
         viewStateFactory: @escaping (State) -> PublishedState,
-                                                      on scheduler: SchedulerType,
-                                                      @SystemBuilder _ components: () -> (InitialState, Feedbacks, Transitions)
+        on scheduler: SchedulerType,
+        @SystemBuilder _ components: () -> (InitialState, Feedbacks, Transitions)
     ) where PublishedState: ViewState {
         let (initialState, feedbacks, transitions) = System.decode(builder: components)
 
         self.init(viewStateFactory: viewStateFactory,
                   initialState: initialState,
                   feedbacks: feedbacks,
                   transitions: transitions,
-                  scheduledStream: { (events: AnyPublisher<Event, Never>) in
-                    events
-                        .subscribe(on: System.defaultQueue)
-                        .receive(on: System.defaultQueue)
-                        .eraseToAnyPublisher()
-                  },
                   viewStateScheduler: scheduler)
     }
 
@@ -131,7 +118,6 @@ public class UISystem<PublishedState: State>: System, ObservableObject {
                   initialState: system.initialState,
                   feedbacks: system.feedbacks,
                   transitions: system.transitions,
-                  scheduledStream: system.scheduledStream,
                   viewStateScheduler: scheduler)
     }
 
@@ -140,14 +126,13 @@ public class UISystem<PublishedState: State>: System, ObservableObject {
         initialState: InitialState,
         feedbacks: Feedbacks,
         transitions: Transitions,
-        scheduledStream: @escaping (AnyPublisher<Event, Never>) -> AnyPublisher<Event, Never>,
         viewStateScheduler: ViewStateSchedulerType
     ) where PublishedState: ViewState {
         // since the initial view state is calculated asynchronously on the viewStateScheduler when the system is started
         // we set it to a initial undefined value
         self.state = PublishedState.undefined
 
-        super.init(initialState: initialState, feedbacks: feedbacks, transitions: transitions, scheduledStream: scheduledStream)
+        super.init(initialState: initialState, feedbacks: feedbacks, transitions: transitions)
 
         let stateFeedback = Self.makeStatePublishingFeedback(
             viewStateFactory: viewStateFactory,

Sources/Feedbacks/System/SideEffect.swift renamed to Sources/Feedbacks/Feedbacks/SideEffect.swift

@@ -4,13 +4,15 @@
 //
 //  Created by Thibault Wittemberg on 2020-12-23.
 //
+import Combine
 
 /// Represents a series of Transitions that drive a State Machine.
 public struct Transitions {
     let transitions: [From]
 
     /// the reducer computed from the state machine's transitions
     public let reducer: (State, Event) -> State
+    public let scheduledReducer: (State, AnyPublisher<Event, Never>) -> AnyPublisher<State, Never>
 
     /// - Parameter transitions: the individual transitions composing the state machine
     /// Transitions {
@@ -31,11 +33,29 @@ public struct Transitions {
 
     init(transitions: [From]) {
         self.transitions = transitions
-        let transitionsForStates = self.transitions.reduce(into: [AnyHashable: (State) -> [AnyHashable: (Event) -> State?]]()) { accumulator, from in
+        self.reducer = Transitions.makeReducer(transitions: transitions)
+        self.scheduledReducer = { [reducer] initialState, events in
+            return events
+                .scan(initialState, reducer)
+                .eraseToAnyPublisher()
+        }
+    }
+
+    init(transitions: [From],
+         reducer: @escaping (State, Event) -> State,
+         scheduledReducer: @escaping (State, AnyPublisher<Event, Never>) -> AnyPublisher<State, Never>) {
+        self.transitions = transitions
+        self.reducer = reducer
+        self.scheduledReducer = scheduledReducer
+    }
+
+    static func makeReducer(transitions: [From]) -> (State, Event) -> State {
+        let transitionsForStates = transitions.reduce(into: [AnyHashable: (State) -> [AnyHashable: (Event) -> State?]]()) { accumulator, from in
             let existingTranstionsForState = accumulator[from.id]
             accumulator[from.id] = { state in from.computeTransitionsForEvents(for: state, existingTranstionsForState: existingTranstionsForState) }
         }
-        self.reducer = { state, event -> State in
+
+        return { state, event -> State in
             if
                 let transitionsForState = transitionsForStates[state.instanceId],
                 let transitionForEvent = transitionsForState(state)[event.instanceId],
@@ -69,4 +89,16 @@ public extension Transitions {
     func disable(_ disabled: @escaping () -> Bool) -> Self {
         Transitions(transitions: self.transitions.map { $0.disable(disabled) })
     }
+
+    /// Alter the scheduler on which the Transitions run. If no schedulers are
+    /// set for the Transitions, then they will be executed on the current scheduler.
+    /// - Parameter scheduler: the scheduler on which to execute the Transitions
+    /// - Returns: The newly scheduled Transitions
+    func execute<SchedulerType: Scheduler>(on scheduler: SchedulerType) -> Self {
+        let newScheduledReducer: (State, AnyPublisher<Event, Never>) -> AnyPublisher<State, Never> = { [scheduledReducer] initialState, events in
+            return scheduledReducer(initialState, events.receive(on: scheduler).eraseToAnyPublisher())
+        }
+
+        return Transitions(transitions: self.transitions, reducer: self.reducer, scheduledReducer: newScheduledReducer)
+    }
 }

Sources/Feedbacks/System/System.swift

@@ -228,6 +228,33 @@ extension FeedbackTests {
 
         cancellable.cancel()
     }
+
+    func testExecute_execute_sideEffect_on_the_inner_scheduler() {
+        let exp = expectation(description: "Feedback.execute(on:)")
+
+        let expectedQueue = UUID().uuidString
+        var receivedQueue = ""
+
+        // Given: a side effect recording its execution queue
+        let spySideEffect: (MockStateA) -> AnyPublisher<Event, Never> = { state in
+            receivedQueue = DispatchQueue.currentLabel
+            return Just(MockEventA(value: 1)).eraseToAnyPublisher()
+        }
+
+        // When: making a Feedback of it, and executing it on the expected Queue
+        let sut = Feedback(on: MockStateA.self, strategy: .continueOnNewState, perform: spySideEffect)
+            .execute(on: DispatchQueue(label: expectedQueue))
+            .execute(on: DispatchQueue(label: UUID().uuidString))
+
+        let cancellable = sut.sideEffect(Just(MockStateA(value: 1)).eraseToAnyPublisher()).sink{ _ in exp.fulfill() }
+
+        waitForExpectations(timeout: 0.5)
+
+        // Then: the side effect is executed on the inner scheduler
+        XCTAssertEqual(receivedQueue, expectedQueue)
+
+        cancellable.cancel()
+    }
 }
 
 // MARK: tests for Feedback.disable(:)