Repository: ReactiveX/RxPY

Issue:
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.

Discussion:
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**.


Patch:
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


Test patch:
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