fix(Observable): errors thrown during subscription are now properly s…

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