Hi I have the following code and I would like to know how to prevent the main (upstream) Observable from getting deleted when an error is thrown.
How can I change the following code so that all numbers expect '4' get displayed?
I am looking for a general pattern solution that would work in other cases with different operators. This is the simplest case I could come up with.
const Rx = require('rxjs/Rx');
function checkValue(n) {
if(n === 4) {
throw new Error("Bad value");
}
return true;
}
const source = Rx.Observable.interval(100).take(10);
source.filter(x => checkValue(x))
.catch(err => Rx.Observable.empty())
.subscribe(v => console.log(v));
RxJS catchError() operator is an error-handling operator used to handle and take care of catching errors on the source observable by returning a new observable or an error.
The catchError operator is going to take the error and pass it to the error handling function. That function is expected to return an Observable which is going to be a replacement Observable for the stream that just errored out.
Good Error Handling Always put the “catchError” operator inside a switchMap (or similar) so that it only ends the API call stream and then returns the stream to the switchMap, which continues the Observable.
You will want to keep the source observable running, but if you let the error happen on the main event stream it will collapse the entire observable and you will no longer receive items.
The solution involves creating a separated stream where you can filter and catch without letting the upstream pipe collapse.
const Rx = require('rxjs/Rx');
function checkValue(n) {
if(n === 4) {
throw new Error("Bad value");
}
return true;
}
const source = Rx.Observable.interval(100).take(10);
source
// pass the item into the projection function of the switchMap operator
.switchMap(x => {
// we create a new stream of just one item
// this stream is created for every item emitted by the source observable
return Observable.of(x)
// now we run the filter
.filter(checkValue)
// we catch the error here within the projection function
// on error this upstream pipe will collapse, but that is ok because it starts within this function and will not effect the source
// the downstream operators will never see the error so they will also not be effect
.catch(err => Rx.Observable.empty());
})
.subscribe(v => console.log(v));
You could also use the second argument passed into the catch selector to restart the observable source, but this will start it as though it hasn't run before.
const Rx = require('rxjs/Rx');
function checkValue(n) {
if(n === 4) {
throw new Error("Bad value");
}
return true;
}
const source = Rx.Observable.interval(100).take(10);
source.filter(x => checkValue(x))
.catch((err, source) => source)
.subscribe(v => console.log(v));
But this does not achieve the desired effect. You will see a stream that emits 1..3 repeatedly until the end of time... or you shutdown the script. Which ever comes first. (this is essential what .retry()
does)
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With