RxJs catch error and continue

Question

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

Thanks in advance

Answer 1

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>

Answer 2

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