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,28 @@
+import { expect } from 'chai';
+import { noop, Subscriber } from 'rxjs';
+import { empty } from 'rxjs/internal/Observer';
+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 observer', () => {
+    expect(canReportError(empty)).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,8 @@ 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 { consoleWarn } from './util/consoleWarn';
 import { toSubscriber } from './util/toSubscriber';
 import { iif } from './observable/iif';
 import { throwError } from './observable/throwError';
@@ -226,7 +228,11 @@ export class Observable<T> implements Subscribable<T> {
         sink.syncErrorThrown = true;
         sink.syncErrorValue = err;
       }
-      sink.error(err);
+      if (canReportError(sink)) {
+        sink.error(err);
+      } else {
+        consoleWarn(err);
+      }
     }
   }
 

src/internal/observable/bindCallback.ts

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

src/internal/observable/bindNodeCallback.ts

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

src/internal/util/canReportError.ts

@@ -0,0 +1,23 @@
+import { Subscriber } from '../Subscriber';
+import { rxSubscriber as rxSubscriberSymbol } from '../symbol/rxSubscriber';
+import { ErrorObserver } from '../types';
+
+/**
+ * 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: ErrorObserver<any>): boolean {
+  while (observer) {
+    const { closed, destination, isStopped } = observer as any;
+    if (closed || isStopped) {
+      return false;
+    } else if (destination instanceof Subscriber || (destination && destination[rxSubscriberSymbol])) {
+      observer = destination;
+    } else {
+      observer = null;
+    }
+  }
+  return true;
+}

src/internal/util/consoleWarn.ts

@@ -0,0 +1,7 @@
+export function consoleWarn(...args: any[]): void {
+  if (console.warn) {
+    console.warn(...args);
+  } else {
+    console.log(...args);
+  }
+}