rxjs error handling > on catchError source stops emitting

Question

I'm a bit confused by the rxjs catchError operator. Here is a simple example using Angular:

(live demo here)

import { Component } from '@angular/core';
import { of, timer } from 'rxjs'
import { tap, catchError } from 'rxjs/operators'

@Component({
  selector: 'my-app',
  templateUrl: './app.component.html',
})
export class AppComponent {
  constructor() {
    const source$ =  timer(1000, 1000).pipe(tap(n => {
      if (n === 3) {
        throw new Error('n === 3')
      }
    }))

    this.value$ = source$.pipe(catchError(err => {
      return of(42)
    }))
  }

  value$
}

{{ value$ | async }}

The source$ observable to which the async pipe subscribes emits 0,1,2 and then errors. This error is caught in the catchError operator which swallows the error silently and emits 42. This I think I understand. However, the emission then stops (I was expecting 4,5,6,...). Can someone please explain why this does not occur and if there is any way of achieving such a behaviour?

This matters to me in practice in a situation such as below, where we load fresh data every time the route params emit. Supposing we navigate to a route that throws an error (e.g. 404, 500,...) I do not want the event stream to stop emitting as then I am left unable to navigate back to any other page. I just want to handle the error gracefully and keep listening for route changes.

this.foo$ = this._route.params.pipe(
  map(params => params.id),
  switchMap(id => this._fooServerClient.getOneById(id)),
  catchError(err => {
    return of(undefined)
  })
)

Mcmurry

Answer 1

It is by design of Observable if an exception occurs in an Observable pipeline, the respective observable [which throws an error] will be in error state and it cannot emit new/further value [i.e. it is like unsubscribe].

Now, to keep your outer observable live [i.e. keep emitting the new values] then handle the error in inner observable [i.e. use the catchError operator in inner observable pipeline like this:

this.foo$ = this._route.params
                .pipe(
                      map(params => params.id),
                      switchMap(id => {
                          return this._fooServerClient.getOneById(id)
                                     .pipe(
                                      catchError(err => {
                                        return of(undefined);
                                      })
                                     )
                        }),

                );

Having catchError in the inner observable pipeline will keep outer observable live [i.e. keep emitting the new value even if inner observable throws exception].

Socher

You can also return EMTY in catchError, so the subscribe function won't have to ignore undefined results. – Thew