1. Introduction
In this article, we’ll take a look at how to handle exceptions and errors using RxJava.
First, keep in mind that the Observable typically does not throw exceptions. Instead, by default, Observable invokes its Observer’s onError() method, notifying the observer that an unrecoverable error just occurred, and then quits without invoking any more of its Observer’s methods.
The error handling operators we are about to introduce change the default behavior by resuming or retrying the Observable sequence.
2. Maven Dependencies
First, let’s add the RxJava in the pom.xml:
<dependency>
<groupId>io.reactivex.rxjava2</groupId>
<artifactId>rxjava</artifactId>
<version>2.1.3</version>
</dependency>
The latest version of the artifact can be found here.
3. Error Handling
When an error occurs, we usually need to handle it in some way. For example, alter related external states, resuming the sequence with default results, or simply leave it be so that the error could propagate.
3.1. Action on Error
With doOnError, we can invoke whatever action that is needed when there is an error:
@Test
public void whenChangeStateOnError_thenErrorThrown() {
TestObserver testObserver = new TestObserver();
AtomicBoolean state = new AtomicBoolean(false);
Observable
.error(UNKNOWN_ERROR)
.doOnError(throwable -> state.set(true))
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
testObserver.assertNoValues();
assertTrue("state should be changed", state.get());
}
In case of an exception being thrown while performing the action, RxJava wraps the exception in a CompositeException:
@Test
public void whenExceptionOccurOnError_thenCompositeExceptionThrown() {
TestObserver testObserver = new TestObserver();
Observable
.error(UNKNOWN_ERROR)
.doOnError(throwable -> {
throw new RuntimeException("unexcepted");
})
.subscribe(testObserver);
testObserver.assertError(CompositeException.class);
testObserver.assertNotComplete();
testObserver.assertNoValues();
}
3.2. Resume With Default Items
Though we can invoke actions with doOnError, but the error still breaks the standard sequence flow. Sometimes we want to resume the sequence with a default option, that’s what onErrorReturnItem does:
@Test
public void whenHandleOnErrorResumeItem_thenResumed(){
TestObserver testObserver = new TestObserver();
Observable
.error(UNKNOWN_ERROR)
.onErrorReturnItem("singleValue")
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(1);
testObserver.assertValue("singleValue");
}
If a dynamic default item supplier is preferred, we can use the onErrorReturn:
@Test
public void whenHandleOnErrorReturn_thenResumed() {
TestObserver testObserver = new TestObserver();
Observable
.error(UNKNOWN_ERROR)
.onErrorReturn(Throwable::getMessage)
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(1);
testObserver.assertValue("unknown error");
}
3.3. Resume with Another Sequence
Instead of falling back to a single item, we may supply fallback data sequence using onErrorResumeNext when encountering errors. This would help prevent error propagation:
@Test
public void whenHandleOnErrorResume_thenResumed() {
TestObserver testObserver = new TestObserver();
Observable
.error(UNKNOWN_ERROR)
.onErrorResumeNext(Observable.just("one", "two"))
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(2);
testObserver.assertValues("one", "two");
}
If the fallback sequence differs according to the specific exception types, or the sequence needs to be generated by a function, we can pass the function to the onErrorResumeNext:
@Test
public void whenHandleOnErrorResumeFunc_thenResumed() {
TestObserver testObserver = new TestObserver();
Observable
.error(UNKNOWN_ERROR)
.onErrorResumeNext(throwable -> Observable
.just(throwable.getMessage(), "nextValue"))
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(2);
testObserver.assertValues("unknown error", "nextValue");
}
3.4. Handle Exception Only
RxJava also provides a fallback method that allows continuing the sequence with a provided Observable when an exception (but no error) is raised:
@Test
public void whenHandleOnException_thenResumed() {
TestObserver testObserver = new TestObserver();
Observable
.error(UNKNOWN_EXCEPTION)
.onExceptionResumeNext(Observable.just("exceptionResumed"))
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(1);
testObserver.assertValue("exceptionResumed");
}
@Test
public void whenHandleOnException_thenNotResumed() {
TestObserver testObserver = new TestObserver();
Observable
.error(UNKNOWN_ERROR)
.onExceptionResumeNext(Observable.just("exceptionResumed"))
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
}
As the code above shows, when an error does occur, the onExceptionResumeNext won’t kick in to resume the sequence.
4. Retry on Error
The normal sequence may be broken by a temporary system failure or backend error. In these situations, we want to retry and wait until the sequence is fixed.
Luckily, RxJava gives us options to perform exactly that.
4.1. Retry
By using retry, the Observable will be re-subscribed infinite times until when there’s no error. But most of the time we would prefer a fixed amount of retries:
@Test
public void whenRetryOnError_thenRetryConfirmed() {
TestObserver testObserver = new TestObserver();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.error(() -> {
atomicCounter.incrementAndGet();
return UNKNOWN_ERROR;
})
.retry(1)
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
testObserver.assertNoValues();
assertTrue("should try twice", atomicCounter.get() == 2);
}
4.2. Retry on Condition
Conditional retry is also feasible in RxJava, using retry with predicates or using retryUntil:
@Test
public void whenRetryConditionallyOnError_thenRetryConfirmed() {
TestObserver testObserver = new TestObserver();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.error(() -> {
atomicCounter.incrementAndGet();
return UNKNOWN_ERROR;
})
.retry((integer, throwable) -> integer < 4)
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
testObserver.assertNoValues();
assertTrue("should call 4 times", atomicCounter.get() == 4);
}
@Test
public void whenRetryUntilOnError_thenRetryConfirmed() {
TestObserver testObserver = new TestObserver();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.error(UNKNOWN_ERROR)
.retryUntil(() -> atomicCounter.incrementAndGet() > 3)
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
testObserver.assertNoValues();
assertTrue("should call 4 times", atomicCounter.get() == 4);
}
4.3. RetryWhen
Beyond these basics options, there’s also an interesting retry method: retryWhen.
This returns an Observable, say “NewO”, that emits the same values as the source ObservableSource, say “OldO”, but if the returned Observable “NewO” calls onComplete or onError, the subscriber’s onComplete or onError will be invoked.
And if “NewO” emits any item, a re-subscription to the source ObservableSource “OldO” will be triggered.
The tests below shows how this works:
@Test
public void whenRetryWhenOnError_thenRetryConfirmed() {
TestObserver testObserver = new TestObserver();
Exception noretryException = new Exception("don't retry");
Observable
.error(UNKNOWN_ERROR)
.retryWhen(throwableObservable -> Observable.error(noretryException))
.subscribe(testObserver);
testObserver.assertError(noretryException);
testObserver.assertNotComplete();
testObserver.assertNoValues();
}
@Test
public void whenRetryWhenOnError_thenCompleted() {
TestObserver testObserver = new TestObserver();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.error(() -> {
atomicCounter.incrementAndGet();
return UNKNOWN_ERROR;
})
.retryWhen(throwableObservable -> Observable.empty())
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertNoValues();
assertTrue("should not retry", atomicCounter.get()==0);
}
@Test
public void whenRetryWhenOnError_thenResubscribed() {
TestObserver testObserver = new TestObserver();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.error(() -> {
atomicCounter.incrementAndGet();
return UNKNOWN_ERROR;
})
.retryWhen(throwableObservable -> Observable.just("anything"))
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertNoValues();
assertTrue("should retry once", atomicCounter.get()==1);
}
A typical usage of retryWhen is limited retries with variable delays:
@Test
public void whenRetryWhenForMultipleTimesOnError_thenResumed() {
TestObserver testObserver = new TestObserver();
long before = System.currentTimeMillis();
Observable
.error(UNKNOWN_ERROR)
.retryWhen(throwableObservable -> throwableObservable
.zipWith(Observable.range(1, 3), (throwable, integer) -> integer)
.flatMap(integer -> Observable.timer(integer, TimeUnit.SECONDS)))
.blockingSubscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertNoValues();
long secondsElapsed = (System.currentTimeMillis() - before)/1000;
assertTrue("6 seconds should elapse",secondsElapsed == 6 );
}
Notice how this logic retries three times and incrementally delays each retry.
5. Summary
In this article, we introduced a number of ways of handling errors and exceptions in RxJava.
There are also several RxJava-specific exceptions relating to error handling – have a look at the official wiki for more details.
As always, the full implementation can be found over on Github.
