Catch error in combined pipe of pipeable rxjs operators

Question

We've just upgraded one of our applications to Angular 5, and started to transition into lettable operators as introduced in rxjs v5.5.

Because of this, we have rewritten our observable pipelines to the new syntax with the .pipe() operator.

Our previous code would look like this, with a .catch() inside the .switchMap() as to not interrupt the running of effects if an error is thrown.

@Effect()
loadData$ = this.actions$
.ofType(LOAD_DATA)
.map((action: LoadData) => action.payload)
.withLatestFrom(this.store.select(getCultureCode))
.switchMap(([payload, cultureCode]) => this.dataService.loadData(payload, cultureCode)
  .map(result => {
    if (!result) {
      return new LoadDataFailed('Could not fetch data!');
    } else {
      return new LoadDataSuccessful(result);
    }
  })
  .catch((err, caught) => {
    return Observable.empty();
  });
  );

In the case of an error thrown in the call to the dataService it would be caught and handled (simplified the error handling here).

With the new syntax and use of .pipe(), we now have this

@Effect()
loadData$ = this.actions$
.ofType(LOAD_DATA)
.pipe(
  map((action: LoadData) => action.payload),
  withLatestFrom(this.store.select(getCultureCode)),
  switchMap(([payload, cultureCode]) => this.dataService.loadData(payload, cultureCode)),
  map(result => {
    if (!result) {
      return new LoadDataFailed('Could not fetch data!');
    } else {
      return new LoadDataSuccessful(result);
    }
  })
  );

How can I in a similar fashion catch any thrown errors in the observable pipeline, using the new syntax?

Solution 1

After refactoring you moved map out of switchMap projection, so any error will close the outer stream. To keep both streams equivalent, you need to use pipe in the projection itself like that:

import { EMPTY } from 'rxjs;

// ...

@Effect()
loadData$ = this.actions$
.ofType(LOAD_DATA)
.pipe(
  map((action: LoadData) => action.payload),
  withLatestFrom(this.store.select(getCultureCode)),
  switchMap(([payload, cultureCode]) =>
    this.dataService.loadData(payload, cultureCode)
      .pipe(
         map(result => {
           if (!result) {
             return new LoadDataFailed('Could not fetch data!');
           } else {
             return new LoadDataSuccessful(result);
           }
          }),
         catchError((err, caught) => {
           return EMPTY;
         })
      )
  )
);

Solution 2

You can also do it that way.

import { of } from 'rxjs';

@Effect()
loadData$ = this.actions$
.ofType(LOAD_DATA)
.pipe(
  map((action: LoadData) => action.payload),
  withLatestFrom(this.store.select(getCultureCode)),
  switchMap(([payload, cultureCode]) =>
    this.dataService.loadData(payload, cultureCode)
      .pipe(
         map(result => {
           if (!result) {
             return new LoadDataFailed('Could not fetch data!');
           } else {
             return new LoadDataSuccessful(result);
           }
          }),
         catchError(err => of('error', err))
      )
  )
);

Comments

[Daniel B]

This works exactly as I intended, though catch has been renamed to catchError! Thanks!

[DauleDK]

I had to do: import { empty } from 'rxjs; and then empty() in rjxs v6

[jk.]

FYI, empty() has been deprecated in favor of EMPTY constant: import {EMPTY} from "rxjs/internal/observable/empty"

[artur grzesiak]

@jk. not sure about it - could you paste a link to a release note? I am on 6.3.3 and there is empty. Moreover importing directly from rxjs/internal seems bit weird to me.

[jk.]

@artur grzesiak Link to what I saw: github.com/ReactiveX/rxjs/blob/master/src/internal/observabl‌​e/… It does seem weird, but it was the only one not deprecated in intellij. I'm at [email protected] and [email protected] sure if I need both of those or not..

[redevill]

... and what if "this.store.select(getCultureCode)" the other source were to error?