android - 如何使用 RxJava Observer 而不是事件总线来处理 401、403、503,500 等 HTTP 错误

Question

我正在使用 Retrofit、OK-HTTP 和 RxJava2 来处理网络调用，我在下面创建了拦截器来处理网络错误响应对于每个网络调用，有没有更好的方法来处理这个问题？EventBus 是这样吗？

我不想在每个方法中检查这个错误异常，

//HTTP客户端

         OkHttpClient tempClient = new OkHttpClient.Builder()
                            .readTimeout(CONNECT_TIMEOUT_IN_SEC, TimeUnit.SECONDS)// connect timeout
                            .connectTimeout(CONNECT_TIMEOUT_IN_SEC, TimeUnit.SECONDS)// socket timeout
                            .followRedirects(false)
                            .cache(provideHttpCache())
                            .addNetworkInterceptor(new ResponseCodeCheckInterceptor())
                            .addNetworkInterceptor(new ResponseCacheInterceptor())
                            .addInterceptor(new AddHeaderAndCookieInterceptor())
                            .build();

HTTP 客户端拦截器

      public class ResponseCodeCheckInterceptor implements Interceptor {
            private static final String TAG = "RespCacheInterceptor";

            @Override
            public Response intercept(Chain chain) throws IOException {
                Response response = chain.proceed(chain.request());
                Request originalRequest = chain.request();
                if (response.code() == HttpStatus.UNAUTHORIZED.value()) {
                      throw new UnAuthorizedException();
                }else if (response.code() == HttpStatus.INTERNAL_SERVER_ERROR.value()) {
                    throw new APIException(response.code(), "Server Internal Error");
                } else if (response.code() == HttpStatus.SERVICE_UNAVAILABLE.value()) {
                  throw new ServiceUnavailableException();
                } else {
                   throw new APIException(code,response.body().toString());
                }
                return response;
            }
        }

API 类

         @GET("customer/account/")
            Single<Customer> getCustomer();
                  ......

存储库类

        @Override
        public Single<Customer> getCustomer() {
            return this.mCustomerRemoteDataStore.getCustomer()
                    .doOnSuccess(new Consumer<Customer>() {
                        @Override
                        public void accept(Customer customer) throws Exception {
                            if (customer != null) {
                                mCustomerLocalDataStore.saveCustomer(customer);
                            }
                        }
                    }).doOnError(new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {

                        }
                    });
        }

Presenter 类

    @Override
    public void getCustomerFullDetails() {
        checkViewAttached();
        getView().showLoading();
        addSubscription(customerRepository.getCustomerFullDetails(true)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableSingleObserver<CustomerDetails>() {
                    @Override
                    public void onSuccess(@io.reactivex.annotations.NonNull CustomerDetails customerDetails) {
                        getView().onCustomerDetailsSuccess();
                    }

                    @Override
                    public void onError(@io.reactivex.annotations.NonNull Throwable throwable) {
                        Log.d(TAG, "error: " + throwable.getLocalizedMessage());
                        if (throwable instanceof UnAuthorizedException) {
                            getView().showLoginPage();
                         else if (throwable instanceof ServiceUnavailableException) {
                            getView().showServiceUnAvaiableMsg();
                        }...
                    }
                })
        );
    }

