[Exploration] Use a run loop observer to trigger render passes asynchronously

swift/Workflow/Sources/RunLoopObserver.swift

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2020 Square Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import CoreFoundation
+
+/// Swift wrapper for `CFRunLoopObserver`
+internal final class RunLoopObserver {
+    private var observer: CFRunLoopObserver!
+
+    /// Creates a `RunLoopObserver` and adds it to the given run loop for the given run loop modes. See the docs for `CFRunLoopObserverCreateWithHandler` for documentation.
+    init(
+        runLoop: CFRunLoop = CFRunLoopGetCurrent(),
+        activityStages: CFRunLoopActivity,
+        repeats: Bool = true,
+        order: CFIndex = 0,
+        runLoopModes: CFRunLoopMode = .defaultMode,
+        callback: @escaping (_ activityStage: CFRunLoopActivity) -> Void
+    ) {
+        observer = CFRunLoopObserverCreateWithHandler(
+            kCFAllocatorDefault,
+            activityStages.rawValue,
+            repeats,
+            order,
+            { (observer, activityStage) in
+                callback(activityStage)
+            }
+        )
+        CFRunLoopAddObserver(runLoop, observer, runLoopModes)
+    }
+
+    deinit {
+        // Clean up the observer
+        CFRunLoopObserverInvalidate(observer)
+    }
+}

swift/Workflow/Sources/WorkflowHost.swift

@@ -45,6 +45,10 @@ public final class WorkflowHost<WorkflowType: Workflow> {
     /// as state transitions occur within the hierarchy.
     public let rendering: Property<WorkflowType.Rendering>
 
+    // Run loop management
+    private var needsRender = false
+    private var runLoopObserver: RunLoopObserver?
+
     /// Initializes a new host with the given workflow at the root.
     ///
     /// - Parameter workflow: The root workflow in the hierarchy
@@ -65,6 +69,23 @@ public final class WorkflowHost<WorkflowType: Workflow> {
             self?.handle(output: output)
         }
 
+        runLoopObserver = RunLoopObserver(activityStages: .beforeWaiting, runLoopModes: .commonModes) { [weak self] activity in
+            self?.renderIfNeeded()
+        }
+    }
+
+    public func setNeedsRender() {
+        needsRender = true
+    }
+
+    func renderIfNeeded() {
+        guard needsRender else { return }
+
+        defer { needsRender = false }
+
+        // Do the rendering
+        mutableRendering.value = rootNode.render()
+        rootNode.enableEvents()
     }
 
     /// Update the input for the workflow. Will cause a render pass.
@@ -81,8 +102,6 @@ public final class WorkflowHost<WorkflowType: Workflow> {
     }
 
     private func handle(output: WorkflowNode<WorkflowType>.Output) {
-        mutableRendering.value = rootNode.render()
-
         if let outputEvent = output.outputEvent {
             outputEventObserver.send(value: outputEvent)
         }
@@ -91,7 +110,7 @@ public final class WorkflowHost<WorkflowType: Workflow> {
             snapshot: rootNode.makeDebugSnapshot(),
             updateInfo: output.debugInfo)
 
-        rootNode.enableEvents()
+        setNeedsRender()
     }
 
     /// A signal containing output events emitted by the root workflow in the hierarchy.

swift/Workflow/Tests/ConcurrencyTests.swift

