Repository: ReactiveX/RxPY

Issue:
observe_on operator does not propagate subscription scheduler
The observe_on scheduler drops the scheduler parameter provided during subscription. As a consequence, the default scheduler provided in the pipe operator is lost when observe_on is used in the chain.

Discussion:

Patch:
diff --git a/rx/core/operators/observeon.py b/rx/core/operators/observeon.py
index d725401b..d1ea4a56 100644
--- a/rx/core/operators/observeon.py
+++ b/rx/core/operators/observeon.py
@@ -22,8 +22,9 @@ def _observe_on(scheduler) -> Callable[[Observable], Observable]:
             Returns the source sequence whose observations happen on
             the specified scheduler.
         """
-        def subscribe(observer, _=None):
-            return source.subscribe(ObserveOnObserver(scheduler, observer))
+        def subscribe(observer, subscribe_scheduler=None):
+            return source.subscribe(ObserveOnObserver(scheduler, observer),
+                                    scheduler=subscribe_scheduler)
 
         return Observable(subscribe)
-    return observe_on
\ No newline at end of file
+    return observe_on


Test patch:
diff --git a/tests/test_observable/test_observeon.py b/tests/test_observable/test_observeon.py
index 450ed134..09d265b4 100644
--- a/tests/test_observable/test_observeon.py
+++ b/tests/test_observable/test_observeon.py
@@ -2,6 +2,7 @@ import unittest
 
 import rx
 from rx import operators as ops
+from rx.concurrency import ImmediateScheduler
 from rx.testing import TestScheduler, ReactiveTest
 
 on_next = ReactiveTest.on_next
@@ -75,3 +76,21 @@ class TestObserveOn(unittest.TestCase):
 
         assert results.messages == []
         assert xs.subscriptions == [subscribe(200, 1000)]
+
+    def test_observe_on_forward_subscribe_scheduler(self):
+        scheduler = ImmediateScheduler()
+        expected_subscribe_scheduler = ImmediateScheduler()
+
+        actual_subscribe_scheduler = None
+
+        def subscribe(observer, scheduler):
+            nonlocal actual_subscribe_scheduler
+            actual_subscribe_scheduler = scheduler
+            observer.on_completed()
+
+        xs = rx.create(subscribe)
+
+        xs.pipe(ops.observe_on(scheduler)).subscribe(
+            scheduler=expected_subscribe_scheduler)
+
+        assert expected_subscribe_scheduler == actual_subscribe_scheduler