Add Effect.until for cancellation

Sources/Automaton.swift

@@ -9,7 +9,7 @@ public final class Automaton<State, Input>
 
     /// Transducer (input & output) mapping with `Effect<Input>` (additional effect) as output,
     /// which may emit next input values for continuous state-transitions.
-    public typealias EffectMapping<Queue> = (State, Input) -> (State, Effect<Input, Queue>?)?
+    public typealias EffectMapping<Queue> = (State, Input) -> (State, Effect<Input, State, Queue>?)?
         where Queue: EffectQueueProtocol
 
     /// `Reply` signal that notifies either `.success` or `.failure` of state-transition on every input.
@@ -37,7 +37,7 @@ public final class Automaton<State, Input>
         self.init(
             state: initialState,
             inputs: inputSignal,
-            mapping: { mapping($0, $1).map { ($0, Effect<Input, Never>?.none) } }
+            mapping: { mapping($0, $1).map { ($0, Effect<Input, State, Never>?.none) } }
         )
     }
 
@@ -50,7 +50,7 @@ public final class Automaton<State, Input>
     ///   - mapping: `EffectMapping` that designates next state and also generates additional effect.
     public convenience init<Queue>(
         state initialState: State,
-        effect initialEffect: Effect<Input, Queue>? = nil,
+        effect initialEffect: Effect<Input, State, Queue>? = nil,
         inputs inputSignal: Signal<Input, Never>,
         mapping: @escaping EffectMapping<Queue>
         ) where Queue: EffectQueueProtocol
@@ -75,7 +75,7 @@ public final class Automaton<State, Input>
                     }
 
                 var effects = mapped
-                    .filterMap { _, _, mapped -> Effect<Input, Queue>? in
+                    .filterMap { _, _, mapped -> Effect<Input, State, Queue>? in
                         guard case let .some(_, effect) = mapped else { return nil }
                         return effect
                     }
@@ -89,7 +89,11 @@ public final class Automaton<State, Input>
                     EffectQueue<Queue>.allCases.map { queue in
                         effects
                             .filter { $0.queue == queue }
-                            .flatMap(queue.flattenStrategy) { $0.producer }
+                            .flatMap(queue.flattenStrategy) { effect -> SignalProducer<Input, Never> in
+                                /// - Note: Cancellation will be triggered regardless of state-transition success or failure.
+                                let until = from.filterMap { effect.until($0, $1) ? () : nil }
+                                return effect.producer.take(until: until)
+                            }
                     }
                 )
 

Sources/Effect.swift

@@ -2,21 +2,45 @@ import ReactiveSwift
 
 /// Managed side-effect that enqueues `producer` on `EffectQueue`
 /// to perform arbitrary `Queue.flattenStrategy`.
-public struct Effect<Input, Queue> where Queue: EffectQueueProtocol
+public struct Effect<Input, State, Queue> where Queue: EffectQueueProtocol
 {
     /// "Cold" stream that runs side-effect and sends next `Input`.
     public let producer: SignalProducer<Input, Never>
 
     /// Effect queue that associates with `producer` to perform various `flattenStrategy`s.
     internal let queue: EffectQueue<Queue>
 
-    /// - Parameter queue: Uses custom queue, or set `nil` as default queue to use `merge` strategy.
+    /// `(input, fromState)` predicate for running `producer` cancellation.
+    /// - Note: Cancellation will be triggered regardless of state-transition success or failure.
+    internal let until: (Input, State) -> Bool
+
+    /// - Parameters:
+    ///   - queue: Uses custom queue, or set `nil` as default queue to use `merge` strategy.
+    ///   - until: `(input, fromState)` predicate for running `producer` cancellation.
     public init(
         _ producer: SignalProducer<Input, Never>,
-        queue: Queue? = nil
+        queue: Queue? = nil,
+        until: @escaping (Input, State) -> Bool = { _, _ in false }
         )
     {
         self.producer = producer
         self.queue = queue.map(EffectQueue.custom) ?? .default
+        self.until = until
+    }
+}
+
+extension Effect where Input: Equatable
+{
+    public init(
+        _ producer: SignalProducer<Input, Never>,
+        queue: Queue? = nil,
+        until input: Input
+        )
+    {
+        self.init(
+            producer,
+            queue: queue,
+            until: { i, _ in i == input }
+        )
     }
 }

Sources/Mapping+Helper.swift