@@ -22,6 +22,7 @@ import ReactiveSwift
 final class ConcurrencyTests: XCTestCase {
 
     // Applying an action from a sink must synchronously update the rendering.
+    // FIXME: This isn't true anymore
     func test_sinkRenderLoopIsSynchronous() {
         let host = WorkflowHost(workflow: TestWorkflow())
 
@@ -41,7 +42,8 @@ final class ConcurrencyTests: XCTestCase {
         XCTAssertEqual(0, initialScreen.count)
         initialScreen.update()
 
-        // This update happens immediately as a new rendering is generated synchronously.
+        // Force a render pass as a new rendering is NOT generated synchronously.
+        host.renderIfNeeded()
         XCTAssertEqual(1, host.rendering.value.count)
 
         wait(for: [expectation], timeout: 1.0)
@@ -56,6 +58,7 @@ final class ConcurrencyTests: XCTestCase {
     }
 
     // Events emitted between `render` on a workflow and `enableEvents` are queued and will be delivered immediately when `enableEvents` is called.
+    // FIXME: This test is broken
     func test_queuedEvents() {
         let host = WorkflowHost(workflow: TestWorkflow())
 
@@ -76,6 +79,7 @@ final class ConcurrencyTests: XCTestCase {
 
         // Updating the screen will cause two events - the `update` here, and the update caused by the first time the rendering changes.
         initialScreen.update()
+        host.renderIfNeeded()
 
         XCTAssertEqual(2, host.rendering.value.count)
 
@@ -98,12 +102,14 @@ final class ConcurrencyTests: XCTestCase {
         // sink of the same type.
         initialScreen.update()
 
+        host.renderIfNeeded()
         let secondScreen = host.rendering.value
         XCTAssertEqual(1, secondScreen.count)
 
         // Send an action from the original screen and sink. It should be proxied through the most recent sink.
         initialScreen.update()
 
+        host.renderIfNeeded()
         let thirdScreen = host.rendering.value
         XCTAssertEqual(2, thirdScreen.count)
     }
@@ -121,6 +127,7 @@ final class ConcurrencyTests: XCTestCase {
         // sink of the same type.
         initialScreen.update()
 
+        host.renderIfNeeded()
         let secondScreen = host.rendering.value
         XCTAssertEqual(1, secondScreen.count)
 
@@ -131,6 +138,7 @@ final class ConcurrencyTests: XCTestCase {
         // If the sink *was* still valid, this would be correct. However, it should just fail and be `1` still.
         //XCTAssertEqual(2, secondScreen.count)
         // Actual expected result, if we had not fatal errored.
+        host.renderIfNeeded()
         XCTAssertEqual(1, host.rendering.value.count)
 
         struct OneShotWorkflow: Workflow {
@@ -205,6 +213,7 @@ final class ConcurrencyTests: XCTestCase {
         let initialScreen = host.rendering.value
         initialScreen.update()
 
+        host.renderIfNeeded()
         XCTAssertEqual(2, debugger.snapshots.count)
         XCTAssertEqual("1", debugger.snapshots[0].stateDescription)
         XCTAssertEqual("2", debugger.snapshots[1].stateDescription)
@@ -220,6 +229,7 @@ final class ConcurrencyTests: XCTestCase {
         XCTAssertEqual(0, initialScreen.count)
         initialScreen.update()
 
+        host.renderIfNeeded()
         // This update happens immediately as a new rendering is generated synchronously.
         // Both the child updates from the action (incrementing state by 1) as well as the
         // parent from the output (incrementing its state by 10)
@@ -300,6 +310,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 expectation = XCTestExpectation()
+        let exp = XCTestExpectation()
+        let outDisposable = host.output.signal.observeValues { output in
+            exp.fulfill()
+        }
+
+        let disposable = host.rendering.signal.observeValues { rendering in
+            expectation.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: [expectation, exp], 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.
@@ -334,12 +381,14 @@ final class ConcurrencyTests: XCTestCase {
 
         // Update using the first action.
         initialScreen.updateFirst()
+        host.renderIfNeeded()
 
         let secondScreen = host.rendering.value
         XCTAssertEqual(1, secondScreen.count)
 
         // Update using the second action.
         secondScreen.updateSecond()
+        host.renderIfNeeded()
         XCTAssertEqual(11, host.rendering.value.count)
 
         struct AnyActionWorkflow: Workflow {
@@ -548,6 +597,7 @@ final class ConcurrencyTests: XCTestCase {
         enum Running {
             case idle
             case signal
+            case doubleSubscribing(secondSignal: TestSignal)
             case worker
         }
         var signal: TestSignal
@@ -573,20 +623,23 @@ 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
                 }
             }
         }
 
         typealias Rendering = TestScreen
 
         func render(state: State, context: RenderContext<TestWorkflow>) -> Rendering {
-
             switch state.running {
             case .idle:
                 break
@@ -595,6 +648,14 @@ final class ConcurrencyTests: XCTestCase {
                     return .update
                 }))
 
+            case .doubleSubscribing(secondSignal: let signal2):
+                context.subscribe(signal: signal.signal.map { _ in
+                    return Action.update
+                })
+                context.subscribe(signal: signal2.signal.map { _ in
+                    return Action.secondUpdate
+                })
+
             case .worker:
                 context.awaitResult(for: TestWorker())
             }

swift/Workflow/Tests/WorkflowHostTests.swift

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 import XCTest
-import Workflow
+@testable import Workflow
 
 
 final class WorkflowHostTests: XCTestCase {
@@ -25,6 +25,7 @@ final class WorkflowHostTests: XCTestCase {
         XCTAssertEqual(1, host.rendering.value)
 
         host.update(workflow: TestWorkflow(step: .second))
+        host.renderIfNeeded()
 
         XCTAssertEqual(2, host.rendering.value)
     }