fix(Subscription): fix leaks caused by unsubscribe functions that throw

- adds a test for the issue

related #1256

spec/Subscription-spec.js

@@ -0,0 +1,39 @@
+/* globals describe, it, expect */
+var Rx = require('../dist/cjs/Rx');
+var Subscription = Rx.Subscription;
+var Observable = Rx.Observable;
+
+describe('Subscription', function () {
+  it('should not leak', function (done) {
+    var tearDowns = [];
+
+    var source1 = Observable.create(function (observer) {
+      return function () {
+        tearDowns.push(1);
+      };
+    });
+
+    var source2 = Observable.create(function (observer) {
+      return function () {
+        tearDowns.push(2);
+        throw new Error('oops, I am a bad unsubscribe!');
+      };
+    });
+
+    var source3 = Observable.create(function (observer) {
+      return function () {
+        tearDowns.push(3);
+      };
+    });
+
+    var subscription = Observable.merge(source1, source2, source3).subscribe();
+
+    setTimeout(function () {
+      expect(function () {
+        subscription.unsubscribe();
+      }).toThrow();
+      expect(tearDowns).toEqual([1, 2, 3]);
+      done();
+    });
+  });
+});

src/Rx.DOM.ts

@@ -110,7 +110,7 @@ import './add/operator/zip';
 import './add/operator/zipAll';
 
 /* tslint:disable:no-unused-variable */
-import {Subscription} from './Subscription';
+import {Subscription, UnsubscriptionError} from './Subscription';
 import {Subscriber} from './Subscriber';
 import {AsyncSubject} from './subject/AsyncSubject';
 import {ReplaySubject} from './subject/ReplaySubject';
@@ -159,5 +159,6 @@ export {
   Notification,
   EmptyError,
   ArgumentOutOfRangeError,
-  ObjectUnsubscribedError
+  ObjectUnsubscribedError,
+  UnsubscriptionError
 };

src/Rx.KitchenSink.ts

@@ -139,7 +139,7 @@ import './add/operator/zipAll';
 
 /* tslint:disable:no-unused-variable */
 import {Observer} from './Observer';
-import {Subscription} from './Subscription';
+import {Subscription, UnsubscriptionError} from './Subscription';
 import {Subscriber} from './Subscriber';
 import {AsyncSubject} from './subject/AsyncSubject';
 import {ReplaySubject} from './subject/ReplaySubject';
@@ -185,6 +185,7 @@ export {
     EmptyError,
     ArgumentOutOfRangeError,
     ObjectUnsubscribedError,
+    UnsubscriptionError,
     TestScheduler,
     VirtualTimeScheduler,
     TimeInterval,

src/Rx.ts

@@ -114,7 +114,7 @@ import './add/operator/zipAll';
 /* tslint:disable:no-unused-variable */
 import {Operator} from './Operator';
 import {Observer} from './Observer';
-import {Subscription} from './Subscription';
+import {Subscription, UnsubscriptionError} from './Subscription';
 import {Subscriber} from './Subscriber';
 import {AsyncSubject} from './subject/AsyncSubject';
 import {ReplaySubject} from './subject/ReplaySubject';
@@ -158,5 +158,6 @@ export {
     Notification,
     EmptyError,
     ArgumentOutOfRangeError,
-    ObjectUnsubscribedError
+    ObjectUnsubscribedError,
+    UnsubscriptionError
 };

src/Subscription.ts

@@ -1,6 +1,8 @@
 import {isArray} from './util/isArray';
 import {isObject} from './util/isObject';
 import {isFunction} from './util/isFunction';
+import {tryCatch} from './util/tryCatch';
+import {errorObject} from './util/errorObject';
 
 export class Subscription {
   public static EMPTY: Subscription = (function(empty: any){
@@ -17,6 +19,8 @@ export class Subscription {
   }
 
   unsubscribe(): void {
+    let hasErrors = false;
+    let errors: any[];
 
     if (this.isUnsubscribed) {
       return;
@@ -29,7 +33,11 @@ export class Subscription {
     (<any> this)._subscriptions = null;
 
     if (isFunction(_unsubscribe)) {
-      _unsubscribe.call(this);
+      let trial = tryCatch(_unsubscribe).call(this);
+      if (trial === errorObject) {
+        hasErrors = true;
+        (errors = errors || []).push(errorObject.e);
+      }
     }
 
     if (isArray(_subscriptions)) {
@@ -38,12 +46,26 @@ export class Subscription {
       const len = _subscriptions.length;
 
       while (++index < len) {
-        const subscription = _subscriptions[index];
-        if (isObject(subscription)) {
-          subscription.unsubscribe();
+        const sub = _subscriptions[index];
+        if (isObject(sub)) {
+          let trial = tryCatch(sub.unsubscribe).call(sub);
+          if (trial === errorObject) {
+            hasErrors = true;
+            errors = errors || [];
+            let err = errorObject.e;
+            if (err instanceof UnsubscriptionError) {
+              errors = errors.concat(err.errors);
+            } else {
+              errors.push(err);
+            }
+          }
         }
       }
     }
+
+    if (hasErrors) {
+      throw new UnsubscriptionError(errors);
+    }
   }
 
   add(subscription: Subscription | Function | void): void {
@@ -98,3 +120,10 @@ export class Subscription {
     }
   }
 }
+
+export class UnsubscriptionError extends Error {
+  constructor(public errors: any[]) {
+    super('unsubscriptoin error(s)');
+    this.name = 'UnsubscriptionError';
+  }
+}