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?

After refactoring you moved `map` out of `switchMap` projection, so any error will close the outer stream. Something like: `switchMap(([payload, cultureCode]) => this.dataService.loadData(payload, cultureCode).pipe(map..., catch...))` should do the job. — artur grzesiak, Nov 15 '17 at 17:55
Works great @arturgrzesiak! Post it as an answer and I'll accept it! :) — Daniel B, Nov 16 '17 at 08:31

score 70 · Accepted Answer · edited Nov 21 '22 at 22:06

70

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;
         })
      )
  )
);

edited Nov 21 '22 at 22:06

Get Off My Lawn

34,175
38
176
338

answered Nov 16 '17 at 08:51

artur grzesiak

20,230
5
46
56

1

This works exactly as I intended, though `catch` has been renamed to `catchError`! Thanks! – Daniel B Nov 16 '17 at 09:02
7

I had to do: `import { empty } from 'rxjs;` and then `empty()` in rjxs v6 – DauleDK Jun 28 '18 at 08:00
1

FYI, empty() has been deprecated in favor of EMPTY constant: import {EMPTY} from "rxjs/internal/observable/empty" – jk. Oct 12 '18 at 19:17
1

@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. – artur grzesiak Oct 12 '18 at 19:21
@artur grzesiak Link to what I saw: https://github.com/ReactiveX/rxjs/blob/master/src/internal/observable/empty.ts It does seem weird, but it was the only one not deprecated in intellij. I'm at rxjs-compat@6.3.2 and rxjs@6.2.0.Not sure if I need both of those or not.. – jk. Oct 12 '18 at 21:11
... and what if "this.store.select(getCultureCode)" the other source were to error? – redevill Jul 01 '22 at 00:16

score 2 · Answer 2 · answered Apr 21 '21 at 17:10

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))
      )
  )
);

Catch error in combined pipe of pipeable rxjs operators

2 Answers2

Linked