feat(OuterSubscriber): notifyNext passes innersubscriber when next emits

- notifyNext() passes inner subscriber as same as notifyError(), notifyComplete() does, supports subscription management closes #1250
ReactiveX · Jan 27, 2016 · 1df8928 · 1df8928
1 parent f3e2f73
commit 1df8928
Show file tree

Hide file tree

Showing 23 changed files with 96 additions and 31 deletions.
diff --git a/src/InnerSubscriber.ts b/src/InnerSubscriber.ts
@@ -8,16 +8,16 @@ export class InnerSubscriber<T, R> extends Subscriber<R> {
     super();
   }
 
-  protected _next(value: R) {
-    this.parent.notifyNext(this.outerValue, value, this.outerIndex, this.index++);
+  protected _next(value: R): void {
+    this.parent.notifyNext(this.outerValue, value, this.outerIndex, this.index++, this);
   }
 
-  protected _error(error: any) {
+  protected _error(error: any): void {
     this.parent.notifyError(error, this);
     this.unsubscribe();
   }
 
-  protected _complete() {
+  protected _complete(): void {
     this.parent.notifyComplete(this);
     this.unsubscribe();
   }

diff --git a/src/OuterSubscriber.ts b/src/OuterSubscriber.ts
@@ -2,7 +2,9 @@ import {Subscriber} from './Subscriber';
 import {InnerSubscriber} from './InnerSubscriber';
 
 export class OuterSubscriber<T, R> extends Subscriber<T> {
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this.destination.next(innerValue);
   }
 

diff --git a/src/operator/buffer.ts b/src/operator/buffer.ts
@@ -3,6 +3,7 @@ import {Subscriber} from '../Subscriber';
 import {Observable} from '../Observable';
 
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -44,7 +45,9 @@ class BufferSubscriber<T, R> extends OuterSubscriber<T, R> {
     this.buffer.push(value);
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     const buffer = this.buffer;
     this.buffer = [];
     this.destination.next(buffer);

diff --git a/src/operator/bufferWhen.ts b/src/operator/bufferWhen.ts
@@ -6,6 +6,7 @@ import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -60,7 +61,9 @@ class BufferWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
     this.subscribing = false;
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this.openBuffer();
   }
 

diff --git a/src/operator/combineLatest.ts b/src/operator/combineLatest.ts
@@ -6,6 +6,7 @@ import {isScheduler} from '../util/isScheduler';
 import {Operator} from '../Operator';
 import {Subscriber} from '../Subscriber';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -116,9 +117,11 @@ export class CombineLatestSubscriber<T, R> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(observable: any, value: R, outerIndex: number, innerIndex: number) {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     const values = this.values;
-    values[outerIndex] = value;
+    values[outerIndex] = innerValue;
     const toRespond = this.toRespond;
 
     if (toRespond.length > 0) {

diff --git a/src/operator/debounce.ts b/src/operator/debounce.ts
@@ -7,6 +7,7 @@ import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -68,7 +69,9 @@ class DebounceSubscriber<T, R> extends OuterSubscriber<T, R> {
     this.destination.complete();
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this.emitValue();
   }
 

diff --git a/src/operator/exhaustMap.ts b/src/operator/exhaustMap.ts
@@ -4,6 +4,7 @@ import {Subscriber} from '../Subscriber';
 import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -64,7 +65,9 @@ class SwitchFirstMapSubscriber<T, R, R2> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     const { resultSelector, destination } = this;
     if (resultSelector) {
       const result = tryCatch(resultSelector)(outerValue, innerValue, outerIndex, innerIndex);

diff --git a/src/operator/expand.ts b/src/operator/expand.ts
@@ -6,6 +6,7 @@ import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 import {Subscription} from '../Subscription';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -96,7 +97,9 @@ export class ExpandSubscriber<T, R> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this._next(innerValue);
   }
 

diff --git a/src/operator/mergeMap.ts b/src/operator/mergeMap.ts
@@ -4,6 +4,7 @@ import {Subscriber} from '../Subscriber';
 import {Subscription} from '../Subscription';
 import {subscribeToResult} from '../util/subscribeToResult';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 
 /**
  * Returns an Observable that emits items based on applying a function that you supply to each item emitted by the
@@ -80,7 +81,9 @@ export class MergeMapSubscriber<T, R, R2> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     if (this.resultSelector) {
       this._notifyResultSelector(outerValue, innerValue, outerIndex, innerIndex);
     } else {
@@ -109,4 +112,4 @@ export class MergeMapSubscriber<T, R, R2> extends OuterSubscriber<T, R> {
       this.destination.complete();
     }
   }
-}
+}
diff --git a/src/operator/mergeMapTo.ts b/src/operator/mergeMapTo.ts
@@ -6,6 +6,7 @@ import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 import {Subscription} from '../Subscription';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 export function mergeMapTo<T, R, R2>(observable: Observable<R>,
@@ -67,7 +68,9 @@ export class MergeMapToSubscriber<T, R, R2> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     const { resultSelector, destination } = this;
     if (resultSelector) {
       const result = tryCatch(resultSelector)(outerValue, innerValue, outerIndex, innerIndex);

diff --git a/src/operator/mergeScan.ts b/src/operator/mergeScan.ts
@@ -6,6 +6,7 @@ import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 import {subscribeToResult} from '../util/subscribeToResult';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 
 export function mergeScan<T, R>(project: (acc: R, value: T) => Observable<R>,
                                 seed: R,
@@ -70,7 +71,9 @@ export class MergeScanSubscriber<T, R> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     const { destination } = this;
     this.acc = innerValue;
     this.hasValue = true;

diff --git a/src/operator/race.ts b/src/operator/race.ts
@@ -5,6 +5,7 @@ import {Operator} from '../Operator';
 import {Subscriber} from '../Subscriber';
 import {Subscription} from '../Subscription';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -79,7 +80,9 @@ export class RaceSubscriber<T, R> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(observable: any, value: R, outerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     if (!this.hasFirst) {
       this.hasFirst = true;
 
@@ -95,6 +98,6 @@ export class RaceSubscriber<T, R> extends OuterSubscriber<T, R> {
       this.subscriptions = null;
     }
 
-    this.destination.next(value);
+    this.destination.next(innerValue);
   }
 }
diff --git a/src/operator/retryWhen.ts b/src/operator/retryWhen.ts
@@ -7,6 +7,7 @@ import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -92,7 +93,9 @@ class RetryWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
     this.retries = null;
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
 
     const { errors, retries, retriesSubscription } = this;
     this.errors = null;

diff --git a/src/operator/sample.ts b/src/operator/sample.ts
@@ -3,6 +3,7 @@ import {Observable} from '../Observable';
 import {Subscriber} from '../Subscriber';
 
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -43,7 +44,9 @@ class SampleSubscriber<T, R> extends OuterSubscriber<T, R> {
     this.hasValue = true;
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this.emitValue();
   }
 

diff --git a/src/operator/skipUntil.ts b/src/operator/skipUntil.ts
@@ -3,6 +3,7 @@ import {Subscriber} from '../Subscriber';
 import {Observable} from '../Observable';
 
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -53,7 +54,9 @@ class SkipUntilSubscriber<T, R> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this.hasValue = true;
   }
 

diff --git a/src/operator/switch.ts b/src/operator/switch.ts
@@ -2,6 +2,7 @@ import {Operator} from '../Operator';
 import {Subscriber} from '../Subscriber';
 import {Subscription} from '../Subscription';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -60,7 +61,9 @@ class SwitchSubscriber<T, R> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: any): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this.destination.next(innerValue);
   }
 

diff --git a/src/operator/switchMap.ts b/src/operator/switchMap.ts
@@ -3,6 +3,7 @@ import {Observable} from '../Observable';
 import {Subscriber} from '../Subscriber';
 import {Subscription} from '../Subscription';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 /**
@@ -83,7 +84,9 @@ class SwitchMapSubscriber<T, R, R2> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     if (this.resultSelector) {
       this._tryNotifyNext(outerValue, innerValue, outerIndex, innerIndex);
     } else {

diff --git a/src/operator/switchMapTo.ts b/src/operator/switchMapTo.ts
@@ -5,6 +5,7 @@ import {Subscription} from '../Subscription';
 import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 export function switchMapTo<T, R, R2>(observable: Observable<R>,
@@ -63,7 +64,9 @@ class SwitchMapToSubscriber<T, R, R2> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number) {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     const { resultSelector, destination } = this;
     if (resultSelector) {
       const result = tryCatch(resultSelector)(outerValue, innerValue, outerIndex, innerIndex);

diff --git a/src/operator/takeUntil.ts b/src/operator/takeUntil.ts
@@ -3,6 +3,7 @@ import {Observable} from '../Observable';
 import {Subscriber} from '../Subscriber';
 
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 export function takeUntil<T>(notifier: Observable<any>): Observable<T> {
@@ -26,7 +27,9 @@ class TakeUntilSubscriber<T, R> extends OuterSubscriber<T, R> {
     this.add(subscribeToResult(this, notifier));
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this.complete();
   }
 

diff --git a/src/operator/throttle.ts b/src/operator/throttle.ts
@@ -6,6 +6,7 @@ import {Subscription} from '../Subscription';
 import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 import {OuterSubscriber} from '../OuterSubscriber';
+import {InnerSubscriber} from '../InnerSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
 
 export function throttle<T>(durationSelector: (value: T) => Observable<number> | Promise<number>): Observable<T> {
@@ -50,7 +51,9 @@ class ThrottleSubscriber<T, R> extends OuterSubscriber<T, R> {
     }
   }
 
-  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
+  notifyNext(outerValue: T, innerValue: R,
+             outerIndex: number, innerIndex: number,
+             innerSub: InnerSubscriber<T, R>): void {
     this._unsubscribe();
   }