RxJava + Retrofit -> BaseObservable for API 调用集中响应处理

Question

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

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

例如：对于来自我的服务器的特定错误代码，我在响应正文中收到一个新的 Auth 令牌，刷新令牌，然后 clone.enqueue 调用。我的服务器响应当然还有其他几种全局行为。

当前解决方案（无 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();
    }

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

我发现这 2 个链接都有很好的起点，但没有具体的解决方案。非常感谢任何帮助。

Forcing request retry after custom API exceptions in RxJava

Answer 1

你考虑过使用rxjava适配器进行改造吗？ https://mvnrepository.com/artifact/com.squareup.retrofit2/adapter-rxjava/2.1.0 在您的 gradle 文件中添加

compile 'com.squareup.retrofit2:adapter-rxjava:2.1.0'

这是改造的界面

public interface Service {
@GET("userauth/login?")
Observable<LoginResponse> getLogin(
        @Query("v") String version,
        @Query("username") String username,
        @Query("password") String password);
}

这是我的实现

Service.getLogin(
            VERSION,
            "username",
            "password")
            .subscribe(new Subscriber<LoginResponse>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(LoginResponse loginResponse) {

                }
            });

请注意，我正在使用 gson 转换器工厂来解析我的响应，因此我得到了一个返回的 pojo（普通 Ole Java 对象）。

Answer 2

你提供的两个链接是一个很好的起点，我用它来构建解决方案来应对意外

网络错误有时是由于暂时没有网络连接，或者切换到低吞吐量网络标准，如 EDGE，这会导致 SocketTimeoutException
服务器错误 -> 由于服务器过载有时会发生

我已重写 CallAdapter.Factory 来处理错误并做出适当的反应。

从您找到的导入RetryWithDelayIf

覆盖 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 只是一个自定义例外。

在Retrofit.Builder

中使用

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

Extra：如果您希望解析来自 API 的错误（不调用 UnknownHostException、HttpException 或MalformedJsonException 等）您需要覆盖 Factory 并在构建 Retrofit 实例期间使用自定义实例。解析响应并检查它是否包含错误。如果是，则抛出错误，错误将在上面的方法内部处理。

Answer 3

看看你是怎么做到的。这里api在这调用并传递请求模型和响应模型

public interface RestService {
//SEARCH_USER
@POST(SEARCH_USER_API_LINK)
Observable<SearchUserResponse> getSearchUser(@Body SearchUserRequest getSearchUserRequest);
}

这是retrofit调用，我用的是retrofit2

public RestService getRestService() {

    Retrofit retrofit = new Retrofit.Builder()
            .baseUrl(ApiConstants.BASE_URL)
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
            .client(getOkHttpClient())
            .build();

    return retrofit.create(RestService.class);
}

//get OkHttp instance
@Singleton
@Provides
public OkHttpClient getOkHttpClient() {

    HttpLoggingInterceptor httpLoggingInterceptor = new HttpLoggingInterceptor();
    httpLoggingInterceptor.setLevel(HttpLoggingInterceptor.Level.BODY);

    OkHttpClient.Builder builder = new OkHttpClient.Builder();
    builder.interceptors().add(httpLoggingInterceptor);
    builder.readTimeout(60, TimeUnit.SECONDS);
    builder.connectTimeout(60, TimeUnit.SECONDS);
    return builder.build();
}

这是 api 调用，请在您的 activity 中调用。

@Inject
Scheduler mMainThread;
@Inject
Scheduler mNewThread;

 //getSearchUser api method
public void getSearchUser(String user_id, String username) {

    SearchUserRequest searchUserRequest = new SearchUserRequest(user_id, username);

    mObjectRestService.getSearchUser(searchUserRequest).
            subscribeOn(mNewThread).
            observeOn(mMainThread).
            subscribe(searchUserResponse -> {
                Timber.e("searchUserResponse :" + searchUserResponse.getResponse().getResult());
                if (isViewAttached()) {
                    getMvpView().hideProgress();
                    if (searchUserResponse.getResponse().getResult() == ApiConstants.STATUS_SUCCESS) {

                    } else {

                    }
                }
            }, throwable -> {
                if (isViewAttached()) {

                }
            });
}

希望对您有所帮助。

RxJava + Retrofit -> BaseObservable for API 调用集中响应处理

RxJava + Retrofit -> BaseObservable for API calls for centralized response handling

android

rx-java

rx-android

retrofit2