Replay-1 semantic to state producer in setup closure.

Author: andersio

Loop/FeedbackLoop.swift

@@ -6,37 +6,25 @@ public final class FeedbackLoop<State, Event> {
     private let token: Lifetime.Token
 
     public var producer: SignalProducer<State, Never> {
-        SignalProducer { observer, lifetime in
-            self.floodgate.withValue { initial, hasStarted -> Void in
-                if hasStarted {
-                    // The feedback loop has started already, so the initial value has to be manually delivered.
-                    // Uninitialized feedback loop that does not start immediately will emit the initial state
-                    // when `start()` is called.
-                    observer.send(value: initial)
-                }
-
-                lifetime += self.floodgate.stateDidChange.observe(observer)
-            }
-        }
+        floodgate.producer
     }
 
+    private let feedbacks: [Feedback]
+
     public init(
         initial: State,
         reduce: @escaping (inout State, Event) -> Void,
         feedbacks: [Feedback]
     ) {
         (lifetime, token) = Lifetime.make()
-        floodgate = Floodgate<State, Event>(state: initial, reducer: reduce)
-        lifetime.observeEnded(floodgate.dispose)
+        self.floodgate = Floodgate<State, Event>(state: initial, reducer: reduce)
+        self.feedbacks = feedbacks
 
-        for feedback in feedbacks {
-            lifetime += feedback
-                .events(floodgate.stateDidChange.producer, floodgate)
-        }
+        lifetime.observeEnded(floodgate.dispose)
     }
 
     public func start() {
-        floodgate.bootstrap()
+        floodgate.bootstrap(with: feedbacks)
     }
 
     public func stop() {
@@ -71,7 +59,58 @@ extension FeedbackLoop {
         /// `Feedback(skippingRepeated:effects:)` enqueues events inside `flatMap(.latest)`, so that unprocessed events
         /// are automatically removed when the inner producer has switched.
         ///
-        /// - important: The `state` producer provided to the setup closure **does not** replay the current state.
+        /// ## State producer in the `setup` closure
+        /// The setup closure provides you a `state` producer — it replays the latest state at starting time, and then
+        /// publishes all state changes.
+        ///
+        /// Loop guarantees only that this `state` producer is **eventually consistent** with events emitted by your
+        /// feedback. This means you should not make any strong assumptions on events you enqueued being immediately
+        /// reflected by `state`.
+        ///
+        /// For example, if you start the `state` producer again, synchronously after enqueuing an event, the event
+        /// may not have been processed yet, and therefore the assertion would fail:
+        /// ```swift
+        /// Feedback { state, output in
+        ///     state
+        ///        .filter { $0.apples.isEmpty == false }
+        ///        .map(value: Event.eatAllApples)
+        ///        .take(first: 1)
+        ///        .concat(
+        ///            state
+        ///                .take(first: 1)
+        ///                .on(value: { state in
+        ///                    guard state.apples.isEmpty else { return }
+        ///
+        ///                    // ❌🙅‍♀️ No guarantee that this is true.
+        ///                    fatalError("It should have eaten all the apples!")
+        ///                })
+        ///        )
+        ///        .enqueue(to: output)
+        /// }
+        /// ```
+        ///
+        /// You can however expect it to be eventually consistent:
+        /// ```swift
+        /// Feedback { state, output in
+        ///     state
+        ///        .filter { $0.apples.isEmpty == false }
+        ///        .map(value: Event.eatAllApples)
+        ///        .take(first: 1)
+        ///        .concat(
+        ///            state
+        ///                .filter { $0.apples.isEmpty }
+        ///                .take(first: 1)
+        ///                .on(value: { state in
+        ///                    guard state.apples.isEmpty else { return }
+        ///
+        ///                    // ✅👍 We would eventually observe this, when the loop event queue
+        ///                    //      has caught up with `.eatAppleApples` we enqueued earlier.
+        ///                    fatalError("It should have eaten all the apples!")
+        ///                })
+        ///        )
+        ///        .enqueue(to: output)
+        /// }
+        /// ```
         ///
         /// - parameters:
         ///   - setup: The setup closure to construct a data flow producing events in respond to changes from `state`,

Loop/Floodgate.swift

@@ -13,28 +13,42 @@ final class Floodgate<State, Event>: FeedbackEventConsumer<Event> {
 
     let (stateDidChange, changeObserver) = Signal<State, Never>.pipe()
 
+    /// Replay the current value, and then publish the subsequent changes.
+    var producer: SignalProducer<State, Never> {
+        SignalProducer { observer, lifetime in
+            self.withValue { initial, hasStarted -> Void in
+                observer.send(value: initial)
+                lifetime += self.stateDidChange.observe(observer)
+            }
+        }
+    }
+
     private let reducerLock = NSLock()
     private var state: State
     private var hasStarted = false
 
     private let queue = Atomic(QueueState())
     private let reducer: (inout State, Event) -> Void
+    private let feedbackDisposables = CompositeDisposable()
 
     init(state: State, reducer: @escaping (inout State, Event) -> Void) {
         self.state = state
         self.reducer = reducer
     }
 
-    func bootstrap() {
-        reducerLock.lock()
-        defer { reducerLock.unlock() }
-
-        guard !hasStarted else { return }
+    deinit {
+        dispose()
+    }
 
-        hasStarted = true
+    func bootstrap(with feedbacks: [FeedbackLoop<State, Event>.Feedback]) {
+        for feedback in feedbacks {
+            // Pass `producer` which has replay-1 semantic.
+            feedbackDisposables += feedback.events(producer, self)
+        }
 
-        changeObserver.send(value: state)
-        drainEvents()
+        reducerLock.perform {
+            drainEvents()
+        }
     }
 
     override func process(_ event: Event, for token: Token) {
@@ -77,8 +91,14 @@ final class Floodgate<State, Event>: FeedbackEventConsumer<Event> {
     }
 
     func dispose() {
-        queue.modify {
+        let shouldDisposeFeedbacks: Bool = queue.modify {
+            let old = $0.isOuterLifetimeEnded
             $0.isOuterLifetimeEnded = true
+            return old == false
+        }
+
+        if shouldDisposeFeedbacks {
+            feedbackDisposables.dispose()
         }
     }
 

LoopTests/FeedbackLoopSystemTests.swift

@@ -159,10 +159,10 @@ class FeedbackLoopSystemTests: XCTestCase {
         var values: [String] = []
         system
             .skipRepeats()
-            .take(first: 2)
+            .take(first: 3)
             .startWithValues { values.append($0) }
 
-        expect(values) == ["initial", "initial_a"]
+        expect(values) == ["initial", "initial_a", "initial_a_a"]
         expect(startCount) == 2
     }
 
@@ -304,4 +304,35 @@ class FeedbackLoopSystemTests: XCTestCase {
 
         expect(results).toEventually(equal([0, 1, 3, 6, 6, 7, 9, 12]))
     }
+
+    func test_feedback_state_producer_replays_latest_value() {
+        let system = SignalProducer<Int, Never>.feedbackLoop(
+            initial: 0,
+            reduce: { (state: inout Int, event: Int) in
+                state += event
+            },
+            feedbacks: [
+                FeedbackLoop.Feedback { state, output in
+                    state
+                        .take(first: 1)
+                        .then(SignalProducer(value: 2))
+                        .concat(
+                            // `state` is NOT GUARANTEED to reflect events emitted earlier in the producer chain.
+                            state
+                                .take(first: 3)
+                                .map { $0 + 1000 }
+                        )
+                        .enqueue(to: output)
+                }
+            ]
+        )
+
+        var results: [Int] = []
+
+        system.startWithValues { value in
+            results.append(value)
+        }
+
+        expect(results) == [0, 2, 1002, 2004, 4006]
+    }
 }