Merge branch 'master' into window-time-with-max-size

src/Observable.ts

@@ -93,7 +93,7 @@ export class Observable<T> implements Subscribable<T> {
     const sink = toSubscriber(observerOrNext, error, complete);
 
     if (operator) {
-      operator.call(sink, this);
+      operator.call(sink, this.source);
     } else {
       sink.add(this._subscribe(sink));
     }

src/observable/ConnectableObservable.ts

@@ -97,7 +97,7 @@ class RefCountOperator<T> implements Operator<T, T> {
     (<any> connectable)._refCount++;
 
     const refCounter = new RefCountSubscriber(subscriber, connectable);
-    const subscription = source._subscribe(refCounter);
+    const subscription = source.subscribe(refCounter);
 
     if (!refCounter.closed) {
       (<any> refCounter).connection = connectable.connect();

src/observable/SubscribeOnObservable.ts

@@ -1,3 +1,4 @@
+import { Action } from '../scheduler/Action';
 import { Scheduler } from '../Scheduler';
 import { Subscriber } from '../Subscriber';
 import { Subscription } from '../Subscription';
@@ -20,9 +21,9 @@ export class SubscribeOnObservable<T> extends Observable<T> {
     return new SubscribeOnObservable(source, delay, scheduler);
   }
 
-  static dispatch<T>(arg: DispatchArg<T>): Subscription {
+  static dispatch<T>(this: Action<T>, arg: DispatchArg<T>): Subscription {
     const { source, subscriber } = arg;
-    return source.subscribe(subscriber);
+    return this.add(source.subscribe(subscriber));
   }
 
   constructor(public source: Observable<T>,

src/operator/audit.ts

@@ -57,7 +57,7 @@ class AuditOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new AuditSubscriber<T, T>(subscriber, this.durationSelector));
+    return source.subscribe(new AuditSubscriber<T, T>(subscriber, this.durationSelector));
   }
 }
 

src/operator/auditTime.ts

@@ -57,7 +57,7 @@ class AuditTimeOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new AuditTimeSubscriber(subscriber, this.duration, this.scheduler));
+    return source.subscribe(new AuditTimeSubscriber(subscriber, this.duration, this.scheduler));
   }
 }
 

src/operator/buffer.ts

@@ -48,7 +48,7 @@ class BufferOperator<T> implements Operator<T, T[]> {
   }
 
   call(subscriber: Subscriber<T[]>, source: any): any {
-    return source._subscribe(new BufferSubscriber(subscriber, this.closingNotifier));
+    return source.subscribe(new BufferSubscriber(subscriber, this.closingNotifier));
   }
 }
 

src/operator/bufferCount.ts

@@ -52,7 +52,7 @@ class BufferCountOperator<T> implements Operator<T, T[]> {
   }
 
   call(subscriber: Subscriber<T[]>, source: any): any {
-    return source._subscribe(new BufferCountSubscriber(subscriber, this.bufferSize, this.startBufferEvery));
+    return source.subscribe(new BufferCountSubscriber(subscriber, this.bufferSize, this.startBufferEvery));
   }
 }
 

src/operator/bufferTime.ts

@@ -86,7 +86,7 @@ class BufferTimeOperator<T> implements Operator<T, T[]> {
   }
 
   call(subscriber: Subscriber<T[]>, source: any): any {
-    return source._subscribe(new BufferTimeSubscriber(
+    return source.subscribe(new BufferTimeSubscriber(
       subscriber, this.bufferTimeSpan, this.bufferCreationInterval, this.maxBufferSize, this.scheduler
     ));
   }

src/operator/bufferToggle.ts

@@ -57,7 +57,7 @@ class BufferToggleOperator<T, O> implements Operator<T, T[]> {
   }
 
   call(subscriber: Subscriber<T[]>, source: any): any {
-    return source._subscribe(new BufferToggleSubscriber(subscriber, this.openings, this.closingSelector));
+    return source.subscribe(new BufferToggleSubscriber(subscriber, this.openings, this.closingSelector));
   }
 }
 
@@ -167,4 +167,4 @@ class BufferToggleSubscriber<T, O> extends OuterSubscriber<T, O> {
       subscription.add(innerSubscription);
     }
   }
-}
+}

src/operator/bufferWhen.ts

