[fix]: update RxSwift bindings to observe on main thread (#294)

Resolves: (#293)[#293]
square · Aug 7, 2024 · 249a89c · 249a89c
1 parent 8b7c069
commit 249a89c
Show file tree

Hide file tree

Showing 2 changed files with 50 additions and 1 deletion.
diff --git a/WorkflowRxSwift/Sources/ObservableWorkflow.swift b/WorkflowRxSwift/Sources/ObservableWorkflow.swift
@@ -20,7 +20,7 @@ import class Workflow.Lifetime
 
 extension Observable: AnyWorkflowConvertible {
     public func asAnyWorkflow() -> AnyWorkflow<Void, Element> {
-        return ObservableWorkflow(observable: self).asAnyWorkflow()
+        ObservableWorkflow(observable: self).asAnyWorkflow()
     }
 }
 
@@ -41,6 +41,7 @@ struct ObservableWorkflow<Value>: Workflow {
             let disposable = observable
                 .map { AnyWorkflowAction(sendingOutput: $0) }
                 .subscribe(on: MainScheduler.asyncInstance)
+                .observe(on: MainScheduler.asyncInstance)
                 .subscribe(onNext: { value in
                     sink.send(value)
                 })

diff --git a/WorkflowRxSwift/Tests/Rx+ReactiveWorkers.swift b/WorkflowRxSwift/Tests/Rx+ReactiveWorkers.swift
@@ -38,6 +38,54 @@ class Rx_ReactiveWorkersTests: XCTestCase {
 
         disposable?.dispose()
     }
+
+    func test_observes_on_main_queue() {
+        struct TestWorkflow: Workflow {
+            enum Action: WorkflowAction {
+                typealias WorkflowType = TestWorkflow
+                case complete
+
+                func apply(toState state: inout State) -> Output? {
+                    switch self {
+                    case .complete:
+                        return .finished
+                    }
+                }
+            }
+
+            enum Output {
+                case finished
+            }
+
+            func render(state: Void, context: RenderContext<Self>) {
+                Single<Void>.create { observer in
+                    DispatchQueue.global().async {
+                        observer(.success(()))
+                    }
+                    return Disposables.create()
+                }
+                .asObservable()
+                .running(in: context) { _ in
+                    XCTAssert(Thread.isMainThread)
+                    return Action.complete
+                }
+            }
+        }
+
+        let host = WorkflowHost(
+            workflow: TestWorkflow()
+        )
+
+        let expectation = XCTestExpectation()
+        let disposable = host.output.signal.observeValues { output in
+            if output == .finished {
+                expectation.fulfill()
+            }
+        }
+
+        wait(for: [expectation], timeout: 1.0)
+        disposable?.dispose()
+    }
 }
 
 struct CombinedWorkflow: Workflow {