fix(retryWhen): fix internal unsubscriptions

Reimplement retryWhen operator. Fix retryWhen to tear down resources
(internal retried subscriptions and the subscription to the notifier
Observable) whenever the result is unsubscribed.

src/operators/retryWhen.ts

@@ -14,49 +14,108 @@ export default function retryWhen<T>(notifier: (errors: Observable<any>) => Obse
 
 class RetryWhenOperator<T, R> implements Operator<T, R> {
  constructor(protected notifier: (errors: Observable<any>) => Observable<any>,
- protected original: Observable<T>) {
+ protected source: Observable<T>) {
  }
 
  call(subscriber: Subscriber<T>): Subscriber<T> {
- return new RetryWhenSubscriber<T>(subscriber, this.notifier, this.original);
+ return new FirstRetryWhenSubscriber<T>(subscriber, this.notifier, this.source);
  }
 }
 
-class RetryWhenSubscriber<T> extends Subscriber<T> {
+class FirstRetryWhenSubscriber<T> extends Subscriber<T> {
+ lastSubscription: Subscription<T>;
+ notificationSubscription: Subscription<T>;
  errors: Subject<any>;
  retryNotifications: Observable<any>;
 
- constructor(destination: Subscriber<T>,
+ constructor(public destination: Subscriber<T>,
  public notifier: (errors: Observable<any>) => Observable<any>,
- public original: Observable<T>) {
- super(destination);
+ public source: Observable<T>) {
+ super(null);
+ this.lastSubscription = this;
  }
 
- _error(err: any) {
- if (!this.retryNotifications) {
- this.errors = new Subject();
- const notifications = tryCatch(this.notifier).call(this, this.errors);
- if (notifications === errorObject) {
- this.destination.error(errorObject.e);
- } else {
- this.retryNotifications = notifications;
- this.add(notifications._subscribe(new RetryNotificationSubscriber(this)));
+ _next(value: T) {
+ this.destination.next(value);
+ }
+
+ error(err?) {
+ if (!this.isUnsubscribed) {
+ super.unsubscribe();
+ if (!this.retryNotifications) {
+ this.errors = new Subject();
+ const notifications = tryCatch(this.notifier).call(this, this.errors);
+ if (notifications === errorObject) {
+ this.destination.error(errorObject.e);
+ } else {
+ this.retryNotifications = notifications;
+ const notificationSubscriber = new RetryNotificationSubscriber(this);
+ this.notificationSubscription = notifications.subscribe(notificationSubscriber);
+ }
  }
+ this.errors.next(err);
  }
- this.errors.next(err);
  }
 
- finalError(err: any) {
+ destinationError(err: any) {
+ this.tearDown();
  this.destination.error(err);
  }
 
+ _complete() {
+ this.destinationComplete();
+ }
+
+ destinationComplete() {
+ this.tearDown();
+ this.destination.complete();
+ }
+
+ unsubscribe() {
+ const lastSubscription = this.lastSubscription;
+ if (lastSubscription === this) {
+ super.unsubscribe();
+ } else {
+ this.tearDown();
+ }
+ }
+
+ tearDown() {
+ super.unsubscribe();
+ this.lastSubscription.unsubscribe();
+ const notificationSubscription = this.notificationSubscription;
+ if (notificationSubscription) {
+ notificationSubscription.unsubscribe();
+ }
+ }
+
  resubscribe() {
- this.original.subscribe(this);
+ this.lastSubscription.unsubscribe();
+ const nextSubscriber = new MoreRetryWhenSubscriber(this);
+ this.lastSubscription = this.source.subscribe(nextSubscriber);
+ }
+}
+
+class MoreRetryWhenSubscriber<T> extends Subscriber<T> {
+ constructor(private parent: FirstRetryWhenSubscriber<T>) {
+ super(null);
+ }
+
+ _next(value: T) {
+ this.parent.destination.next(value);
+ }
+
+ _error(err: any) {
+ this.parent.errors.next(err);
+ }
+
+ _complete() {
+ this.parent.destinationComplete();
  }
 }
 
 class RetryNotificationSubscriber<T> extends Subscriber<T> {
- constructor(public parent: RetryWhenSubscriber<any>) {
+ constructor(public parent: FirstRetryWhenSubscriber<any>) {
  super(null);
  }
 
@@ -65,10 +124,10 @@ class RetryNotificationSubscriber<T> extends Subscriber<T> {
  }
 
  _error(err: any) {
- this.parent.finalError(err);
+ this.parent.destinationError(err);
  }
 
  _complete() {
- this.parent.complete();
+ this.parent.destinationComplete();
  }
 }