frederikaalund: Fixes #489
MainRo: > I guess this is related to the fact that scheduler is a keyword argument only, so scheduler must be given as `subscribe(observer, scheduler=scheduler)`, otherwise it will be dismissed.

Yes, the second argument to the subscribe method is on_error. So scheduler has to be set with kw unless the 3 callbacks are provided before.

Nice!
coveralls: 
[![Coverage Status](https://coveralls.io/builds/28920383/badge)](https://coveralls.io/builds/28920383)

Coverage remained the same at 92.924% when pulling **ee14b7c90d3ae0fd23393c30a380087b3d486a64 on jcafhe:fix-scheduler-ConnectableObservable** into **5fc2cfb5911e29818adde7dd70ef74ddb085eba4 on ReactiveX:master**.

diff --git a/rx/core/observable/connectableobservable.py b/rx/core/observable/connectableobservable.py
index f2d3637e..ab0acd51 100644
--- a/rx/core/observable/connectableobservable.py
+++ b/rx/core/observable/connectableobservable.py
@@ -28,7 +28,7 @@ class ConnectableObservable(Observable):
             def dispose():
                 self.has_subscription = False
 
-            subscription = self.source.subscribe(self.subject, scheduler)
+            subscription = self.source.subscribe(self.subject, scheduler=scheduler)
             self.subscription = CompositeDisposable(subscription, Disposable(dispose))
 
         return self.subscription

Scheduler provided in subscribe is not used for all observable factories
We now have `Observable.subcribe(..., scheduler=scheduler)`, which is really nice. Let's take the API all the way.

[The documentation says that](https://rxpy.readthedocs.io/en/latest/get_started.html#default-scheduler):

> Operators that accept a scheduler select the scheduler to use in the following way:
>  * If a scheduler is provided for the operator, then use it.
>  * If a default scheduler is provided in subscribe, then use it.
>  * Otherwise use the default scheduler of the operator.

I know that this applies to operators but I think that it would make sense if it also applied to observable factories. This way, we can finally get rid of all the `scheduler=scheduler` boilerplate. See #183 for additional info.

The `timer` observable factory, for instance, still defaults to the `TimeoutScheduler` .

I guess this is a feature request.
diff --git a/tests/test_observable/test_connectableobservable.py b/tests/test_observable/test_connectableobservable.py
index 00709145..8ab7a4e2 100644
--- a/tests/test_observable/test_connectableobservable.py
+++ b/tests/test_observable/test_connectableobservable.py
@@ -241,3 +241,19 @@ class TestConnectableObservable(unittest.TestCase):
             subscribe(225, 241),
             subscribe(249, 255),
             subscribe(275, 295)]
+
+    def test_connectable_observable_forward_scheduler(self):
+        scheduler = TestScheduler()
+        subscribe_scheduler = 'unknown'
+
+        def subscribe(observer, scheduler=None):
+            nonlocal subscribe_scheduler
+            subscribe_scheduler = scheduler
+
+        xs = rx.create(subscribe)
+        subject = MySubject()
+
+        conn = ConnectableObservable(xs, subject)
+        conn.connect(scheduler)
+
+        assert subscribe_scheduler is scheduler

pip install -e .
parse_log_pytest
requirements.txt
pytest>=4.6.1 pytest-asyncio>=0.10.0 coverage>=4.5.3
apt-get update apt-get install -y gcc
3.9
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
coverage==7.8.0
exceptiongroup==1.2.2
iniconfig==2.1.0
packaging==24.2
pluggy==1.5.0
pytest==8.3.5
pytest-asyncio==0.26.0
-e git+https://github.com/ReactiveX/RxPY.git@5fc2cfb5911e29818adde7dd70ef74ddb085eba4#egg=Rx
tomli==2.2.1
typing_extensions==4.13.0

name: RxPY
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - coverage==7.8.0
      - exceptiongroup==1.2.2
      - iniconfig==2.1.0
      - packaging==24.2
      - pluggy==1.5.0
      - pytest==8.3.5
      - pytest-asyncio==0.26.0
      - tomli==2.2.1
      - typing-extensions==4.13.0
prefix: /opt/conda/envs/RxPY

tests/test_observable/test_connectableobservable.py::TestConnectableObservable::test_connectable_observable_forward_scheduler
tests/test_observable/test_connectableobservable.py::TestConnectableObservable::test_connectable_observable_connected tests/test_observable/test_connectableobservable.py::TestConnectableObservable::test_connectable_observable_creation tests/test_observable/test_connectableobservable.py::TestConnectableObservable::test_connectable_observable_disconnect_future tests/test_observable/test_connectableobservable.py::TestConnectableObservable::test_connectable_observable_disconnected tests/test_observable/test_connectableobservable.py::TestConnectableObservable::test_connectable_observable_multiple_non_overlapped_connections tests/test_observable/test_connectableobservable.py::TestConnectableObservable::test_connectable_observable_not_connected
rx/core/observable/connectableobservable.py
rx/core/observable/connectableobservable.py