Implement Feedback.autoconnect operator

- Implement a new `autoconnect` operator  on `Feedback` that disposes
and restores its "parent" Feedback observation according to a predicate
on `State`.

Author: p4checo

Loop/Floodgate.swift

@@ -34,8 +34,7 @@ final class Floodgate<State, Event>: FeedbackEventConsumer<Event> {
 
     private let queue = Atomic(QueueState())
     private let reducer: (inout State, Event) -> Void
-    private var feedbacks: [Loop<State, Event>.Feedback] = []
-    private var feedbackDisposables = CompositeDisposable()
+    private let feedbackDisposables = CompositeDisposable()
 
     init(state: State, reducer: @escaping (inout State, Event) -> Void) {
         self.state = state
@@ -47,12 +46,6 @@ final class Floodgate<State, Event>: FeedbackEventConsumer<Event> {
     }
 
     func bootstrap(with feedbacks: [Loop<State, Event>.Feedback]) {
-        self.feedbacks = feedbacks
-
-        plugFeedbacks()
-    }
-
-    func plugFeedbacks() {
         for feedback in feedbacks {
             // Pass `producer` which has replay-1 semantic.
             feedbackDisposables += feedback.events(
@@ -67,11 +60,6 @@ final class Floodgate<State, Event>: FeedbackEventConsumer<Event> {
         }
     }
 
-    func unplugFeedbacks() {
-        feedbackDisposables.dispose()
-        feedbackDisposables = CompositeDisposable()
-    }
-
     override func process(_ event: Event, for token: Token) {
         enqueue(event, for: token)
 

Loop/LoopBox.swift

@@ -44,14 +44,6 @@ internal class ScopedLoopBox<RootState, RootEvent, ScopedState, ScopedEvent>: Lo
             event: { [eventTransform] in eventTransform(event($0)) }
         )
     }
-
-    override func pause() {
-        root.pause()
-    }
-
-    override func resume() {
-        root.resume()
-    }
 }
 
 internal class RootLoopBox<State, Event>: LoopBoxBase<State, Event> {
@@ -91,14 +83,6 @@ internal class RootLoopBox<State, Event>: LoopBoxBase<State, Event> {
         ScopedLoopBox(root: self, value: scope, event: event)
     }
 
-    override func pause() {
-        floodgate.unplugFeedbacks()
-    }
-
-    override func resume() {
-        floodgate.plugFeedbacks()
-    }
-
     func start(with feedbacks: [Loop<State, Event>.Feedback]) {
         floodgate.bootstrap(with: feedbacks + [input.feedback])
     }
@@ -131,10 +115,6 @@ internal class LoopBoxBase<State, Event> {
     ) -> LoopBoxBase<S, E> {
         subclassMustImplement()
     }
-
-    func pause() { subclassMustImplement() }
-
-    func resume() { subclassMustImplement() }
 }
 
 @inline(never)

Loop/Public/FeedbackLoop.swift

@@ -453,6 +453,17 @@ extension Loop {
             }
         }
 
+        public func autoconnect(followingChangesIn predicate: @escaping (State) -> Bool) -> Feedback {
+            return Feedback { state, consumer in
+                self.events(
+                    state
+                        .skipRepeats { predicate($0.0) == predicate($1.0) }
+                        .flatMap(.latest) { predicate($0.0) ? state.filter { predicate($0.0) } : .empty },
+                    consumer
+                )
+            }
+        }
+
         public static func combine(_ feedbacks: Loop<State, Event>.Feedback...) -> Feedback {
             return Feedback { state, consumer in
                 feedbacks.map { feedback in

Loop/Public/Loop.swift

@@ -69,14 +69,6 @@ public final class Loop<State, Event> {
             box: box.scoped(to: { $0 }, event: { _ in fatalError() })
         )
     }
-
-    public func pause() {
-        box.pause()
-    }
-
-    public func resume() {
-        box.resume()
-    }
 }
 
 extension Loop {

LoopTests/FeedbackVariantTests.swift

@@ -174,4 +174,39 @@ class FeedbackVariantTests: XCTestCase {
         loop.send("world")
         expect(hasCancelled) == true
     }
+
+    func test_autoconnect_disposes_and_restores_flows_according_to_predicate() {
+
+        let loop = Loop<String, String>(
+            initial: "",
+            reducer: { content, string in
+                content = string
+            },
+            feedbacks: [
+                Loop<String, String>.Feedback
+                    .init(
+                        lensing: { $0.hasPrefix("hello") ? $0 : nil },
+                        effects: { SignalProducer(value: $0.uppercased()) }
+                    )
+                    .autoconnect(followingChangesIn: { $0.contains("disconnect") == false })
+            ]
+        )
+
+        expect(loop.box._current) == ""
+
+        // This should trigger an uppercased event
+        loop.send("hello1")
+        expect(loop.box._current) == "HELLO1"
+
+        loop.send("hello2")
+        expect(loop.box._current) == "HELLO2"
+
+        // This should lead to feedabck "disconnection", which in turn should cancel and disable the uppercasing effect.
+        loop.send("hello disconnect")
+        expect(loop.box._current) == "hello disconnect"
+
+        // This should lead feedback "connection", which in turn should reestablish the uppercasing effect.
+        loop.send("hello3")
+        expect(loop.box._current) == "HELLO3"
+    }
 }