@@ -89,7 +89,7 @@ public func | <State, Input>(
 
 public func | <State, Input, Queue>(
     mapping: @escaping Automaton<State, Input>.Mapping,
-    effect: Effect<Input, Queue>?
+    effect: Effect<Input, State, Queue>?
     ) -> Automaton<State, Input>.EffectMapping<Queue>
 {
     return { fromState, input in

Tests/ReactiveAutomatonTests/EffectCancellationSpec.swift

@@ -7,20 +7,18 @@ class EffectCancellationSpec: QuickSpec
 {
     override func spec()
     {
-        typealias Automaton = ReactiveAutomaton.Automaton<State, Input>
-        typealias EffectMapping = Automaton.EffectMapping<Never>
-
-        let (signal, observer) = Signal<Input, Never>.pipe()
-        var automaton: Automaton?
-        var lastReply: Reply<State, Input>?
-        var testScheduler: TestScheduler!
-        var isEffectDetected: Bool = false
-
-        // WARNING:
-        // Handling `Lifetime` is actually a side-effect that should not run inside `EffectMapping`.
-        // So, this test is actually not a good example of handling cancellation.
         describe("Cancellation using Lifetime") {
 
+            typealias State = _State<Lifetime.Token>
+            typealias Automaton = ReactiveAutomaton.Automaton<State, Input>
+            typealias EffectMapping = Automaton.EffectMapping<Never>
+
+            let (signal, observer) = Signal<Input, Never>.pipe()
+            var automaton: Automaton?
+            var lastReply: Reply<State, Input>?
+            var testScheduler: TestScheduler!
+            var isEffectDetected: Bool = false
+
             beforeEach {
                 testScheduler = TestScheduler()
                 isEffectDetected = false
@@ -33,9 +31,6 @@ class EffectCancellationSpec: QuickSpec
                 let mapping: EffectMapping = { fromState, input in
                     switch (fromState.status, input) {
                     case (.idle, .userAction(.request)):
-                        // WARNING:
-                        // Handling `Lifetime` is actually a side-effect that should not run inside `EffectMapping`.
-                        // So, this test is actually not a good example of handling cancellation.
                         let (lifetime, token) = Lifetime.make()
                         let toState = fromState.with {
                             $0.status = .requesting(token)
@@ -121,6 +116,113 @@ class EffectCancellationSpec: QuickSpec
 
         }
 
+        describe("Cancellation using Effect.until") {
+
+            typealias State = _State<_Void>
+            typealias Automaton = ReactiveAutomaton.Automaton<State, Input>
+            typealias EffectMapping = Automaton.EffectMapping<Never>
+
+            let (signal, observer) = Signal<Input, Never>.pipe()
+            var automaton: Automaton?
+            var lastReply: Reply<State, Input>?
+            var testScheduler: TestScheduler!
+            var isEffectDetected: Bool = false
+
+            beforeEach {
+                testScheduler = TestScheduler()
+                isEffectDetected = false
+
+                /// Sends `.loginOK` after delay, simulating async work during `.loggingIn`.
+                let requestOKProducer =
+                    SignalProducer<Input, Never>(value: .requestOK)
+                        .delay(1, on: testScheduler)
+
+                let mapping: EffectMapping = { fromState, input in
+                    switch (fromState.status, input) {
+                    case (.idle, .userAction(.request)):
+                        let toState = fromState.with {
+                            $0.status = .requesting(_Void())
+                        }
+                        let effect = requestOKProducer
+                            .on(value: { _ in
+                                isEffectDetected = true
+                            })
+                        return (toState, Effect(effect, until: .userAction(.cancel)))
+
+                    case (.requesting, .userAction(.cancel)):
+                        let toState = fromState.with {
+                            $0.status = .idle
+                        }
+                        return (toState, nil)
+
+                    case (.requesting, .requestOK):
+                        let toState = fromState.with {
+                            $0.status = .idle
+                        }
+                        return (toState, nil)
+
+                    default:
+                        return nil
+                    }
+                }
+
+                automaton = Automaton(state: State(), inputs: signal, mapping: mapping)
+
+                _ = automaton?.replies.observeValues { reply in
+                    lastReply = reply
+                }
+
+                lastReply = nil
+            }
+
+            it("request success") {
+                expect(automaton?.state.value.status) == .idle
+                expect(lastReply).to(beNil())
+
+                observer.send(value: .userAction(.request))
+
+                expect(lastReply?.input) == .userAction(.request)
+                expect(lastReply?.fromState.status) == .idle
+                expect(lastReply?.toState?.status.requesting).toNot(beNil())
+                expect(automaton?.state.value.status.requesting).toNot(beNil())
+
+                // `loginOKProducer` will automatically send `.loginOK`
+                testScheduler.advance(by: .seconds(2))
+
+                expect(lastReply?.input) == .requestOK
+                expect(lastReply?.fromState.status.requesting).toNot(beNil())
+                expect(lastReply?.toState?.status) == .idle
+                expect(automaton?.state.value.status) == .idle
+                expect(isEffectDetected) == true
+            }
+
+            it("request cancelled") {
+                expect(automaton?.state.value.status) == .idle
+                expect(lastReply).to(beNil())
+
+                observer.send(value: .userAction(.request))
+
+                expect(lastReply?.input) == .userAction(.request)
+                expect(lastReply?.fromState.status) == .idle
+                expect(lastReply?.toState?.status.requesting).toNot(beNil())
+                expect(automaton?.state.value.status.requesting).toNot(beNil())
+
+                // `loginOKProducer` will automatically send `.loginOK`
+                observer.send(value: .userAction(.cancel))
+
+                expect(lastReply?.input) == .userAction(.cancel)
+                expect(lastReply?.fromState.status.requesting).toNot(beNil())
+                expect(lastReply?.toState?.status) == .idle
+                expect(automaton?.state.value.status) == .idle
+
+                lastReply = nil // clear `lastReply` to not retain `Lifetime.Token`
+                testScheduler.advance(by: .seconds(2))
+
+                expect(isEffectDetected) == false
+            }
+
+        }
+
     }
 }
 
@@ -138,18 +240,21 @@ private enum Input: Equatable
     }
 }
 
-private struct State: With, Equatable
+private struct _State<Requesting>: With, Equatable
+    where Requesting: Equatable
 {
     var status: Status = .idle
 
     enum Status: Equatable {
         case idle
-        case requesting(Lifetime.Token)
+        case requesting(Requesting)
 
-        var requesting: Lifetime.Token?
+        var requesting: Requesting?
         {
             guard case let .requesting(value) = self else { return nil }
             return value
         }
     }
 }
+
+private struct _Void: Equatable {}