feat(subscribeOn): add higher-order lettable version of subscribeOn

Author: benlesh

src/operator/subscribeOn.ts

@@ -1,9 +1,7 @@
-import { Operator } from '../Operator';
+
 import { IScheduler } from '../Scheduler';
-import { Subscriber } from '../Subscriber';
 import { Observable } from '../Observable';
-import { TeardownLogic } from '../Subscription';
-import { SubscribeOnObservable } from '../observable/SubscribeOnObservable';
+import { subscribeOn as higherOrder } from '../operators';
 
 /**
  * Asynchronously subscribes Observers to this Observable on the specified IScheduler.
@@ -17,16 +15,5 @@ import { SubscribeOnObservable } from '../observable/SubscribeOnObservable';
  * @owner Observable
  */
 export function subscribeOn<T>(this: Observable<T>, scheduler: IScheduler, delay: number = 0): Observable<T> {
-  return this.lift(new SubscribeOnOperator<T>(scheduler, delay));
-}
-
-class SubscribeOnOperator<T> implements Operator<T, T> {
-  constructor(private scheduler: IScheduler,
-              private delay: number) {
-  }
-  call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return new SubscribeOnObservable(
-      source, this.delay, this.scheduler
-    ).subscribe(subscriber);
-  }
+  return higherOrder(scheduler, delay)(this);
 }

src/operators/index.ts

@@ -26,6 +26,7 @@ export { race } from './race';
 export { reduce } from './reduce';
 export { refCount } from './refCount';
 export { scan } from './scan';
+export { subscribeOn } from './subscribeOn';
 export { switchAll } from './switchAll';
 export { switchMap } from './switchMap';
 export { takeLast } from './takeLast';

src/operators/subscribeOn.ts

@@ -0,0 +1,35 @@
+import { Operator } from '../Operator';
+import { IScheduler } from '../Scheduler';
+import { Subscriber } from '../Subscriber';
+import { Observable } from '../Observable';
+import { TeardownLogic } from '../Subscription';
+import { SubscribeOnObservable } from '../observable/SubscribeOnObservable';
+import { MonoTypeOperatorFunction } from '../interfaces';
+
+/**
+ * Asynchronously subscribes Observers to this Observable on the specified IScheduler.
+ *
+ * <img src="./img/subscribeOn.png" width="100%">
+ *
+ * @param {Scheduler} scheduler - The IScheduler to perform subscription actions on.
+ * @return {Observable<T>} The source Observable modified so that its subscriptions happen on the specified IScheduler.
+ .
+ * @method subscribeOn
+ * @owner Observable
+ */
+export function subscribeOn<T>(scheduler: IScheduler, delay: number = 0): MonoTypeOperatorFunction<T> {
+  return function subscribeOnOperatorFunction(source: Observable<T>): Observable<T> {
+    return source.lift(new SubscribeOnOperator<T>(scheduler, delay));
+  };
+}
+
+class SubscribeOnOperator<T> implements Operator<T, T> {
+  constructor(private scheduler: IScheduler,
+              private delay: number) {
+  }
+  call(subscriber: Subscriber<T>, source: any): TeardownLogic {
+    return new SubscribeOnObservable(
+      source, this.delay, this.scheduler
+    ).subscribe(subscriber);
+  }
+}