@@ -52,7 +52,7 @@ class BufferWhenOperator<T> implements Operator<T, T[]> {
   }
 
   call(subscriber: Subscriber<T[]>, source: any): any {
-    return source._subscribe(new BufferWhenSubscriber(subscriber, this.closingSelector));
+    return source.subscribe(new BufferWhenSubscriber(subscriber, this.closingSelector));
   }
 }
 

src/operator/catch.ts

@@ -29,7 +29,7 @@ class CatchOperator<T, R> implements Operator<T, R> {
   }
 
   call(subscriber: Subscriber<R>, source: any): any {
-    return source._subscribe(new CatchSubscriber(subscriber, this.selector, this.caught));
+    return source.subscribe(new CatchSubscriber(subscriber, this.selector, this.caught));
   }
 }
 

src/operator/combineLatest.ts

@@ -87,7 +87,7 @@ export class CombineLatestOperator<T, R> implements Operator<T, R> {
   }
 
   call(subscriber: Subscriber<R>, source: any): any {
-    return source._subscribe(new CombineLatestSubscriber(subscriber, this.project));
+    return source.subscribe(new CombineLatestSubscriber(subscriber, this.project));
   }
 }
 

src/operator/count.ts

@@ -58,7 +58,7 @@ class CountOperator<T> implements Operator<T, number> {
   }
 
   call(subscriber: Subscriber<number>, source: any): any {
-    return source._subscribe(new CountSubscriber(subscriber, this.predicate, this.source));
+    return source.subscribe(new CountSubscriber(subscriber, this.predicate, this.source));
   }
 }
 

src/operator/debounce.ts

@@ -58,7 +58,7 @@ class DebounceOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new DebounceSubscriber(subscriber, this.durationSelector));
+    return source.subscribe(new DebounceSubscriber(subscriber, this.durationSelector));
   }
 }
 

src/operator/debounceTime.ts

@@ -60,7 +60,7 @@ class DebounceTimeOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new DebounceTimeSubscriber(subscriber, this.dueTime, this.scheduler));
+    return source.subscribe(new DebounceTimeSubscriber(subscriber, this.dueTime, this.scheduler));
   }
 }
 

src/operator/defaultIfEmpty.ts

@@ -47,7 +47,7 @@ class DefaultIfEmptyOperator<T, R> implements Operator<T, T | R> {
   }
 
   call(subscriber: Subscriber<T | R>, source: any): any {
-    return source._subscribe(new DefaultIfEmptySubscriber(subscriber, this.defaultValue));
+    return source.subscribe(new DefaultIfEmptySubscriber(subscriber, this.defaultValue));
   }
 }
 

src/operator/delay.ts

@@ -59,7 +59,7 @@ class DelayOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new DelaySubscriber(subscriber, this.delay, this.scheduler));
+    return source.subscribe(new DelaySubscriber(subscriber, this.delay, this.scheduler));
   }
 }
 

src/operator/delayWhen.ts

@@ -66,7 +66,7 @@ class DelayWhenOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new DelayWhenSubscriber(subscriber, this.delayDurationSelector));
+    return source.subscribe(new DelayWhenSubscriber(subscriber, this.delayDurationSelector));
   }
 }
 

src/operator/dematerialize.ts

@@ -44,7 +44,7 @@ export function dematerialize<T>(this: Observable<T>): Observable<any> {
 
 class DeMaterializeOperator<T extends Notification<any>, R> implements Operator<T, R> {
   call(subscriber: Subscriber<any>, source: any): any {
-    return source._subscribe(new DeMaterializeSubscriber(subscriber));
+    return source.subscribe(new DeMaterializeSubscriber(subscriber));
   }
 }
 

src/operator/distinct.ts

@@ -34,7 +34,7 @@ class DistinctOperator<T, K> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new DistinctSubscriber(subscriber, this.keySelector, this.flushes));
+    return source.subscribe(new DistinctSubscriber(subscriber, this.keySelector, this.flushes));
   }
 }
 

src/operator/distinctUntilChanged.ts

@@ -29,7 +29,7 @@ class DistinctUntilChangedOperator<T, K> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new DistinctUntilChangedSubscriber(subscriber, this.compare, this.keySelector));
+    return source.subscribe(new DistinctUntilChangedSubscriber(subscriber, this.compare, this.keySelector));
   }
 }
 

