UndeliverableException while calling onError of Ob

问题:

I have a method which creates an emitter like below, there are a problem(maybe it is normal behavior) with calling onError in retrofit callback. I got UndeliverableException when try to call onError.

I can solve this by checking subscriber.isDiposed() by I wonder how can call onError coz i need to notify my UI level.

• Addition 1

--> RxJava2CallAdapterFactoryalready implemented   
private static Retrofit.Builder builderSwift = new Retrofit.Builder()
           .baseUrl(URL_SWIFT)
           .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
           .addConverterFactory(GsonConverterFactory.create())
           .addConverterFactory(new ToStringConverterFactory());

--> When i added below code to application class app won't crash
--> but i get java.lang.exception instead of my custom exception  

RxJavaPlugins.setErrorHandler(Functions<Throwable>emptyConsumer());

@Override
    public void onFileUploadError(Throwable e) {
        Log.d(TAG, "onFileUploadError: " + e.getMessage());
    }

public Observable<UploadResponseBean> upload(final UploadRequestBean uploadRequestBean, final File file) {
    return Observable.create(new ObservableOnSubscribe<UploadResponseBean>() {
        @Override
        public void subscribe(@NonNull final ObservableEmitter<UploadResponseBean> subscriber) throws Exception {

                 // ---> There are no problem with subscriber while calling onError        

                // ---> Retrofit2 service request 
                ftsService.upload(token, uploadRequestBean, body).enqueue(new Callback<UploadResponseBean>() {
                    @Override
                    public void onResponse(Call<UploadResponseBean> call, Response<UploadResponseBean> response) {

                        if (response.code() == 200){
                            // --->  calling onNext works properly
                            subscriber.onNext(new UploadResponseBean(response.body().getUrl()));
                        }
                        else{
                            // --->  calling onError throws UndeliverableException
                            subscriber.onError(new NetworkConnectionException(response.message()));                
                        }
                    }

                    @Override
                    public void onFailure(Call call, Throwable t) {
                        subscriber.onError(new NetworkConnectionException(t.getMessage()));
                    }
                });
        }
    });
}

回答1:

Since version 2.1.1 tryOnError is available:

The emitter API (such as FlowableEmitter, SingleEmitter, etc.) now features a new method, tryOnError that tries to emit the Throwable if the sequence is not cancelled/disposed. Unlike the regular onError, if the downstream is no longer willing to accept events, the method returns false and doesn't signal an UndeliverableException.

https://github.com/ReactiveX/RxJava/blob/2.x/CHANGES.md

回答2:

The problem is like you say you need to check if Subscriber is already disposed, that's because RxJava2 is more strict regarding errors that been thrown after Subscriber already disposed.
RxJava2 deliver this kind of error to RxJavaPlugins.onError that by default print to stack trace and calls to thread uncaught exception handler. you can read full explanation here.

Now what's happens here, is that you probably unsubscribed (dispose) from this Observable before query was done and error delivered and as such - you get the UndeliverableException.

I wonder how can call onError coz i need to notify my UI level.

as this is happened after your UI been unsubscribed the UI shouldn't care. in normal flow this error should delivered properly.

Some general points regarding your implementation:

• the same issue will happen at the onError in case you've been unsubscribed before.
• there is no cancellation logic here (that's what causing this problem) so request continue even if Subscriber unsubscribed.
• even if you'll implement this logic (using ObservableEmitter.setCancellable() / setDisposable()) you will still encounter this problem in case you will unsubscribe before request is done - this will cause cancellation and your onFailure logic will call onError() and the same issue will happen.
• as you performing an async call via Retrofit the specified subscription Scheduler will not make the actual request happen on the Scheduler thread but just the subscription. you can use Observable.fromCallable and Retrofit blocking call execute to gain more control over the actual thread call is happened.

to sum it up - guarding calls to onError() with ObservableEmitter.isDiposed() is a good practice in this case.
But I think the best practice is to use Retrofit RxJava call adapter, so you'll get wrapped Observable that doing the Retrofit call and already have all this considerations.

回答3:

I found out that this issue was caused by using incorrect context when retrieving view model in Fragment:

ViewModelProviders.of(requireActivity(), myViewModelFactory).get(MyViewModel.class);

Because of this, the view model lived in context of activity instead of fragment. Changing it to following code fixed the problem.

ViewModelProviders.of(this, myViewModelFactory).get(MyViewModel.class);