UndeliverableException with Observables.combineLatest()

[j796160836]

Hi there, I'm newbie for RxKotlin, trying write some code at Android. But I found a (maybe?) a bug and got crazy on this issues. There is the simplified example.

I wrote a sample module with a error method with error.
(think it as network APIs but no internet)

class MyModule {
    companion object {
        fun asyncWithError(): Observable<String> {
            return Observable.create { _ ->
                Thread.sleep(3000)
                throw Throwable("My error")
            }
        }
    }
}

And I do a Observables.combineLatest() to combine different results. For simplify example, I using combine same things twice. (think it as combining different network APIs)

Observables.combineLatest(MyModule.asyncWithError(), MyModule.asyncWithError())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeBy(onNext = { (a, b) ->
            println(a + b)
        }, onError = { error ->
            println(error.message)
        })

In this scenario, two method both will cause error.

So I will get this UndeliverableException.

io.reactivex.exceptions.UndeliverableException: java.lang.Throwable: My error
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:349)
    at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:74)
    at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:43)
    at io.reactivex.Observable.subscribe(Observable.java:10955)
    at io.reactivex.internal.operators.observable.ObservableCombineLatest$LatestCoordinator.subscribe(ObservableCombineLatest.java:118)
    at io.reactivex.internal.operators.observable.ObservableCombineLatest.subscribeActual(ObservableCombineLatest.java:72)
    at io.reactivex.Observable.subscribe(Observable.java:10955)
    at io.reactivex.internal.operators.observable.ObservableObserveOn.subscribeActual(ObservableObserveOn.java:45)
    at io.reactivex.Observable.subscribe(Observable.java:10955)
    at io.reactivex.Observable.subscribe(Observable.java:10941)
    at io.reactivex.Observable.subscribe(Observable.java:10901)
    at io.reactivex.rxkotlin.SubscribersKt.subscribeBy(subscribers.kt:19)
    at io.reactivex.rxkotlin.SubscribersKt.subscribeBy$default(subscribers.kt:18)
    at com.johnny.rxplayground.MainActivity$onCreate$1.onClick(MainActivity.kt:31)
    at android.view.View.performClick(View.java:6294)
    at android.view.View$PerformClick.run(View.java:24770)
    at android.os.Handler.handleCallback(Handler.java:790)
    at android.os.Handler.dispatchMessage(Handler.java:99)
    at android.os.Looper.loop(Looper.java:164)
    at android.app.ActivityThread.main(ActivityThread.java:6494)
    at java.lang.reflect.Method.invoke(Native Method)
    at com.android.internal.os.RuntimeInit$MethodAndArgsCaller.run(RuntimeInit.java:438)
    at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:807)
Caused by: java.lang.Throwable: My error
    at com.johnny.rxplayground.MyModule$Companion$asyncWithError$1.subscribe(MyModule.kt:16)
    at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
    ... 20 more

I supposed that will catch at the onError = { ... } method, but actually not. How should I do for this error handling? I use these modules for the sample.

compile 'io.reactivex.rxjava2:rxkotlin:2.2.0'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

[j796160836]

I refer these https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling https://github.com/ReactiveX/RxJava/issues/5242 https://github.com/ReactiveX/RxJava/issues/5099

so I wrote this and get app stable

RxJavaPlugins.setErrorHandler { e ->
    when (e) {
        is UndeliverableException -> {
            // sallow that
        }
        else -> {
            throw e
        }
    }
}

Am I doing right?

[TarasMazepa]

You should probably update RxJava dependency to the latest version (2.1.9). RxKotlin is only set of handy extension functions over RxJava and RxJava is providing you with actual functionality. Next time you will have any issues it is better to recreate your code with pure java and post ticket to RxJava github page - where more people will see your post and you will get better and quicker support :)

[j796160836]

okay, thanks Taras. I will try to translate to pure Java code and ask again.