fix(catch): fix catch to dispose old subscriptions

Fix catch operator to not have anymore a shared underlying Subscription, and instead reset the
subscription for each new observable replacing the caught error. This fixes a potential memory leak
if catch is used as an infinite retry, because subscriptions would be retained since the beginning,
and would increasing each time a catch is performed.

Resolves issue #763.

spec/operators/catch-spec.js

@@ -25,7 +25,7 @@ describe('Observable.prototype.catch()', function () {
 
  it('should catch error and replace it with a cold Observable', function () {
  var e1 = hot('--a--b--#----| ');
- var e1subs = '^ !';
+ var e1subs = '^ ! ';
  var e2 = cold( '1-2-3-4-5-|');
  var e2subs = ' ^ !';
  var expected = '--a--b--1-2-3-4-5-|';
@@ -37,9 +37,23 @@ describe('Observable.prototype.catch()', function () {
  expectSubscriptions(e2.subscriptions).toBe(e2subs);
  });
 
+ it('should allow unsubscribing explicitly and early', function () {
+ var e1 = hot('--1-2-3-4-5-6---#');
+ var unsub = ' ! ';
+ var e1subs = '^ ! ';
+ var expected = '--1-2-3- ';
+
+ var result = e1.catch(function () {
+ return Observable.of('X', 'Y', 'Z');
+ });
+
+ expectObservable(result, unsub).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
  it('should catch error and replace it with a hot Observable', function () {
  var e1 = hot('--a--b--#----| ');
- var e1subs = '^ !';
+ var e1subs = '^ ! ';
  var e2 = hot('1-2-3-4-5-6-7-8-9-|');
  var e2subs = ' ^ !';
  var expected = '--a--b--5-6-7-8-9-|';
@@ -54,8 +68,8 @@ describe('Observable.prototype.catch()', function () {
  it('should catch and allow the cold observable to be repeated with the third ' +
  '(caught) argument', function () {
  var e1 = cold('--a--b--c--------| ');
- var subs = ['^ !',
- ' ^ !',
+ var subs = ['^ ! ',
+ ' ^ ! ',
  ' ^ !'];
  var expected = '--a--b----a--b----a--b--#';
 
@@ -81,7 +95,7 @@ describe('Observable.prototype.catch()', function () {
  it('should catch and allow the hot observable to proceed with the third ' +
  '(caught) argument', function () {
  var e1 = hot('--a--b--c----d---|');
- var subs = ['^ !',
+ var subs = ['^ ! ',
  ' ^ !'];
  var expected = '--a--b-------d---|';
 
@@ -132,41 +146,44 @@ describe('Observable.prototype.catch()', function () {
 
  it('should complete if you return Observable.empty()', function () {
  var e1 = hot('--a--b--#');
- var subs = '^ !';
+ var e1subs = '^ !';
+ var e2 = cold( '|');
+ var e2subs = ' (^!)';
  var expected = '--a--b--|';
 
- var result = e1.catch(function (err) {
- return Observable.empty();
- });
+ var result = e1.catch(function () { return e2; });
 
  expectObservable(result).toBe(expected);
- expectSubscriptions(e1.subscriptions).toBe(subs);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ expectSubscriptions(e2.subscriptions).toBe(e2subs);
  });
 
  it('should raise error if you return Observable.throw()', function () {
  var e1 = hot('--a--b--#');
- var subs = '^ !';
+ var e1subs = '^ !';
+ var e2 = cold( '#');
+ var e2subs = ' (^!)';
  var expected = '--a--b--#';
 
- var result = e1.catch(function (err) {
- return Observable.throw('error');
- });
+ var result = e1.catch(function () { return e2; });
 
  expectObservable(result).toBe(expected);
- expectSubscriptions(e1.subscriptions).toBe(subs);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ expectSubscriptions(e2.subscriptions).toBe(e2subs);
  });
 
  it('should never terminate if you return Observable.never()', function () {
  var e1 = hot('--a--b--#');
- var subs = '^ ';
+ var e1subs = '^ !';
+ var e2 = cold( '-');
+ var e2subs = ' ^';
  var expected = '--a--b---';
 
- var result = e1.catch(function (err) {
- return Observable.never();
- });
+ var result = e1.catch(function () { return e2; });
 
  expectObservable(result).toBe(expected);
- expectSubscriptions(e1.subscriptions).toBe(subs);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ expectSubscriptions(e2.subscriptions).toBe(e2subs);
  });
 
  it('should pass the error as the first argument', function (done) {

src/operators/catch.ts

@@ -1,6 +1,7 @@
 import {Operator} from '../Operator';
 import {Subscriber} from '../Subscriber';
 import {Observable} from '../Observable';
+import {Subscription} from '../Subscription';
 import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 
@@ -20,12 +21,9 @@ export function _catch<T>(selector: (err: any, caught: Observable<any>) => Obser
 }
 
 class CatchOperator<T, R> implements Operator<T, R> {
- selector: (err: any, caught: Observable<any>) => Observable<any>;
  caught: Observable<any>;
- source: Observable<T>;
 
- constructor(selector: (err: any, caught: Observable<any>) => Observable<any>) {
- this.selector = selector;
+ constructor(private selector: (err: any, caught: Observable<any>) => Observable<any>) {
  }
 
  call(subscriber: Subscriber<T>): Subscriber<T> {
@@ -34,23 +32,35 @@ class CatchOperator<T, R> implements Operator<T, R> {
 }
 
 class CatchSubscriber<T> extends Subscriber<T> {
- selector: (err: any, caught: Observable<any>) => Observable<any>;
- caught: Observable<any>;
+ private lastSubscription: Subscription<T>;
+
+ constructor(public destination: Subscriber<T>,
+ private selector: (err: any, caught: Observable<any>) => Observable<any>,
+ private caught: Observable<any>) {
+ super(null);
+ this.lastSubscription = this;
+ }
 
- constructor(destination: Subscriber<T>,
- selector: (err: any, caught: Observable<any>) => Observable<any>,
- caught: Observable<any>) {
- super(destination);
- this.selector = selector;
- this.caught = caught;
+ _next(value: T) {
+ this.destination.next(value);
  }
 
  _error(err) {
  const result = tryCatch(this.selector)(err, this.caught);
  if (result === errorObject) {
  this.destination.error(errorObject.e);
  } else {
- this.add(result.subscribe(this.destination));
+ this.lastSubscription.unsubscribe();
+ this.lastSubscription = result.subscribe(this.destination);
  }
  }
+
+ _complete() {
+ this.lastSubscription.unsubscribe();
+ this.destination.complete();
+ }
+
+ _unsubscribe() {
+ this.lastSubscription.unsubscribe();
+ }
 }