fix(catchError): inner synchronous observables will properly terminate

spec/operators/catchError-spec.ts

@@ -458,7 +458,7 @@ describe('catchError operator', () => {
   });
 
   // TODO: fix firehose unsubscription
-  it.skip('should stop listening to a synchronous observable when unsubscribed', () => {
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
     const sideEffects: number[] = [];
     const synchronousObservable = new Observable(subscriber => {
       // This will check to see if the subscriber was closed on each loop

src/internal/operators/catchError.ts

@@ -125,9 +125,8 @@ export function catchError<T, O extends ObservableInput<any>>(
         }
       };
 
-      innerSub = source.subscribe({
-        next: (value) => subscriber.next(value),
-        error: (err) => {
+      innerSub = source.subscribe(
+        new CatchErrorSubscriber(subscriber, (err) => {
           handleError(err);
           if (handledResult) {
             if (innerSub) {
@@ -138,9 +137,8 @@ export function catchError<T, O extends ObservableInput<any>>(
               syncUnsub = true;
             }
           }
-        },
-        complete: () => subscriber.complete(),
-      });
+        })
+      );
 
       if (syncUnsub) {
         innerSub.unsubscribe();
@@ -153,3 +151,19 @@ export function catchError<T, O extends ObservableInput<any>>(
       return subscription;
     });
 }
+
+/**
+ * This must exist to ensure that the `closed` state of the inner subscriber is set at
+ * the proper time to ensure operators like `take` can stop the inner subscription if
+ * it is a synchronous firehose.
+ */
+class CatchErrorSubscriber<T, C> extends Subscriber<T> {
+  constructor(destination: Subscriber<T | C>, private onError: (err: any) => void) {
+    super(destination);
+  }
+
+  _error(err: any) {
+    this.onError(err);
+    this.unsubscribe();
+  }
+}