src/operator/do.ts

@@ -64,7 +64,7 @@ class DoOperator<T> implements Operator<T, T> {
               private complete?: () => void) {
   }
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new DoSubscriber(subscriber, this.nextOrObserver, this.error, this.complete));
+    return source.subscribe(new DoSubscriber(subscriber, this.nextOrObserver, this.error, this.complete));
   }
 }
 

src/operator/elementAt.ts

@@ -54,7 +54,7 @@ class ElementAtOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new ElementAtSubscriber(subscriber, this.index, this.defaultValue));
+    return source.subscribe(new ElementAtSubscriber(subscriber, this.index, this.defaultValue));
   }
 }
 

src/operator/every.ts

@@ -23,7 +23,7 @@ class EveryOperator<T> implements Operator<T, boolean> {
   }
 
   call(observer: Subscriber<boolean>, source: any): any {
-    return source._subscribe(new EverySubscriber(observer, this.predicate, this.thisArg, this.source));
+    return source.subscribe(new EverySubscriber(observer, this.predicate, this.thisArg, this.source));
   }
 }
 

src/operator/exhaust.ts

@@ -47,7 +47,7 @@ export function exhaust<T>(this: Observable<T>): Observable<T> {
 
 class SwitchFirstOperator<T> implements Operator<T, T> {
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new SwitchFirstSubscriber(subscriber));
+    return source.subscribe(new SwitchFirstSubscriber(subscriber));
   }
 }
 

src/operator/exhaustMap.ts

@@ -67,7 +67,7 @@ class SwitchFirstMapOperator<T, I, R> implements Operator<T, R> {
   }
 
   call(subscriber: Subscriber<R>, source: any): any {
-    return source._subscribe(new SwitchFirstMapSubscriber(subscriber, this.project, this.resultSelector));
+    return source.subscribe(new SwitchFirstMapSubscriber(subscriber, this.project, this.resultSelector));
   }
 }
 

src/operator/expand.ts

@@ -74,7 +74,7 @@ export class ExpandOperator<T, R> implements Operator<T, R> {
   }
 
   call(subscriber: Subscriber<R>, source: any): any {
-    return source._subscribe(new ExpandSubscriber(subscriber, this.project, this.concurrent, this.scheduler));
+    return source.subscribe(new ExpandSubscriber(subscriber, this.project, this.concurrent, this.scheduler));
   }
 }
 

src/operator/filter.ts

@@ -62,7 +62,7 @@ class FilterOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new FilterSubscriber(subscriber, this.predicate, this.thisArg));
+    return source.subscribe(new FilterSubscriber(subscriber, this.predicate, this.thisArg));
   }
 }
 

src/operator/finally.ts

@@ -20,7 +20,7 @@ class FinallyOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new FinallySubscriber(subscriber, this.callback));
+    return source.subscribe(new FinallySubscriber(subscriber, this.callback));
   }
 }
 

src/operator/find.ts

@@ -60,7 +60,7 @@ export class FindValueOperator<T> implements Operator<T, T> {
   }
 
   call(observer: Subscriber<T>, source: any): any {
-    return source._subscribe(new FindValueSubscriber(observer, this.predicate, this.source, this.yieldIndex, this.thisArg));
+    return source.subscribe(new FindValueSubscriber(observer, this.predicate, this.source, this.yieldIndex, this.thisArg));
   }
 }
 

src/operator/first.ts

@@ -87,7 +87,7 @@ class FirstOperator<T, R> implements Operator<T, R> {
   }
 
   call(observer: Subscriber<R>, source: any): any {
-    return source._subscribe(new FirstSubscriber(observer, this.predicate, this.resultSelector, this.defaultValue, this.source));
+    return source.subscribe(new FirstSubscriber(observer, this.predicate, this.resultSelector, this.defaultValue, this.source));
   }
 }
 

src/operator/groupBy.ts

