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

The translation of the article will be discussed UndeliverableExceptionin the RxJava2version 2.0.6and newer. If someone ran into and can not figure out, or did not hear at all about this problem - I ask under the cat. Encouraged to transfer the problem in productionafter the transition from RxJava1to 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 mistakes in onErrorwhen 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 the application will fall due to RxJava, even if correctly implemented onError.

Basic error handler in RxJava

In the role of the basic error handler in is RxJavaused RxJavaPlugins.onError. 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.6This behavior is described:

One of the goals of design 2.x is the lack of lost mistakes. Sometimes the sequence ends or is canceled before the source calls onError. In this case, the error has nowhere to go and it goes toRxJavaPlugins.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 with the version 2.0.6, it RxJavaPlugins.onErrortries to be smarter and shares the library / implementation errors and situations when the error cannot be delivered. Errors classified as “bugs” are called as they are, the rest turn into UndeliverableExceptionand after are caused. All this logic can be seen here (methods onErrorand isBug).

One of the major mistakes that newcomers face is RxJava- OnErrorNotImplementedException. This error occurs if it observablecauses an error, and the method is not implemented in the subscriber onError. This error is an example of an error for a basic error handler.RxJavais a "bug" and does not turn into UndeliverableException.

UndeliverableException

Since errors relating to the "bugs" are easy to fix - we will not dwell on them. The errors that it RxJavawraps in UndeliverableExceptionare more interesting, since it may not always be obvious why the 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.

Example with zipWith ()

The first option in which you can call UndeliverableException- the operator 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 it onErrorwill be called twice, but this contradicts the specificationReactive streams .

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

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

An easy way to get into a situation like this is to use zipnetwork calls to join (for example, two calls Retrofitthat 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 implementation onError, and the second will be delivered to the basic error handler ( RxJavaPlugins.onError).

Example with ConnectableObservable without subscribers

ConnectableObservablecan also cause UndeliverableException. It is worth recalling that ConnectableObservablecauses events, regardless of the presence of active subscribers, just call the method connect(). If in the absence of subscribers ConnectableObservablean error occurs, it will be delivered to the basic error handler.

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

someApi.retrofitCall() // Сетевой вызов с использованием Retrofit
    .publish()
    .connect()

If an someApi.retrofitCall()error occurs (for example, there is no connection to the Internet), the application will crash, as the network error will be delivered to the basic error handler RxJava.

This example seems fictional, but it is very easy to get into a situation where ConnectableObservableeverything is still connected (connected), but it has no subscribers. I encountered this when used autoConnect()when calling the API. autoConnect()does not automatically shut off Observable. I unsubscribed in the onStopmethod Activity, but the result of the network call was returned after the destruction Activityand the application fell from 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 errors in RxJavaPlugins.onError.

The solution to the example with zipWithis to take one or both sources and implement one of the methods for catching errors in them . For example, you can use the onErrorReturndefault instead of an error.

The example with the ConnectableObservablefix is ​​easier - just make sure that you disconnect Observableat the moment when the last subscriber unsubscribes. autoConnect(), for example, has an overloaded implementation that takes a function that captures the moment of the connection ( more can be found here).

Another way to solve the problem is to replace the basic error handler with your own. Method RxJavaPlugins.setErrorHandler(Consumer<Throwable>)will help you with this. If this is the right solution for you, you can intercept all the errors sent to RxJavaPlugins.onErrorand process them as you see fit. This solution can be quite complicated - remember that it RxJavaPlugins.onErrorreceives errors from all streams RxJavain the application.

If you manually create your own Observable, you can emitter.onError()call instead 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 implementedonErrorin subscribers. You should be aware of situations in which errors may be inaccessible to subscribers, and make sure that these situations are handled.