fix(config.useDeprecatedSynchronousErrorThrowing): reentrant error th…

spec/Observable-spec.ts

@@ -5,6 +5,8 @@ import { Observer, TeardownLogic } from '../src/internal/types';
 import { cold, expectObservable, expectSubscriptions } from './helpers/marble-testing';
 import { map } from '../src/internal/operators/map';
 import * as HostReportErrorModule from '../src/internal/util/hostReportError';
+import { noop } from '../src/internal/util/noop';
+
 //tslint:disable-next-line
 require('./helpers/test-helper');
 
@@ -533,18 +535,64 @@ describe('Observable', () => {
       });
     });
 
-    describe('if config.useDeprecatedSynchronousThrowing === true', () => {
+    describe('config.useDeprecatedSynchronousErrorHandling', () => {
+
+      it('should log when it is set and unset', () => {
+        const _log = console.log;
+        const logCalledWith: any[][] = [];
+        console.log = (...args: any[]) => {
+          logCalledWith.push(args);
+        };
+
+        const _warn = console.warn;
+        const warnCalledWith: any[][] = [];
+        console.warn = (...args: any[]) => {
+          warnCalledWith.push(args);
+        };
+
+        Rx.config.useDeprecatedSynchronousErrorHandling = true;
+        expect(warnCalledWith.length).to.equal(1);
+
+        Rx.config.useDeprecatedSynchronousErrorHandling = false;
+        expect(logCalledWith.length).to.equal(1);
+
+        console.log = _log;
+        console.warn = _warn;
+      });
+    });
+
+    describe('if config.useDeprecatedSynchronousErrorHandling === true', () => {
       beforeEach(() => {
+        const _warn = console.warn;
+        console.warn = noop;
         Rx.config.useDeprecatedSynchronousErrorHandling = true;
+        console.warn = _warn;
       });
 
       it('should throw synchronously', () => {
         expect(() => Observable.throwError(new Error()).subscribe())
           .to.throw();
       });
 
+      it('should rethrow if sink has syncErrorThrowable = false', () => {
+        const observable = new Observable(observer => {
+          observer.next(1);
+        });
+
+        const sink = Rx.Subscriber.create(() => {
+          throw 'error!';
+        });
+
+        expect(() => {
+          observable.subscribe(sink);
+        }).to.throw('error!');
+      });
+
       afterEach(() => {
+        const _log = console.log;
+        console.log = noop;
         Rx.config.useDeprecatedSynchronousErrorHandling = false;
+        console.log = _log;
       });
     });
   });

spec/operators/repeatWhen-spec.ts

@@ -40,6 +40,7 @@ describe('Observable.prototype.repeatWhen', () => {
     let retried = false;
     const expected = [1, 2, 1, 2];
     let i = 0;
+    try {
     Observable.of(1, 2)
       .map((n: number) => {
         return n;
@@ -58,6 +59,9 @@ describe('Observable.prototype.repeatWhen', () => {
         expect(err).to.be.an('error', 'done');
         done();
       });
+    } catch (err) {
+      done(err);
+    }
   });
 
   it('should retry when notified and complete on returned completion', (done: MochaDone) => {

src/internal/Observable.ts

@@ -192,7 +192,7 @@ export class Observable<T> implements Subscribable<T> {
     if (operator) {
       operator.call(sink, this.source);
     } else {
-      sink.add(this.source ? this._subscribe(sink) : this._trySubscribe(sink));
+      sink.add(this.source || !sink.syncErrorThrowable ? this._subscribe(sink) : this._trySubscribe(sink));
     }
 
     if (config.useDeprecatedSynchronousErrorHandling) {

src/internal/Subscriber.ts

@@ -70,6 +70,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
         }
         if (typeof destinationOrNext === 'object') {
           if (destinationOrNext instanceof Subscriber) {
+            this.syncErrorThrowable = destinationOrNext.syncErrorThrowable;
             this.destination = (<Subscriber<any>> destinationOrNext);
             (<any> this.destination).add(this);
           } else {