Fix subscribe_on: forward the scheduler to the source.

reactivex/operators/_subscribeon.py

@@ -43,7 +43,7 @@ def subscribe(
 
             def action(scheduler: abc.SchedulerBase, state: Optional[Any] = None):
                 d.disposable = ScheduledDisposable(
-                    scheduler, source.subscribe(observer)
+                    scheduler, source.subscribe(observer, scheduler=scheduler)
                 )
 
             m.disposable = scheduler.schedule(action)