RxJava + Retrofit -> BaseObservable 用于 API 调用以进行集中响应处理

我是 RxJava 新手，所以如果这听起来太新手，请原谅我:-)。

截至目前，我有一个实现 Retofit 回调的抽象 CallbackClass。在那里，我捕获了回调的“onResponse”和“onError”方法，并在最终转发到自定义实现的方法之前处理各种错误类型。 我还使用这个集中类来进行请求/响应应用程序日志记录和其他内容。

例如：对于来自我的服务器的特定错误代码，我在响应正文中收到一个新的身份验证令牌，刷新该令牌，然后将调用克隆入队列。 当然，我的服务器的响应还有其他几种全局行为。

当前解决方案（无 Rx）：

    public abstract void onResponse(Call<T> call, Response<T> response, boolean isSuccess);

    public abstract void onFailure(Call<T> call, Response<T> response, Throwable t, boolean isTimeout);

    @Override
    public void onResponse(Call<T> call, Response<T> response) {
        if (_isCanceled) return;

        if (response != null && !response.isSuccessful()) {
            if (response.code() == "SomeCode" && retryCount < RETRY_LIMIT) {
                TokenResponseModel newToken = null;
                try {
                    newToken = new Gson().fromJson(new String(response.errorBody().bytes(), "UTF-8"), TokenResponseModel.class);
                } catch (Exception e) {
                    e.printStackTrace();
                }

                    SomeClass.token = newToken.token;
                    retryCount++;
                    call.clone().enqueue(this);
                    return;
                }
            }
        } else {
            onResponse(call, response, true);
            removeFinishedRequest();
            return;
        }

        onFailure(call, response, null, false);
        removeFinishedRequest();
    }

    @Override
    public void onFailure(Call<T> call, Throwable t) {
        if (_isCanceled) return;

        if (t instanceof UnknownHostException)
            if (eventBus != null)
                eventBus.post(new NoConnectionErrorEvent());

        onFailure(call, null, t, false);
        removeFinishedRequest();
    }

我的问题是：在最终链接（或重试）回订阅者方法之前，有什么方法可以实现这种集中式响应处理行为？

我发现这两个链接都有一个很好的起点，但没有具体的解决方案。任何帮助将不胜感激。

RxJava 中自定义 API 异常后强制请求重试 https://stackoverflow.com/questions/26919190/forcing-request-retry-after-custom-api-exceptions-in-rxjava?rq=1

Retrofit 2 和 RxJava 错误处理运算符 https://stackoverflow.com/questions/33740066/retrofit-2-and-rxjava-error-handling-operators

您提供的两个链接是一个非常好的起点，我用它来构建解决方案以应对意外情况

• 有时会由于暂时缺少网络连接或切换到低吞吐量网络标准（例如 EDGE）而发生网络错误，这会导致 SocketTimeoutException
• 服务器错误 -> 有时由于服务器过载而发生

我已经覆盖了CallAdapter.Factory处理错误并对错误做出适当反应。

1. Import RetryWithDelayIf来自solution https://stackoverflow.com/a/33768184/6507689你发现了

2. 覆盖CallAdapter.Factory处理错误：

   public class RxCallAdapterFactoryWithErrorHandling extends CallAdapter.Factory {
       private final RxJavaCallAdapterFactory original;
   
       public RxCallAdapterFactoryWithErrorHandling() {
           original = RxJavaCallAdapterFactory.create();
       }
   
       @Override
       public CallAdapter<?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
           return new RxCallAdapterWrapper(retrofit, original.get(returnType, annotations, retrofit));
       }
   
       public class RxCallAdapterWrapper implements CallAdapter<Observable<?>> {
           private final Retrofit retrofit;
           private final CallAdapter<?> wrapped;
   
           public RxCallAdapterWrapper(Retrofit retrofit, CallAdapter<?> wrapped) {
               this.retrofit = retrofit;
               this.wrapped = wrapped;
           }
   
           @Override
           public Type responseType() {
               return wrapped.responseType();
           }
   
           @SuppressWarnings("unchecked")
           @Override
           public <R> Observable<?> adapt(final Call<R> call) {
               return ((Observable) wrapped.adapt(call)).onErrorResumeNext(new Func1<Throwable, Observable>() {
                   @Override
                   public Observable call(Throwable throwable) {
                       Throwable returnThrowable = throwable;
                       if (throwable instanceof HttpException) {
                           HttpException httpException = (HttpException) throwable;
                           returnThrowable = httpException;
                           int responseCode = httpException.response().code();
                           if (NetworkUtils.isClientError(responseCode)) {
                               returnThrowable = new HttpClientException(throwable);
                           }
                           if (NetworkUtils.isServerError(responseCode)) {
                               returnThrowable = new HttpServerException(throwable);
                           }
                       }
   
                       if (throwable instanceof UnknownHostException) {
                           returnThrowable = throwable;
                       }
   
                       return Observable.error(returnThrowable);
                   }
               }).retryWhen(new RetryWithDelayIf(3, DateUtils.SECOND_IN_MILLIS, new Func1<Throwable, Boolean>() {
                   @Override
                   public Boolean call(Throwable throwable) {
                       return throwable instanceof HttpServerException
                               || throwable instanceof SocketTimeoutException
                               || throwable instanceof UnknownHostException;
                   }
               }));
           }
       }
   }
   

   HttpServerException只是一个自定义异常。

3. 使用它在Retrofit.Builder https://square.github.io/retrofit/2.x/retrofit/retrofit2/Retrofit.Builder.html

   Retrofit retrofit = new Retrofit.Builder()
           .addCallAdapterFactory(new RxCallAdapterFactoryWithErrorHandling())
           .build();
   

Extra：如果你想解析来自API的错误（不调用的错误UnknownHostException, HttpException or MalformedJsonException或等等）你需要覆盖Factory https://square.github.io/retrofit/2.x/retrofit/retrofit2/Converter.Factory.html并在构建过程中使用自定义的Retrofit实例。解析响应并检查它是否包含错误。如果是，则抛出错误，错误将在上面的方法内处理。