fix(subscribe): don't swallow internal errors

spec/Observable-spec.ts

@@ -901,5 +901,20 @@ describe('Observable.lift', () => {
  ]);
  done();
  });
- });
+ });
+
+ it('should not swallow internal errors', () => {
+ const consoleStub = sinon.stub(console, 'warn');
+ try {
+ let source = new Observable<number>(observer => observer.next(42));
+ for (let i = 0; i < 10000; ++i) {
+ let base = source;
+ source = new Observable<number>(observer => base.subscribe(observer));
+ }
+ source.subscribe();
+ expect(consoleStub).to.have.property('called', true);
+ } finally {
+ consoleStub.restore();
+ }
+ });
 });

spec/observables/bindCallback-spec.ts

@@ -280,4 +280,18 @@ describe('bindCallback', () => {
  expect(calls).to.equal(0);
  });
  });
+
+ it('should not swallow post-callback errors', () => {
+ function badFunction(callback: (answer: number) => void): void {
+ callback(42);
+ throw new Error('kaboom');
+ }
+ const consoleStub = sinon.stub(console, 'warn');
+ try {
+ bindCallback(badFunction)().subscribe();
+ expect(consoleStub).to.have.property('called', true);
+ } finally {
+ consoleStub.restore();
+ }
+ });
 });

spec/observables/bindNodeCallback-spec.ts

@@ -278,4 +278,18 @@ describe('bindNodeCallback', () => {
  expect(results1).to.deep.equal([42, 'done']);
  expect(results2).to.deep.equal([42, 'done']);
  });
+
+ it('should not swallow post-callback errors', () => {
+ function badFunction(callback: (error: Error, answer: number) => void): void {
+ callback(null, 42);
+ throw new Error('kaboom');
+ }
+ const consoleStub = sinon.stub(console, 'warn');
+ try {
+ bindNodeCallback(badFunction)().subscribe();
+ expect(consoleStub).to.have.property('called', true);
+ } finally {
+ consoleStub.restore();
+ }
+ });
 });

spec/util/canReportError-spec.ts

@@ -0,0 +1,29 @@
+import { expect } from 'chai';
+import { noop, Subject, Subscriber } from 'rxjs';
+import { canReportError } from 'rxjs/internal/util/canReportError';
+
+describe('canReportError', () => {
+ it('should report errors to an observer if possible', () => {
+ const subscriber = new Subscriber<{}>(noop, noop);
+ expect(canReportError(subscriber)).to.be.true;
+ });
+
+ it('should not report errors to a stopped observer', () => {
+ const subscriber = new Subscriber<{}>(noop, noop);
+ subscriber.error(new Error('kaboom'));
+ expect(canReportError(subscriber)).to.be.false;
+ });
+
+ it('should not report errors to a closed subject', () => {
+ const subject = new Subject<{}>();
+ subject.unsubscribe();
+ expect(canReportError(subject)).to.be.false;
+ });
+
+ it('should not report errors an observer with a stopped destination', () => {
+ const destination = new Subscriber<{}>(noop, noop);
+ const subscriber = new Subscriber<{}>(destination);
+ destination.error(new Error('kaboom'));
+ expect(canReportError(subscriber)).to.be.false;
+ });
+});

src/internal/Observable.ts

@@ -2,6 +2,7 @@ import { Operator } from './Operator';
 import { Subscriber } from './Subscriber';
 import { Subscription } from './Subscription';
 import { TeardownLogic, OperatorFunction, PartialObserver, Subscribable } from './types';
+import { canReportError } from './util/canReportError';
 import { toSubscriber } from './util/toSubscriber';
 import { iif } from './observable/iif';
 import { throwError } from './observable/throwError';
@@ -226,7 +227,11 @@ export class Observable<T> implements Subscribable<T> {
  sink.syncErrorThrown = true;
  sink.syncErrorValue = err;
  }
- sink.error(err);
+ if (canReportError(sink)) {
+ sink.error(err);
+ } else {
+ console.warn(err);
+ }
  }
  }
 

src/internal/Subscriber.ts

@@ -325,6 +325,6 @@ export class SafeSubscriber<T> extends Subscriber<T> {
  }
 }
 
-function isTrustedSubscriber(obj: any) {
+export function isTrustedSubscriber(obj: any) {
  return obj instanceof Subscriber || ('_addParentTeardownLogic' in obj && obj[rxSubscriberSymbol]);
 }

src/internal/observable/bindCallback.ts

@@ -3,6 +3,7 @@ import { Observable } from '../Observable';
 import { AsyncSubject } from '../AsyncSubject';
 import { Subscriber } from '../Subscriber';
 import { map } from '../operators/map';
+import { canReportError } from '../util/canReportError';
 import { isArray } from '../util/isArray';
 import { isScheduler } from '../util/isScheduler';
 
@@ -204,7 +205,11 @@ export function bindCallback<T>(
  try {
  callbackFunc.apply(context, [...args, handler]);
  } catch (err) {
- subject.error(err);
+ if (canReportError(subject)) {
+ subject.error(err);
+ } else {
+ console.warn(err);
+ }
  }
  }
  return subject.subscribe(subscriber);

src/internal/observable/bindNodeCallback.ts

@@ -3,6 +3,7 @@ import { AsyncSubject } from '../AsyncSubject';
 import { Subscriber } from '../Subscriber';
 import { SchedulerAction, SchedulerLike } from '../types';
 import { map } from '../operators/map';
+import { canReportError } from '../util/canReportError';
 import { isScheduler } from '../util/isScheduler';
 import { isArray } from '../util/isArray';
 
@@ -198,7 +199,11 @@ export function bindNodeCallback<T>(
  try {
  callbackFunc.apply(context, [...args, handler]);
  } catch (err) {
- subject.error(err);
+ if (canReportError(subject)) {
+ subject.error(err);
+ } else {
+ console.warn(err);
+ }
  }
  }
  return subject.subscribe(subscriber);

src/internal/util/canReportError.ts

@@ -0,0 +1,22 @@
+import { isTrustedSubscriber, Subscriber } from '../Subscriber';
+import { Subject } from '../Subject';
+
+/**
+ * Determines whether the ErrorObserver is closed or stopped or has a
+ * destination that is closed or stopped - in which case errors will
+ * need to be reported via a different mechanism.
+ * @param observer the observer
+ */
+export function canReportError(observer: Subscriber<any> | Subject<any>): boolean {
+ while (observer) {
+ const { closed, destination, isStopped } = observer as any;
+ if (closed || isStopped) {
+ return false;
+ } else if (destination && isTrustedSubscriber(destination)) {
+ observer = destination;
+ } else {
+ observer = null;
+ }
+ }
+ return true;
+}