fix: catch within innerSubscribe (#6186)

* test: add failing invalid ObservableInput tests

* fix: catch within innerSubscribe

Closes #5344

spec/operators/concatMap-spec.ts

@@ -691,4 +691,17 @@ describe('Observable.prototype.concatMap', () => {
         done(new Error('Subscriber complete handler not supposed to be called.'));
       });
   });
+
+  it('should report invalid observable inputs via error notifications', () => {
+    const e1 =    hot('--1|');
+    const e1subs =    '^ !';
+    const expected =  '--#';
+
+    const result = e1.pipe(concatMap(() => null as any));
+
+    expectObservable(result).toBe(expected, null, new TypeError(
+      "You provided 'null' where a stream was expected. You can provide an Observable, Promise, Array, or Iterable."
+    ));
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
 });

spec/operators/exhaustMap-spec.ts

@@ -434,4 +434,17 @@ describe('exhaustMap', () => {
     expectSubscriptions(x.subscriptions).toBe(xsubs);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
+
+  it('should report invalid observable inputs via error notifications', () => {
+    const e1 =    hot('--1|');
+    const e1subs =    '^ !';
+    const expected =  '--#';
+
+    const result = e1.pipe(exhaustMap(() => null as any));
+
+    expectObservable(result).toBe(expected, null, new TypeError(
+      "You provided 'null' where a stream was expected. You can provide an Observable, Promise, Array, or Iterable."
+    ));
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
 });

spec/operators/mergeMap-spec.ts

@@ -852,4 +852,17 @@ describe('mergeMap', () => {
       expectObservable(result).toBe(expected, undefined, noXError);
     });
   });
+
+  it('should report invalid observable inputs via error notifications', () => {
+    const e1 =    hot('--1|');
+    const e1subs =    '^ !';
+    const expected =  '--#';
+
+    const result = e1.pipe(mergeMap(() => null as any));
+
+    expectObservable(result).toBe(expected, null, new TypeError(
+      "You provided 'null' where a stream was expected. You can provide an Observable, Promise, Array, or Iterable."
+    ));
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
 });

spec/operators/switchMap-spec.ts

@@ -444,4 +444,17 @@ describe('switchMap', () => {
     expectSubscriptions(x.subscriptions).toBe(xsubs);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
+
+  it('should report invalid observable inputs via error notifications', () => {
+    const e1 =    hot('--1|');
+    const e1subs =    '^ !';
+    const expected =  '--#';
+
+    const result = e1.pipe(switchMap(() => null as any));
+
+    expectObservable(result).toBe(expected, null, new TypeError(
+      "You provided 'null' where a stream was expected. You can provide an Observable, Promise, Array, or Iterable."
+    ));
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
 });

src/internal/innerSubscribe.ts

@@ -110,5 +110,11 @@ export function innerSubscribe(result: any, innerSubscriber: Subscriber<any>): S
   if (result instanceof Observable) {
     return result.subscribe(innerSubscriber);
   }
-  return subscribeTo(result)(innerSubscriber) as Subscription;
+  let subscription: Subscription;
+  try {
+    subscription = subscribeTo(result)(innerSubscriber) as Subscription;
+  } catch (error) {
+    innerSubscriber.error(error);
+  }
+  return subscription;
 }