fix(subscribe): ignore syncError when deprecated (#3749)

* test(catchError): add failing test

* fix(subscribe): ignore syncError when deprecated

Unless using the deprecated syncError handling, the syncErrorThrowable
flag should be ignored.

spec/operators/catch-spec.ts

@@ -1,5 +1,5 @@
 import { expect } from 'chai';
-import { of, throwError, EMPTY, from } from 'rxjs';
+import { concat, Observable, of, throwError, EMPTY, from } from 'rxjs';
 import { catchError, map, mergeMap } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import * as sinon from 'sinon';
@@ -285,6 +285,21 @@ describe('catchError operator', () => {
     });
   });
 
+  it('should catch errors throw from within the constructor', () => {
+    // See https://github.com/ReactiveX/rxjs/issues/3740
+    const source = concat(
+      new Observable<string>(o => {
+        o.next('a');
+        throw 'kaboom';
+      }).pipe(
+        catchError(_ => of('b'))
+      ),
+      of('c')
+    );
+    const expected = '(abc|)';
+    expectObservable(source).toBe(expected);
+  });
+
   context('fromPromise', () => {
     type SetTimeout = (callback: (...args: any[]) => void, ms: number, ...args: any[]) => NodeJS.Timer;
 

src/internal/Observable.ts

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