fix(forEach): ensure that teardown logic is called when nextHandler throws

fixes #1411

Author: benlesh

spec/Observable-spec.ts

@@ -80,6 +80,62 @@ describe('Observable', () => {
       })
       .then(done);
     });
+
+    it('should handle a synchronous throw from the next handler and tear down', (done: DoneSignature) => {
+      let unsubscribeCalled = false;
+      const syncObservable = new Observable<number>((observer: Rx.Observer<number>) => {
+        observer.next(1);
+        observer.next(2);
+        observer.next(3);
+
+        return () => {
+          unsubscribeCalled = true;
+        };
+      });
+
+      const results = [];
+      syncObservable.forEach((x) => {
+        results.push(x);
+        if (x === 2) {
+          throw new Error('I told, you Bobby Boucher, twos are the debil!');
+        }
+      }).then(
+        () => done.fail(),
+        (err) => {
+          results.push(err);
+          expect(results).toEqual([1, 2, new Error('I told, you Bobby Boucher, twos are the debil!')]);
+          expect(unsubscribeCalled).toBe(true);
+          done();
+        });
+    });
+
+    it('should handle an asynchronous throw from the next handler and tear down', (done: DoneSignature) => {
+      let unsubscribeCalled = false;
+      const syncObservable = new Observable<number>((observer: Rx.Observer<number>) => {
+        let i = 1;
+        const id = setInterval(() => observer.next(i++));
+
+        return () => {
+          clearInterval(id);
+          unsubscribeCalled = true;
+        };
+      });
+
+      const results = [];
+      syncObservable.forEach((x) => {
+        results.push(x);
+        if (x === 2) {
+          throw new Error('I told, you Bobby Boucher, twos are the debil!');
+        }
+      }).then(
+        () => done.fail(),
+        (err) => {
+          results.push(err);
+          expect(results).toEqual([1, 2, new Error('I told, you Bobby Boucher, twos are the debil!')]);
+          expect(unsubscribeCalled).toBe(true);
+          done();
+        });
+    });
   });
 
   describe('subscribe', () => {

src/Observable.ts

@@ -6,8 +6,6 @@ import {root} from './util/root';
 import {CoreOperators} from './CoreOperators';
 import {SymbolShim} from './util/SymbolShim';
 import {toSubscriber} from './util/toSubscriber';
-import {tryCatch} from './util/tryCatch';
-import {errorObject} from './util/errorObject';
 
 import {combineLatestStatic} from './operator/combineLatest';
 import {concatStatic} from './operator/concat';
@@ -230,13 +228,27 @@ export class Observable<T> implements CoreOperators<T>  {
       throw new Error('no Promise impl found');
     }
 
-    const source = this;
-
     return new PromiseCtor<void>((resolve, reject) => {
-      source.subscribe((value: T) => {
-        const result: any = tryCatch(next)(value);
-        if (result === errorObject) {
-          reject(errorObject.e);
+      const subscription = this.subscribe((value) => {
+        if (subscription) {
+          // if there is a subscription, then we can surmise
+          // the next handling is asynchronous. Any errors thrown
+          // need to be rejected explicitly and unsubscribe must be
+          // called manually
+          try {
+            next(value);
+          } catch (err) {
+            reject(err);
+            subscription.unsubscribe();
+          }
+        } else {
+          // if there is NO subscription, then we're getting a nexted
+          // value synchronously during subscription. We can just call it.
+          // If it errors, Observable's `subscribe` imple will ensure the
+          // unsubscription logic is called, then synchronously rethrow the error.
+          // After that, Promise will trap the error and send it
+          // down the rejection path.
+          next(value);
         }
       }, reject, resolve);
     });
@@ -369,4 +381,4 @@ export class Observable<T> implements CoreOperators<T>  {
   [SymbolShim.observable]() {
     return this;
   }
-}
+}