rxjava throws OnErrorNotImplementedException

Question

Observable.create(new ObservableOnSubscribe())
    .subscribeOn(Schedulers.io())
    .compose(new ParserTransformer())
    .map()
    .subscribe()

When something wrong in my ObservableOnSubscribe, it throws an exception(java.net.UnknownHostException ...).

ObservableOnSubscribe:

public void subscribe(ObservableEmitter oe) throws Exception {
    try{
        ...
    } catch {
        if (!oe.isDisposed()) {
            if (ex instanceof IOException) {
                throw new NetException(...); //sometimes, it make the app crash!
            }
        } else{
            ...
        }
    }
    oe.onComplete();
}

here's the log :

D OkHttp: error: java.net.UnknownHostException
D OkHttp: oe.isDisposed: false
W System.err: io.reactivex.exceptions.OnErrorNotImplementedException: Unable to resolve host ******: No address associated with hostname
W System.err:   at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
W System.err:   at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
W System.err:   at io.reactivex.internal.observers.LambdaObserver.onError(LambdaObserver.java:77)
W System.err:   at io.reactivex.internal.observers.BasicFuseableObserver.onError(BasicFuseableObserver.java:100)
W System.err:   at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.checkTerminated(ObservableObserveOn.java:276)
W System.err:   at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:172)
W System.err:   at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:252)
W System.err:   at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:109)
W System.err:   at android.os.Handler.handleCallback(Handler.java:754)
W System.err:   at android.os.Handler.dispatchMessage(Handler.java:95)
W System.err:   at android.os.Looper.loop(Looper.java:163)
W System.err:   at android.app.ActivityThread.main(ActivityThread.java:6343)
W System.err:   at java.lang.reflect.Method.invoke(Native Method)
W System.err:   at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:880)
W System.err:   at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:770)

Why this happen? Why rxjava says that OnError is not implemented?

Answer 1

You should use Emitter.onError instead of throwing Exception.

Edit: Remember about javadocs: OnErrorNotImplementedException

public void subscribe(ObservableEmitter oe) throws Exception {
    try{
        ...
    } catch {
        if (!oe.isDisposed()) {
            if (ex instanceof IOException) {
                oe.onError(new NetException(...)); //pass error to emitter
            }
        } else{
            ...
        }
    }
    oe.onComplete();
}