Standard Error Handler in RxJava2 or why RxJava causes the application to crash even if onError is implemented

The translation of the article will deal with UndeliverableException in RxJava2 version 2.0.6 and newer. If someone collided and cannot understand, or did not hear at all about this problem - I ask under cat. Prompted to transfer the problem in production after the transition from RxJava1 to RxJava2 . The original was written on December 28, 2017, but it’s better to learn late than never.
We are all good developers and we catch errors in onError when we use RxJava . This means that we protected ourselves from the application crashes, right?

No, not right.

Below we look at a couple of examples in which an application will fall due to RxJava , even if onError correctly implemented.
')

Basic error handler in RxJava

In the role of the basic error handler in RxJava , RxJavaPlugins.onError used. It handles all errors that cannot be delivered to the subscriber. By default, all errors are sent exactly to it, so application crashes may occur.
2.0.6 this behavior is described:

One of the goals of 2.x design is the absence of lost bugs. Sometimes the sequence ends or is canceled before the source calls onError . In this case, the error has nowhere to go and it is sent to RxJavaPlugins.onError

If RxJava does not have a basic error handler, such errors will be hidden from us and the developers will be in the dark about potential problems in the code.

Starting from version 2.0.6 , RxJavaPlugins.onError tries to be smarter and shares the library / implementation errors and situations when the error cannot be delivered. Errors categorized as “bugs” are caused as they are, the rest are wrapped in UndeliverableException and then called. All this logic can be seen here ( onError and isBug ).

One of the major bugs that newcomers face in RxJava is OnErrorNotImplementedException . This error occurs when the observable causes an error, and the onError method is not implemented in the subscriber. This error is an example of an error that for the basic error handler RxJava is a “bug” and does not turn into a UndeliverableException .

UndeliverableException

Since the errors related to the "bugs" are easy to fix - we will not dwell on them. The errors that RxJava wraps in a UndeliverableException are more interesting, since it may not always be obvious why an error cannot be delivered before onError .

The cases in which this can occur depend on what the sources and subscribers specifically do. Examples are discussed below, but in general we can say that such an error occurs if there is no active subscriber to whom an error can be delivered.

ZipWith () example

The first option in which you can call UndeliverableException is the zipWith .

 val observable1 = Observable.error<Int>(Exception()) val observable2 = Observable.error<Int>(Exception()) val zipper = BiFunction<Int, Int, String> { one, two -> "$one - $two" } observable1.zipWith(observable2, zipper) .subscribe( { System.out.println(it) }, { it.printStackTrace() } ) 

We combine together two sources, each of which causes an error. What do we expect? We can assume that onError will be called twice, but this is contrary to the Reactive streams specification .

After a single call to a terminal event ( onError , onCompelete ), it is required that no calls be made again.

It turns out that with a single onError call onError second call is no longer possible. What happens when a second error occurs at the source? It will be delivered to RxJavaPlugins.onError .

A simple way to get into a similar situation is to use zip to join network calls (for example, two Retrofit calls that return Observable ). If an error occurs in both calls (for example, there is no internet connection), both sources will cause errors, the first of which will go to the onError implementation, and the second will be delivered to the basic error handler ( RxJavaPlugins.onError ).

Example with ConnectableObservable without subscribers

ConnectableObservable can also cause UndeliverableException . It is worth recalling that ConnectableObservable raises events regardless of the presence of active subscribers, just call the connect() method. If there are no errors in ConnectableObservable if there are no subscribers, it will be delivered to the underlying error handler.

Here is a rather innocent example that could cause such an error:

 someApi.retrofitCall() //     Retrofit .publish() .connect() 

If someApi.retrofitCall() causes an error (for example, there is no internet connection), the application will crash, since the network error will be delivered to the basic RxJava error RxJava .

This example seems fictional, but it is very easy to get into a situation where the ConnectableObservable is still connected, but it has no subscribers. I encountered this when using autoConnect() when calling the API. autoConnect() does not automatically disable Observable . I unsubscribed in the onStop method of the Activity , but the result of the network call was returned after the destruction of the Activity and the application UndeliverableException with UndeliverableException .

We process errors

So what to do with these errors?

The first step is to look at the errors that occur and try to determine what causes them. Ideally, if you manage to fix the problem at its source, to prevent the transmission of the error in RxJavaPlugins.onError .

The solution to the zipWith example is to take one or both sources and implement one of the methods for catching errors in them. For example, you can use onErrorReturn to pass a default value instead of an error.

The example with ConnectableObservable easier to fix - just make sure you disconnect the Observable at the time when the last subscriber unsubscribes. autoConnect() , for example, has an overloaded implementation that accepts a function that captures the connection point ( more can be found here ).

Another way to solve the problem is to replace the basic error handler with your own. The RxJavaPlugins.setErrorHandler(Consumer<Throwable>) method will help you with this. If this is a suitable solution for you, you can intercept all errors sent to RxJavaPlugins.onError and process them as you see fit. This solution can be quite complicated - remember that RxJavaPlugins.onError receives errors from all streams of RxJava in the application.

If you manually create your own Observable , you can call emitter.tryOnError() instead of emitter.tryOnError() . This method sends an error only if the stream (stream) is not destroyed (terminated) and has subscribers. Remember that this method is experimental.

The moral of this article is that you cannot be sure that there are no errors when working with RxJava if you simply implemented onError in the subscribers. You should be aware of situations in which errors may be inaccessible to subscribers, and make sure that these situations are handled.