Remove context.subscribe

This has been replaced with SignalWorker.

swift/Samples/SampleApp/Sources/DemoWorkflow.swift

@@ -185,9 +185,9 @@ extension DemoWorkflow {
             subscribeTitle = "Subscribe"
         case .subscribing:
             // Subscribe to the timer signal, simulating the title being tapped.
-            context.subscribe(signal: state.signal.signal.map({ _ -> Action in
+            context.awaitResult(for: state.signal.signal.asWorker(key: "Timer")) { _ -> Action in
                 return .titleButtonTapped
-            }))
+            }
             subscribeTitle = "Stop"
         }
 

swift/Workflow/Sources/RenderContext.swift

@@ -71,11 +71,6 @@ public class RenderContext<WorkflowType: Workflow>: RenderContextType {
         fatalError()
     }
 
-    @available(*, deprecated, message: "Use a SignalWorker instead")
-    public func subscribe<Action>(signal: Signal<Action, Never>) where Action : WorkflowAction, WorkflowType == Action.WorkflowType {
-        fatalError()
-    }
-
     public func awaitResult<W, Action>(for worker: W, outputMap: @escaping (W.Output) -> Action) where W : Worker, Action : WorkflowAction, WorkflowType == Action.WorkflowType {
         fatalError()
     }
@@ -109,12 +104,6 @@ public class RenderContext<WorkflowType: Workflow>: RenderContextType {
             return implementation.makeSink(of: actionType)
         }
 
