The Catch operator intercepts an onError
notification from the source Observable and, instead of passing it through to any
observers, replaces it with some other item or sequence of items, potentially allowing
the resulting Observable to terminate normally or not to terminate at all.
There are several variants of the Catch operator, and a variety of names used by different ReactiveX implementations to describe this operation, as you can see in the sections below.
In some ReactiveX implementations, there is an operator called something like
“OnErrorResumeNext” that behaves like a Catch
variant: specifically reacting to an onError notification from the source
Observable. In others, there is an operator with that name that behaves more like a
Concat variant: performing the concatenation operation
regardless of whether the source Observable terminates normally or with an error. This is
unfortunate and confusing, but something we have to live with.
RxClojure implements this operator as catch*. This operator takes two arguments,
both of which are functions of your choosing that take the exception raised by
onError as their single parameters. The first function is a predicate. If it
returns false, catch* passes the onError notification
unchanged to its observers. If it returns true, however, catch*
swallows the error, calls the second function (which returns an Observable), and passes along
the emissions and notifications from this new Observable to its observers.
You may replace the first function parameter (the predicate that evaluates the exception)
with a class object representing a variety of exception. If you do this, catch*
will treat it as equivalent to predicate that performs an instance? check to see
if the exception from the onError notification is an instance of the class
object. In other words:
ReactiveX
(->> my-observable
(catch* IllegalArgumentException
(fn [e] (rx/return 1)))
)is equivalent to:
(->> my-observable
(catch* (fn [e] (-> instance? IllegalArgumentException e))
(fn [e] (rx/return 1)))
)RxCpp does not implement the Catch operator.
RxGroovy implements the Catch operator in the same way as does RxJava. There are three distinct operators that provide this functionality:
onErrorReturnonErrorResumeNextonExceptionResumeNextonErrorReturn
The onErrorReturn method returns an Observable that mirrors the behavior of the
source Observable, unless that Observable invokes onError in which case, rather
than propagating that error to the observer, onErrorReturn will instead emit a
specified item and invoke the observer’s onCompleted method, as shown in
the following sample code:
def myObservable = Observable.create({ aSubscriber ->
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('Four');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('Three');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('Two');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('One');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onError();
});
myObservable.onErrorReturn({ return('Blastoff!'); }).subscribe(
{ println(it); }, // onNext
{ println("Error: " + it.getMessage()); }, // onError
{ println("Sequence complete"); } // onCompleted
);Four Three Two One Blastoff! Sequence complete
onErrorReturn(Func1)onErrorResumeNext
The onErrorResumeNext method returns an Observable that mirrors the behavior of
the source Observable, unless that Observable invokes onError in which case,
rather than propagating that error to the observer, onErrorResumeNext will
instead begin mirroring a second, backup Observable, as shown in the following sample code:
def myObservable = Observable.create({ aSubscriber ->
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('Three');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('Two');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('One');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onError();
});
def myFallback = Observable.create({ aSubscriber ->
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('0');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('1');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onNext('2');
if(false == aSubscriber.isUnsubscribed()) aSubscriber.onCompleted();
});
myObservable.onErrorResumeNext(myFallback).subscribe(
{ println(it); }, // onNext
{ println("Error: " + it.getMessage()); }, // onError
{ println("Sequence complete"); } // onCompleted
);Three Two One 0 1 2 Sequence complete
onErrorResumeNext(Func1)onErrorResumeNext(Observable)onExceptionResumeNext
Much like onErrorResumeNext method, this returns an Observable that mirrors the
behavior of the source Observable, unless that Observable invokes onError in
which case, if the Throwable passed to onError is an Exception, rather than
propagating that Exception to the observer, onExceptionResumeNext will instead
begin mirroring a second, backup Observable. If the Throwable is not an Exception, the
Observable returned by onExceptionResumeNext will propagate it to its
observer’s onError method and will not invoke its backup Observable.
onExceptionResumeNext(Observable)RxJava implements the Catch operator with three distinct operators:
onErrorReturnonErrorResumeNextonExceptionResumeNextonErrorReturn
The onErrorReturn method returns an Observable that mirrors the behavior of the
source Observable, unless that Observable invokes onError in which case, rather
than propagating that error to the observer, onErrorReturn will instead emit a
specified item and invoke the observer’s onCompleted method.
onErrorReturn(Func1)onErrorResumeNext
The onErrorResumeNext method returns an Observable that mirrors the behavior of
the source Observable, unless that Observable invokes onError in which case,
rather than propagating that error to the observer, onErrorResumeNext will
instead begin mirroring a second, backup Observable.
onErrorResumeNext(Func1)onErrorResumeNext(Observable)onExceptionResumeNext
Much like onErrorResumeNext method, this returns an Observable that mirrors the
behavior of the source Observable, unless that Observable invokes onError in
which case, if the Throwable passed to onError is an Exception, rather than
propagating that Exception to the observer, onExceptionResumeNext will instead
begin mirroring a second, backup Observable. If the Throwable is not an Exception, the
Observable returned by onExceptionResumeNext will propagate it to its
observer’s onError method and will not invoke its backup Observable.
onExceptionResumeNext(Observable)RxJS implements the Catch operator with two distinct operators:
catchonErrorResumeNextcatch
catch is found in the following distributions:
rx.jsrx.all.jsrx.all.compat.jsrx.compat.jsrx.lite.jsrx.lite.compat.jsonErrorResumeNext
This implementation borrows the confusing nomenclature from Rx.NET, in which
onErrorResumeNext switches to a back-up Observable both on an error and
on a normal, error-free termination of the source Observable.
onErrorResumeNext is found in the following distributions:
rx.jsrx.compat.jsRxKotlin implements the Catch operator in the same way as does RxJava. There are three distinct operators that provide this functionality:
onErrorReturnonErrorResumeNextonExceptionResumeNextonErrorReturn
The onErrorReturn method returns an Observable that mirrors the behavior of the
source Observable, unless that Observable invokes onError in which case, rather
than propagating that error to the observer, onErrorReturn will instead emit a
specified item and invoke the observer’s onCompleted method.
onErrorResumeNext
The onErrorResumeNext method returns an Observable that mirrors the behavior of
the source Observable, unless that Observable invokes onError in which case,
rather than propagating that error to the observer, onErrorResumeNext will
instead begin mirroring a second, backup Observable.
onExceptionResumeNext
Much like onErrorResumeNext method, this returns an Observable that mirrors the
behavior of the source Observable, unless that Observable invokes onError in
which case, if the Throwable passed to onError is an Exception, rather than
propagating that Exception to the observer, onExceptionResumeNext will instead
begin mirroring a second, backup Observable. If the Throwable is not an Exception, the
Observable returned by onExceptionResumeNext will propagate it to its
observer’s onError method and will not invoke its backup Observable.
Rx.NET implements the Catch operator with two distinct operators:
CatchOnErrorResumeNextCatch
The Catch operator has a variant that allows you to specify which sort of
Exception you want to catch. If you use that variant of the operator, any other Exceptions
will be passed through to the observer as if the Catch operator had not been
applied.
OnErrorResumeNext
This implementation introduces a confusing nomenclature, in which in spite of its name
OnErrorResumeNext switches to a back-up Observable both on an error and
on a normal, error-free termination of the source Observable. It is therefore more like a
concatenation operator.
RxPHP implements this operator as catchError.
Continues an observable sequence that is terminated by an exception with the next observable sequence.
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/catch/catchError.php
$obs2 = Rx\Observable::just(42);
$source = \Rx\Observable::error(new Exception("Some error"))
->catchError(function (Exception $e, \Rx\Observable $sourceObs) use ($obs2) {
return $obs2;
});
$subscription = $source->subscribe($stdoutObserver);
Next value: 42
Complete!
RxPY implements the Catch operator with two distinct operators:
catch_exceptionon_error_resume_nextcatch_exception
You may pass catch_exception a set of back-up Observables either as individual
function parameters or as a single array of Observables. If it encounters an
onError notification from the source Observable, it will subscribe to and begin
mirroring the first of these back-up Observables. If this back-up Observable itself
issues an onError notification, catch_exception will swallow it
and switch over to the next back-up Observable. If any of these Observables issues an
onCompleted notification, catch_exception will pass this along
and will stop.
on_error_resume_next
You may pass on_error_resume_next a set of back-up Observables either as
individual function parameters, as a single array of Observables, or as a factory function
that generates Observables. When the source Observable terminates, whether normally or with
an error, on_error_resume_next will subscribe to and begin mirroring the first
of these back-up Observables, and then will recursively continue this concatenation process
for each additional Observable until there are no more Observables to mirror, at which time
it will pass on the onError or onCompleted notification from the
last of these Observables.
Rx.rb implements the Catch operator with two distinct operators:
rescue_erroron_error_resume_nextrescue_error
You may pass rescue_error either an Observable or a factory action that
generates an Observable.
on_error_resume_next
In Rx.rb, on_error_resume_next inherits the misleading nomenclature from Rx.NET
in that it concatenates the second Observable sequence to the source sequence whether that
source sequence terminates normally or with an error.
Rx.rb implements the Catch operator with four distinct operators:
onErrorFlatMaponError notifications from a misbehaving Observable into the emissions from a secondary ObservableonErrorResumeNextonErrorReturnonExceptionResumeNextonErrorFlatMap
onErrorFlatMap handles a special case: a source Observable that is noncompliant
with the Observable contract in such a way that it may interleave
onError notifications with its emissions without terminating. This operator allows you to
replace those onError notifications with the emissions of an Observable of your choosing
without unsubscribing from the source, so that any future items emitted from the source will
be passed along to observers as though the sequence had not been interrupted with an
onError notification.
Because onErrorFlatMap is designed to work with pathological source Observables
that do not terminate after issuing an error, it is mostly useful in debugging/testing
scenarios.
Note that you should apply onErrorFlatMap directly to the pathological source
Observable, and not to that Observable after it has been modified by additional operators,
as such operators may effectively renormalize the source Observable by unsubscribing from it
immediately after it issues an error. Above, for example, is an illustration showing how
onErrorFlatMap will respond to two error-generating Observables that have been
merged by the Merge operator:
Note that onErrorFlatMap will not react to both errors generated by both
Observables, but only to the single error passed along by merge.
onErrorResumeNext
The onErrorResumeNext method returns an Observable that mirrors the behavior of
the source Observable, unless that Observable invokes onError in which case,
rather than propagating that error to the observer, onErrorResumeNext will
instead begin mirroring a second, backup Observable.
onErrorReturn
The onErrorReturn method returns an Observable that mirrors the behavior of the
source Observable, unless that Observable invokes onError in which case, rather
than propagating that error to the observer, onErrorReturn will instead emit a
specified item and invoke the observer’s onCompleted method.
onExceptionResumeNext
Much like onErrorResumeNext method, this returns an Observable that mirrors the
behavior of the source Observable, unless that Observable invokes onError in
which case, if the Throwable passed to onError is an Exception, rather than
propagating that Exception to the observer, onExceptionResumeNext will instead
begin mirroring a second, backup Observable. If the Throwable is not an Exception, the
Observable returned by onExceptionResumeNext will propagate it to its
observer’s onError method and will not invoke its backup Observable.