fix(Observable): errors thrown during subscription are now properly sent down error channel (#2313)

- Fixes `Observable.create(fn)` and `new Observable(fn)` such that any error thrown in `fn` on subscription will be sent to the subscriber's error handler.
- Fixes a subject test that was relying on the errant behavior.

fixes #1833

Author: benlesh

spec/Observable-spec.ts

@@ -29,6 +29,18 @@ describe('Observable', () => {
     source.subscribe(function (x) { expect(x).to.equal(1); }, null, done);
   });
 
+  it('should send errors thrown in the constructor down the error path', (done) => {
+    new Observable((observer) => {
+      throw new Error('this should be handled');
+    })
+    .subscribe({
+      error(err) {
+        expect(err).to.deep.equal(new Error('this should be handled'));
+        done();
+      }
+    });
+  });
+
   describe('forEach', () => {
     it('should iterate and return a Promise', (done: MochaDone) => {
       const expected = [1, 2, 3];
@@ -582,6 +594,18 @@ describe('Observable.create', () => {
     });
     expect(called).to.be.true;
   });
+
+  it('should send errors thrown in the passed function down the error path', (done) => {
+    Observable.create((observer) => {
+      throw new Error('this should be handled');
+    })
+    .subscribe({
+      error(err) {
+        expect(err).to.deep.equal(new Error('this should be handled'));
+        done();
+      }
+    });
+  });
 });
 
 /** @test {Observable} */

spec/Subject-spec.ts

@@ -261,10 +261,8 @@ describe('Subject', () => {
     expect(() => {
       subject.subscribe(
         function (x) { results3.push(x); },
-        function (e) { results3.push('E'); },
-        () => { results3.push('C'); }
       );
-    }).to.throw();
+    }).to.throw(Rx.ObjectUnsubscribedError);
 
     expect(results1).to.deep.equal([1, 2, 3, 4, 5]);
     expect(results2).to.deep.equal([3, 4, 5]);

src/Observable.ts

@@ -95,7 +95,7 @@ export class Observable<T> implements Subscribable<T> {
     if (operator) {
       operator.call(sink, this.source);
     } else {
-      sink.add(this._subscribe(sink));
+      sink.add(this._trySubscribe(sink));
     }
 
     if (sink.syncErrorThrowable) {
@@ -108,6 +108,16 @@ export class Observable<T> implements Subscribable<T> {
     return sink;
   }
 
+  private _trySubscribe(sink: Subscriber<T>): TeardownLogic {
+    try {
+      return this._subscribe(sink);
+    } catch (err) {
+      sink.syncErrorThrown = true;
+      sink.syncErrorValue = err;
+      sink.error(err);
+    }
+  }
+
   /**
    * @method forEach
    * @param {Function} next a handler for each value emitted by the observable