75

I have a list of items to parse, but the parsing of one of them can fail.

What is the "Rx-Way" to catch error but continue executing the sequence

Code Sample:

var observable = Rx.Observable.from([0,1,2,3,4,5])
.map(
  function(value){
      if(value == 3){
        throw new Error("Value cannot be 3");
      }
    return value;
  });

observable.subscribe(
  function(value){
  console.log("onNext " + value);
  },
  function(error){
    console.log("Error: " + error.message);
  },
  function(){
    console.log("Completed!");
  });
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.6/rx.all.js"></script>

What I want to do in a non-Rx-Way:

var items = [0,1,2,3,4,5];

for (var item in items){
  try{
    if(item == 3){
      throw new Error("Value cannot be 3");
    }
    console.log(item);
  }catch(error){
     console.log("Error: " + error.message);
  }
}

3
  • See Rx.Observable.onErrorResumeNext Commented Jul 29, 2016 at 2:40
  • Saw it, but can't make it work this way... could you please write an example? Commented Jul 29, 2016 at 2:54
  • 1
    See Ben Leshs talk about this especially at 23:43. Commented Jan 18, 2018 at 1:10

5 Answers 5

62

I would suggest that you use flatMap (now mergeMap in rxjs version 5) instead, which will let you collapse errors if you don't care about them. Effectively, you will create an inner Observable that can be swallowed if an error occurs. The advantage of this approach is that you can chain together operators and if an error occurs anywhere in the pipeline it will automatically get forwarded to the catch block.

const {from, iif, throwError, of, EMPTY} = rxjs;
const {map, flatMap, catchError} = rxjs.operators;

// A helper method to let us create arbitrary operators
const {pipe} = rxjs;

// Create an operator that will catch and squash errors
// This returns a function of the shape of Observable<T> => Observable<R>
const mapAndContinueOnError = pipe(
  //This will get skipped if upstream throws an error
  map(v => v * 2),
  catchError(err => {
    console.log("Caught Error, continuing")
    //Return an empty Observable which gets collapsed in the output
    return EMPTY;
  })
)

const observable = from([0, 1, 2, 3, 4, 5]).pipe(
  flatMap((value) => 
    iif(() => value != 3, 
      of(value), 
      throwError(new Error("Value cannot be 3"))
    ).pipe(mapAndContinueOnError)
  )
);

observable.subscribe(
  (value) => console.log("onNext " + value), (error) => console.log("Error: " + error.message), () => console.log("Completed!")
);
<script src="https://unpkg.com/[email protected]/dist/bundles/rxjs.umd.min.js"></script>

Sign up to request clarification or add additional context in comments.

6 Comments

what if an error occurs somewhere on a chain after the flatMap ?
@nonybrighto Then it would kill the stream and emit an error to the subscriber, unless there was another catch somewhere down stream.
This does not seem to work in RxJS v6. I tried migrating the code and the observable completes after the error is caught. Would you mind updating your solution. I have created a stackblitz here for reference. Stackblitz reproducable code.
@ManishShrestha There you go!
Love this answer for - the naked pipe function, the working rxJs example and the const destructuring of the imports. Every day's a school day! - oh and for the iif!
|
36

You need to switch to a new disposable stream, and if an error occurs within it will be disposed safely, and keep the original stream alive:

Rx.Observable.from([0,1,2,3,4,5])
    .switchMap(value => {

        // This is the disposable stream!
        // Errors can safely occur in here without killing the original stream

        return Rx.Observable.of(value)
            .map(value => {
                if (value === 3) {
                    throw new Error('Value cannot be 3');
                }
                return value;
            })
            .catch(error => {
                // You can do some fancy stuff here with errors if you like
                // Below we are just returning the error object to the outer stream
                return Rx.Observable.of(error);
            });

    })
    .map(value => {
        if (value instanceof Error) {
            // Maybe do some error handling here
            return `Error: ${value.message}`;
        }
        return value;
    })
    .subscribe(
      (x => console.log('Success', x)),
      (x => console.log('Error', x)),
      (() => console.log('Complete'))
    );
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.1/Rx.min.js"></script>

More info on this technique in this blog post: The Quest for Meatballs: Continue RxJS Streams When Errors Occur

2 Comments

How would that look with the lettable and pipe operators ?
BE CAREFUL, if the inner observable takes some time to emit it may never emit when using switchMap() as it cancels the previous observable. To test it yourself replace Rx.Observable.of(value) of this answer with Rx.Observable.from(new Promise((resolve, reject) => setTimeout(() => resolve(i)))). The result will only be: "Success 5, Complete". Replacing the switchMap() with mergeMap() fixes the problem and gives the expected results normally.
27

To keep your endlessObservable$ from dying you can put your failingObservable$ in a higher-order mapping operator (e.g. switchMap, concatMap, exhaustMap...) and swallow the error there by terminating the inner stream with an EMPTY observable returning no values.

endlessObservable$.pipe(
  switchMap(() => failingObservable$.pipe(catchError(error => EMPTY)))
);

6 Comments

Trying to keep it as simple as possible for people looking for a working solution.
a more illustrative example of this "double wrapping" at learnrxjs.io/operators/error_handling/catch.html (example no. 3) or stackblitz.com/edit/rxjs-catcherror-withmapoperators
This will not cancel unfinished failingObservable$ when endlessObservable$ emits?
@isevcik yes it will - it's just an example; you can use any of the other higher-order mapping operators depending on behavior you need.
emtpy() is now deprecated, EMPTY should be used instead.
|
1

In a case where you don't want or can't access the inner observable that causes the error, you can do something like this :

Using RxJS 7 :

const numbers$ = new Subject();

numbers$
  .pipe(
    tap(value => {
      if (value === 3) {
        throw new Error('Value cannot be 3');
      }
    }),
    tap(value => {
      console.log('Value:', value);
    }),
    catchError((err, caught) => {
      console.log('Error:', err.message);
      return caught;
    })
  )
  .subscribe();

for (let n = 1; n <= 10; n++) {
  numbers$.next(n);
}

What is interesting here is the "caught" argument in the catchError operator that can be returned. https://rxjs.dev/api/operators/catchError

It only works when the source observable is Hot.

In my case, I use redux-observable and I wanted a way to handle my errors in a single place.

I came up with this :

const allEpics = combineEpics(
    // all my epics
);

export const rootEpic = (action$, state$, dependencies) => {
  return allEpics(action$, state$, dependencies).pipe(
    catchError((err, caught) => {
        if (err instanceof MyError) {
        return concat(of(displayAlert(err.message)), caught);
      }
      throw err;
    })
  );
};

If any of my epic throw a "MyError" exception, It will be caught and another action will be dispatched.

Comments

-3

You can actually use try/catch inside your map function to handle the error. Here is the code snippet

var source = Rx.Observable.from([0, 1, 2, 3, 4, 5])
    .map(
        function(value) {
            try {
                if (value === 3) {
                    throw new Error("Value cannot be 3");
                }
                return value;

            } catch (error) {
                console.log('I caught an error');
                return undefined;
            }
        })
    .filter(function(x) {
        return x !== undefined; });


source.subscribe(
    function(value) {
        console.log("onNext " + value);
    },
    function(error) {
        console.log("Error: " + error.message);
    },
    function() {
        console.log("Completed!");
    });
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.6/rx.all.js"></script>

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.