-        override func subscribe<Action>(signal: Signal<Action, Never>) where WorkflowType == Action.WorkflowType, Action : WorkflowAction {
-            assertStillValid()
-            return implementation.subscribe(signal: signal)
-        }
-
-
         override func awaitResult<W, Action>(for worker: W, outputMap: @escaping (W.Output) -> Action) where W : Worker, Action : WorkflowAction, WorkflowType == Action.WorkflowType {
             assertStillValid()
             implementation.awaitResult(for: worker, outputMap: outputMap)
@@ -136,8 +125,6 @@ internal protocol RenderContextType: class {
 
     func makeSink<Action>(of actionType: Action.Type) -> Sink<Action> where Action: WorkflowAction, Action.WorkflowType == WorkflowType
 
-    func subscribe<Action>(signal: Signal<Action, Never>) where Action: WorkflowAction, Action.WorkflowType == WorkflowType
-
     func awaitResult<W, Action>(for worker: W, outputMap: @escaping (W.Output) -> Action) where W: Worker, Action: WorkflowAction, Action.WorkflowType == WorkflowType
 
 }

swift/Workflow/Sources/SubtreeManager.swift

@@ -36,9 +36,6 @@ extension WorkflowNode {
         /// The current array of workers
         private (set) internal var childWorkers: [AnyChildWorker] = []
 
-        /// Subscriptions from the outside world.
-        private var subscriptions: Subscriptions = Subscriptions(eventSources: [], eventPipe: EventPipe())
-
         init() {}
 
         /// Performs an update pass using the given closure.
@@ -67,16 +64,13 @@ extension WorkflowNode {
             /// as a result of this call to `render`.
             self.childWorkflows = context.usedChildWorkflows
             self.childWorkers = context.usedChildWorkers
-            /// Merge all of the signals together from the subscriptions.
-            self.subscriptions = Subscriptions(eventSources: context.eventSources, eventPipe: EventPipe())
 
             /// Captured the reusable sinks from this render pass.
             self.previousSinks = context.sinkStore.usedSinks
 
             /// Capture all the pipes to be enabled after render completes.
             self.eventPipes = context.eventPipes
             self.eventPipes.append(contentsOf: context.sinkStore.eventPipes)
-            self.eventPipes.append(self.subscriptions.eventPipe)
 
             /// Set all event pipes to `pending`.
             self.eventPipes.forEach { $0.setPending() }
@@ -158,8 +152,6 @@ extension WorkflowNode.SubtreeManager {
         private let originalChildWorkers: [AnyChildWorker]
         private (set) internal var usedChildWorkers: [AnyChildWorker]
 
-        private (set) internal var eventSources: [Signal<AnyWorkflowAction<WorkflowType>, Never>] = []
-
         internal init(previousSinks: [ObjectIdentifier:AnyReusableSink], originalChildWorkflows: [ChildKey:AnyChildWorkflow], originalChildWorkers: [AnyChildWorker]) {
             self.eventPipes = []
 
@@ -228,10 +220,6 @@ extension WorkflowNode.SubtreeManager {
             return sink
         }
 
-        func subscribe<Action>(signal: Signal<Action, Never>) where Action : WorkflowAction, WorkflowType == Action.WorkflowType {
-            eventSources.append(signal.map { AnyWorkflowAction($0) })
-        }
-
         func awaitResult<W, Action>(for worker: W, outputMap: @escaping (W.Output) -> Action) where W : Worker, Action : WorkflowAction, WorkflowType == Action.WorkflowType {
 
             let outputMap = { AnyWorkflowAction(outputMap($0)) }
@@ -470,31 +458,6 @@ extension WorkflowNode.SubtreeManager {
 }
 
 
-// MARK: - Subscriptions
-
-extension WorkflowNode.SubtreeManager {
-    fileprivate final class Subscriptions {
-        private var (lifetime, token) = Lifetime.make()
-        private (set) internal var eventPipe: EventPipe
-
-        init(eventSources: [Signal<AnyWorkflowAction<WorkflowType>, Never>], eventPipe: EventPipe) {
-            self.eventPipe = eventPipe
-
-            Signal
-                .merge(eventSources)
-                .map({ action -> Output in
-                    return Output.update(action, source: .external)
-                })
-                .observe(on: QueueScheduler.workflowExecution)
-                .take(during: lifetime)
-                .observeValues({ output in
-                    eventPipe.handle(event: output)
-                })
-        }
-    }
-}
-
-
 // MARK: - Child Workflows
 
 extension WorkflowNode.SubtreeManager {

swift/Workflow/Tests/ConcurrencyTests.swift

@@ -300,6 +300,43 @@ final class ConcurrencyTests: XCTestCase {
         disposable?.dispose()
     }
 
+    func test_allSubscriptionActionsAreApplied() {
+        let signal1 = TestSignal()
+        let signal2 = TestSignal()
+        let host = WorkflowHost(
+            workflow: TestWorkflow(
+                running: .doubleSubscribing(secondSignal: signal2),
+                signal: signal1
+            )
+        )
+
+        let renderingExpectation = XCTestExpectation()
+        let outputExpectation = XCTestExpectation()
+        let outDisposable = host.output.signal.observeValues { output in
+            outputExpectation.fulfill()
+        }
+
+        let disposable = host.rendering.signal.observeValues { rendering in
+            renderingExpectation.fulfill()
+        }
+
+        let screen = host.rendering.value
+
+        XCTAssertEqual(0, screen.count)
+
+        signal1.send(value: 1)
+        signal2.send(value: 2)
+
+        XCTAssertEqual(0, host.rendering.value.count)
+
+        wait(for: [renderingExpectation, outputExpectation], timeout: 1.0)
+
+        XCTAssertEqual(101, host.rendering.value.count)
+
+        disposable?.dispose()
+        outDisposable?.dispose()
+    }
+
     // Workers are subscribed on a different scheduler than the UI scheduler,
     // which means that if they fire immediately, the action will be received after
     // `render` has completed.
@@ -549,6 +586,7 @@ final class ConcurrencyTests: XCTestCase {
         enum Running {
             case idle
             case signal
+            case doubleSubscribing(secondSignal: TestSignal)
             case worker
         }
         var signal: TestSignal
@@ -574,12 +612,16 @@ final class ConcurrencyTests: XCTestCase {
             typealias WorkflowType = TestWorkflow
 
             case update
+            case secondUpdate
 
             func apply(toState state: inout State) -> Output? {
                 switch self {
                 case .update:
                     state.count += 1
                     return .emit
+                case .secondUpdate:
+                    state.count += 100
+                    return nil
                 }
             }
         }
@@ -592,9 +634,17 @@ final class ConcurrencyTests: XCTestCase {
             case .idle:
                 break
             case .signal:
-                context.subscribe(signal: signal.signal.map({ _ -> Action in
+                context.awaitResult(for: signal.signal.asWorker(key: "signal1")) { _ -> Action in
+                    return .update
+                }
+
+            case .doubleSubscribing(secondSignal: let signal2):
+                context.awaitResult(for: signal2.signal.asWorker(key: "signal2")) { _ -> Action in
+                    return .secondUpdate
+                }
+                context.awaitResult(for: signal.signal.asWorker(key: "signal1")) { _ -> Action in
                     return .update
-                }))
+                }
 
             case .worker:
                 context.awaitResult(for: TestWorker())

swift/Workflow/Tests/SubtreeManagerTests.swift

@@ -241,9 +241,9 @@ final class SubtreeManagerTests: XCTestCase {
 
             func render(state: SubscribingWorkflow.State, context: RenderContext<SubscribingWorkflow>) -> Bool {
                 if let signal = signal {
-                    context.subscribe(signal: signal.map({ _ -> AnyWorkflowAction<SubscribingWorkflow> in
+                    context.awaitResult(for: signal.asWorker(key: "signal")) { _ -> AnyWorkflowAction<SubscribingWorkflow> in
                         return AnyWorkflowAction.noAction
-                    }))
+                    }
                     return true
                 } else {
                     return false

swift/WorkflowTesting/Sources/WorkflowRenderTester.swift

@@ -270,16 +270,14 @@ fileprivate final class RenderTestContext<T: Workflow>: RenderContextType {
         let sink = Sink<Action> { action in
             observer.send(value: AnyWorkflowAction(action))
         }
-        subscribe(signal: signal)
-        return sink
-    }
 
-    func subscribe<Action>(signal: Signal<Action, Never>) where Action : WorkflowAction, RenderTestContext<T>.WorkflowType == Action.WorkflowType {
         signal
             .take(during: lifetime)
             .observeValues { [weak self] action in
                 self?.apply(action: action)
             }
+
+        return sink
     }
 
     func awaitResult<W, Action>(for worker: W, outputMap: @escaping (W.Output) -> Action) where W : Worker, Action : WorkflowAction, RenderTestContext<T>.WorkflowType == Action.WorkflowType {

swift/WorkflowUI/Tests/ContainerViewControllerTests.swift

@@ -135,13 +135,12 @@ fileprivate struct MockWorkflow: Workflow {
     }
 
     func render(state: State, context: RenderContext<MockWorkflow>) -> TestScreen {
-
-        context.subscribe(signal: subscription.map { output in
+        context.awaitResult(for: subscription.asWorker(key: "signal")) { output in
             return AnyWorkflowAction { state in
                 state = output
                 return output
             }
-        })
+        }
 
         return TestScreen(string: "\(state)")
     }