@@ -56,7 +56,7 @@ class GroupByOperator<T, K, R> implements Operator<T, GroupedObservable<K, R>> {
   }
 
   call(subscriber: Subscriber<GroupedObservable<K, R>>, source: any): any {
-    return source._subscribe(new GroupBySubscriber(
+    return source.subscribe(new GroupBySubscriber(
       subscriber, this.keySelector, this.elementSelector, this.durationSelector, this.subjectSelector
     ));
   }

src/operator/ignoreElements.ts

@@ -19,7 +19,7 @@ export function ignoreElements<T>(this: Observable<T>): Observable<T> {
 
 class IgnoreElementsOperator<T, R> implements Operator<T, R> {
   call(subscriber: Subscriber<R>, source: any): any {
-    return source._subscribe(new IgnoreElementsSubscriber(subscriber));
+    return source.subscribe(new IgnoreElementsSubscriber(subscriber));
   }
 }
 

src/operator/isEmpty.ts

@@ -17,7 +17,7 @@ export function isEmpty<T>(this: Observable<T>): Observable<boolean> {
 
 class IsEmptyOperator implements Operator<any, boolean> {
   call (observer: Subscriber<boolean>, source: any): any {
-    return source._subscribe(new IsEmptySubscriber(observer));
+    return source.subscribe(new IsEmptySubscriber(observer));
   }
 }
 

src/operator/last.ts

@@ -56,7 +56,7 @@ class LastOperator<T, R> implements Operator<T, R> {
   }
 
   call(observer: Subscriber<R>, source: any): any {
-    return source._subscribe(new LastSubscriber(observer, this.predicate, this.resultSelector, this.defaultValue, this.source));
+    return source.subscribe(new LastSubscriber(observer, this.predicate, this.resultSelector, this.defaultValue, this.source));
   }
 }
 

src/operator/map.ts

@@ -47,7 +47,7 @@ export class MapOperator<T, R> implements Operator<T, R> {
   }
 
   call(subscriber: Subscriber<R>, source: any): any {
-    return source._subscribe(new MapSubscriber(subscriber, this.project, this.thisArg));
+    return source.subscribe(new MapSubscriber(subscriber, this.project, this.thisArg));
   }
 }
 

src/operator/mapTo.ts

@@ -41,7 +41,7 @@ class MapToOperator<T, R> implements Operator<T, R> {
   }
 
   call(subscriber: Subscriber<R>, source: any): any {
-    return source._subscribe(new MapToSubscriber(subscriber, this.value));
+    return source.subscribe(new MapToSubscriber(subscriber, this.value));
   }
 }
 

src/operator/materialize.ts

@@ -46,7 +46,7 @@ export function materialize<T>(this: Observable<T>): Observable<Notification<T>>
 
 class MaterializeOperator<T> implements Operator<T, Notification<T>> {
   call(subscriber: Subscriber<Notification<T>>, source: any): any {
-    return source._subscribe(new MaterializeSubscriber(subscriber));
+    return source.subscribe(new MaterializeSubscriber(subscriber));
   }
 }
 

src/operator/mergeAll.ts

@@ -58,7 +58,7 @@ export class MergeAllOperator<T> implements Operator<Observable<T>, T> {
   }
 
   call(observer: Observer<T>, source: any): any {
-    return source._subscribe(new MergeAllSubscriber(observer, this.concurrent));
+    return source.subscribe(new MergeAllSubscriber(observer, this.concurrent));
   }
 }
 

src/operator/mergeMap.ts

@@ -77,7 +77,7 @@ export class MergeMapOperator<T, I, R> implements Operator<T, I> {
   }
 
   call(observer: Subscriber<I>, source: any): any {
-    return source._subscribe(new MergeMapSubscriber(
+    return source.subscribe(new MergeMapSubscriber(
       observer, this.project, this.resultSelector, this.concurrent
     ));
   }

src/operator/mergeMapTo.ts

@@ -74,7 +74,7 @@ export class MergeMapToOperator<T, I, R> implements Operator<Observable<T>, R> {
   }
 
   call(observer: Subscriber<R>, source: any): any {
-    return source._subscribe(new MergeMapToSubscriber(observer, this.ish, this.resultSelector, this.concurrent));
+    return source.subscribe(new MergeMapToSubscriber(observer, this.ish, this.resultSelector, this.concurrent));
   }
 }
 

src/operator/mergeScan.ts

@@ -29,7 +29,7 @@ export class MergeScanOperator<T, R> implements Operator<T, R> {
   }
 
   call(subscriber: Subscriber<R>, source: any): any {
-    return source._subscribe(new MergeScanSubscriber(
+    return source.subscribe(new MergeScanSubscriber(
       subscriber, this.project, this.seed, this.concurrent
     ));
   }