在RXjava2中调用ObservableEmitter的onError时出现UndeliverableException

Question

我有一个创建像下面的发射器的方法,有一个问题(也许这是正常的行为)在改装回调中调用onError.尝试调用onError时,我得到了UndeliverableException.

我可以通过检查subscriber.isDiposed()来解决这个问题,我想知道如何调用onError coz我需要通知我的UI级别.

• 增加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()));
                    }
                });
        }
    });
}

Answer 1

由于版本2.1.1 tryOnError可用:

发射器API(例如FlowableEmitter,SingleEmitter等)现在具有一个新方法tryOnError,如果未取消/处理序列,它尝试发出Throwable.与常规onError不同,如果下游不再愿意接受事件,则该方法返回false并且不发出UndeliverableException信号.

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

Answer 2

问题就像你说你需要检查是否Subscriber已经处理掉,这是因为RxJava2对于Subscriber已经处理后抛出的错误更加严格.
RxJava2将此类错误传递给RxJavaPlugins.onError默认情况下打印到堆栈跟踪并调用线程未捕获的异常处理程序.你可以在这里阅读完整的解释.

现在发生的事情是,你可能Observable在查询完成之前取消订阅(处置)并发送错误,因此 - 你得到了UndeliverableException.

我想知道如何调用onError coz我需要通知我的UI级别.

因为这是在您的UI取消订阅后发生的,UI不应该关心.在正常流程中,此错误应正确传递.

关于您的实施的一些一般要点:

• onError如果您以前取消订阅,也会发生同样的问题.
• 这里没有取消逻辑(这就是造成这个问题的原因)所以即使Subscriber取消订阅,请求也会继续.
• 即使你将实现这个逻辑(使用ObservableEmitter.setCancellable()/ setDisposable()),你仍然会遇到这个问题,如果你在请求完成之前取消订阅 - 这将导致取消,你的 onFailure逻辑将调用onError(),同样的问题将发生.
• 当您通过Retrofit执行异步调用时,指定的订阅Scheduler不会使实际请求发生在Scheduler线程上,而只是订阅.您可以使用Observable.fromCallable和Retrofit阻塞调用execute来获得对实际线程调用的更多控制.

概括起来-要守着电话onError()与ObservableEmitter.isDiposed()在这种情况下,一个很好的做法.
但我认为最好的做法是使用Retrofit RxJava调用适配器,这样你就Observable可以进行Retrofit调用,并且已经考虑了所有这些因素.