更新代码============

    public class CheckConnectivityInterceptor implements Interceptor {

        private static final String TAG = CheckConnectivityInterceptor.class.getSimpleName() ;
        private boolean isNetworkActive;

        private RxEventBus eventBus;

        private Context mContext;

        public CheckConnectivityInterceptor(RxEventBus eventBus, Context mContext) {
            this.mContext = mContext;
            this.eventBus = eventBus;
        }

        @Override
        public Response intercept(Interceptor.Chain chain) throws IOException {
            Request originalRequest = chain.request();
            String requestPath = originalRequest.url().url().getPath();
            if (!NetworkUtil.isConnected(this.mContext)) {
                eventBus.send(new ErrorState(ErrorType.NO_INTERNET_CONNECTION,
                        this.mContext.getString(R.string.no_network_connection), requestPath));
                        //Added this exception so it's not trying to execute the chain
                throw new NoConnectivityException();
            } else {
                Response originalResponse = null;
                try {
                    originalResponse = chain.proceed(chain.request());
                } catch (Exception ex) {
                    eventBus.send(new ErrorState(ErrorType.SERVICE_ERROR, this.mContext.getString(R.string.connection_failed), requestPath));
                    Log.e(TAG, "check connectivity intercept: ",ex );
                    throw new IOException("IO Exception occurred");
                }
                return originalResponse;
            }
        }
    }
   ==================== 
    public class HTTPResponseCodeCheckInterceptor implements Interceptor {

    private RxEventBus eventBus;

     public HTTPResponseCodeCheckInterceptor(RxEventBus eventBus) {
            this.eventBus = eventBus;
        }


     @Override
        public Response intercept(Chain chain) throws IOException {

        if (!responseSuccess) {
                    if (code == HttpStatus.MOVED_TEMPORARILY.value()) {
                        eventBus.send(new ErrorState(ErrorType.STEP_UP_AUTHENTICATION,requestPath,rSecureCode));
                    } else if (code == HttpStatus.INTERNAL_SERVER_ERROR.value()) { // Error code 500
                            eventBus.send(new ErrorState(ErrorType.SERVICE_ERROR, getAPIError(responseStringOrig),requestPath));
                    } else if (code == HttpStatus.SERVICE_UNAVAILABLE.value()) {
                         eventBus.send(new ErrorState(ErrorType.SERVICE_UNAVAILABLE, getOutageMessage(responseStringOrig),requestPath));
                    } else {
                        eventBus.send(new ErrorState(ErrorType.SERVICE_ERROR,new APIErrorResponse(500, "Internal Server Error"),requestPath));
                    }
                }
        }
    }
===================    
    public class RxEventBus {

        private PublishSubject<ErrorState> bus = PublishSubject.create();

        private RxEventBus() {
        }

        private static class SingletonHolder {
            private static final RxEventBus INSTANCE = new RxEventBus();
        }

        public static RxEventBus getBus() {
            return RxEventBus.SingletonHolder.INSTANCE;
        }


        public void send(ErrorState o) {
            bus.onNext(o);
        }

        public Observable<ErrorState> toObserverable() {
            return bus;
        }

        public boolean hasObservers() {
           return bus.hasObservers();
        }

        public  static void register(Object subscriber) {
            //bus.register(subscriber);
        }

        public  static void unregister(Object subscriber) {
          //  bus.unregister(subscriber);
        }
    }

 =====================   
    public class BasePresenter<V extends MVPView> implements MVPPresenter<V> {

      private final CompositeDisposable mCompositeDisposable;

       @Override
        public void subscribe() {
            initRxBus();
        }

        @Override
        public void unsubscribe() {
            RxUtil.unsubscribe(mCompositeDisposable);
        }


     public void addSubscription(Disposable disposable){
            if(mCompositeDisposable != null){
                mCompositeDisposable.add(disposable);
                Log.d(TAG, "addSubscription: "+mCompositeDisposable.size());
            }
        }

    private void initRxBus() {
            addSubscription(EPGApplication.getAppInstance().eventBus()
                    .toObserverable()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<ErrorState>() {
                        @Override
                        public void accept(ErrorState errorState) throws Exception {
                            if (mMvpView != null) {
                                mMvpView.hideLoadingIndicator();
                                if (ErrorType.STEP_UP_AUTHENTICATION == errorState.getType()) {
                                    mMvpView.showStepUpAuthentication(errorState.getSecureRequestCode());
                                } else if (ErrorType.SERVICE_ERROR == errorState.getType()) {
                                    mMvpView.showServiceError(((APIErrorResponse) errorState.getErrorData()).getErrorMessage());
                                } else if (ErrorType.SERVICE_UNAVAILABLE == errorState.getType()) {
                                    mMvpView.showServiceUnavailable(((OutageBody) errorState.getErrorData()));
                                } else if (ErrorType.UNAUTHORIZED == errorState.getType()) {
                                    mMvpView.sessionTokenExpiredRequestLogin();
                                } else if (ErrorType.GEO_BLOCK == errorState.getType()) {
                                    mMvpView.showGeoBlockErrorMessage();
                                } else if (ErrorType.SESSION_EXPIRED == errorState.getType()) {
                                    mMvpView.sessionTokenExpiredRequestLogin();
                                }else if (ErrorType.NO_INTERNET_CONNECTION == errorState.getType()) {
                                    mMvpView.showNoNetworkConnectivityMessage();
                                    mMvpView.showServiceError(resourceProvider.getString(R.string.no_network_connection));
                                }
                            }
                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {
                            Log.e(TAG, "base excpetion: ", throwable);
                        }
                    }));
        }
    }
    }
  ==================  
    public class ProfilePresenter<V extends ProfileContract.View> extends BasePresenter<V>
            implements ProfileContract.Presenter<V> {



         public ProfilePresenter(ProfileContract.View view, CustomerRepository repository) {
                super();
                this.repository = repository;
            }


             private void updateCustomerAccountDetails(JSONObject payload) {
                    getMvpView().showLoadingIndicator();
                    addSubscription(repository.updateCustomerDetails(sharedPreferencesRepository.isStepUpAuthRequired(), AppConfig.CUSTOMER_ACCOUNT_HOLDER_UPDATE
                            , payload)
                            .compose(RxUtil.applySingleSchedulers())
                            .subscribeWith(new DisposableSingleObserver<BaseServerResponse>() {
                                @Override
                                public void onSuccess(BaseServerResponse response) {
                                    if (!isViewAttached()) {
                                        return;
                                    }
                                    getMvpView().hideLoadingIndicator();
                                    getMvpView().onSuccessProfileInfoUpdate();
                                }


                                @Override
                                public void onError(Throwable throwable) {

                                    if (!isViewAttached()) {
                                        return;
                                    }

                                    if (throwable instanceof NoConnectivityException) {
                                        getMvpView().showNoNetworkConnectivityMessage();
                                    }
                                    getMvpView().hideLoadingIndicator();
                                }
                            }));
                }
      }

Answer 1

    Observable.just(new Object())
            .subscribeOn(Schedulers.io())
            .subscribe(new DefaultObserver<Object>() {
                @Override
                public void onNext(Object o) {
                }

                @Override
                public void onError(Throwable e) {

                    if(e instanceof HttpException) {
                        HttpException httpException = (HttpException) e;

                        if(httpException.code() == 400)
                            Log.d(TAG, "onError: BAD REQUEST");
                        else if(httpException.code() == 401)
                            Log.d(TAG, "onError: NOT AUTHORIZED");
                        else if(httpException.code() == 403)
                            Log.d(TAG, "onError: FORBIDDEN");
                        else if(httpException.code() == 404)
                            Log.d(TAG, "onError: NOT FOUND");
                        else if(httpException.code() == 500)
                            Log.d(TAG, "onError: INTERNAL SERVER ERROR");
                        else if(httpException.code() == 502)
                            Log.d(TAG, "onError: BAD GATEWAY");

                    }
                }

                @Override
                public void onComplete() {

                }
            });

请注意，所有代码为 2xx 的响应都将在 onNext 中使用，代码以 4xx 和 5xx 开头 将在 onError 中被消耗。

仅供引用。 ResponseCodeCheckInterceptor 在这种情况下不是必需的。只需尝试不使用自定义拦截器，这应该可以解决问题。

更新

自定义观察者

public abstract class CustomObserver extends DefaultObserver implements Observer{

@Override
public void onNext(Object o) {

}

@Override
public void onError(Throwable e) {
    if(e instanceof HttpException) {
        HttpException httpException = (HttpException) e;

        if(httpException.code() == 400)
            onBadRequest(e);
        else if(httpException.code() == 401)
            onNotAuthorized(e);
        else if(httpException.code() == 502)
            onBadGateway(e);

    }
}

public abstract void onNotAuthorized(Throwable e);

public abstract void onBadGateway(Throwable e);

public abstract void onBadRequest(Throwable e);

@Override
public void onComplete() {

}

}

实现

Observable.just(new Object())
            .subscribeOn(Schedulers.io())
            .subscribe(new CustomObserver() {
                @Override
                public void onNext(Object o) {
                    super.onNext(o);
                }

                @Override
                public void onError(Throwable e) {
                    super.onError(e);
                }

                @Override
                public void onNotAuthorized(Throwable e) {

                }

                @Override
                public void onBadGateway(Throwable e) {

                }

                @Override
                public void onBadRequest(Throwable e) {

                }

                @Override
                public void onComplete() {
                    super.onComplete();
                }
            });