使用 RxJava 时如何重试对 HTTP 错误 (401) 的 Retrofit 调用？

Question

我当前的 Android 应用程序正在使用 Retrofit(2.4.0) 和 RxJava(2.1.16) 来执行我的 Web 服务调用。

我正在使用 Google SignIn 进行用户身份验证。

我希望我的 Retrofit 调用检测 HTTP 401（未经授权）并尝试使用 Google Signin 静默登录 然后重试 Retrofit 调用。

我的改装电话与此类似

@Headers(HEADER_ACCEPT_JSON)
@GET("resources")
Observable<Response<String>> getResources(@Header(HEADER_AUTHORIZATION) @NonNull final String authenticationToken, @QueryMap(encoded = true) @NonNull Map<String, Object> queryMap);



API_SERVICE.getResources(Login.getAuthorizationToken(), id)
                .subscribeOn(Schedulers.io())
                .subscribe(Network::manageResource, Network::handle));

通过谷歌搜索我可以看到 retry/retryWhen 只会在我的 RxJava 链中发生错误时触发， 但是 HTTP 401 错误不会引发这种情况。

作为 RxJava 的新手，我如何检测我的 HTTP 401 代码和..

一）。执行 Google SignIn 静默登录

b)。静默登录完成 OK，重试我的 API 调用？

更新

我用下面的代码更接近了

@Headers(HEADER_ACCEPT_JSON)
@GET("resources")
Single<Response<String>> getResources(@Header(HEADER_AUTHORIZATION) @NonNull final String authenticationToken, @QueryMap(encoded = true) @NonNull Map<String, Object> queryMap);



API_SERVICE.getResources(Login.getAuthorizationToken(), id)
  .subscribeOn(Schedulers.io())
  .flatMap(new Function<Response<Article>,                                           
 SingleSource<Response<Article>>>() {
                        @Override
                        public SingleSource<Response<Article>> apply(final Response<Article> response) {
                            Log.d(TAG, "apply() called with: response = [" + response + "]");
                            if (response.isSuccessful()) {
                                return Single.just(response);
                            } else {
                                return Single.error(new RuntimeException());
                            }
                        }
                    })
                    .retryWhen(errors -> errors.take(1).flatMap(new Function<Throwable, Publisher<?>>() {
                        @Override
                        public Publisher<?> apply(final Throwable throwable) {
                            Log.d(TAG, "apply() called with: throwable = [" + throwable + "]");
                            Login.loginSilently().subscribe();

                            return Flowable.just("DELAY").delay(10, TimeUnit.SECONDS);
                        }
                    }))
                        .subscribe(Network::manageResource, Network::handle));

我不喜欢 Flowable.just("DELAY").delay() 调用，即使我现在正在捕获异常并静默登录 OK，但我得到了这个异常

09-10 16:39:29.878 7651-7718/research.android E/Network: handle: 
    java.util.NoSuchElementException
        at io.reactivex.internal.operators.flowable.FlowableSingleSingle$SingleElementSubscriber.onComplete(FlowableSingleSingle.java:116)
        at io.reactivex.subscribers.SerializedSubscriber.onComplete(SerializedSubscriber.java:168)
        at io.reactivex.internal.operators.flowable.FlowableRepeatWhen$WhenReceiver.onComplete(FlowableRepeatWhen.java:119)
        at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.drainLoop(FlowableFlatMap.java:426)
        at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.drain(FlowableFlatMap.java:366)
        at io.reactivex.internal.operators.flowable.FlowableFlatMap$InnerSubscriber.onComplete(FlowableFlatMap.java:673)
        at io.reactivex.subscribers.SerializedSubscriber.onComplete(SerializedSubscriber.java:168)
        at io.reactivex.internal.operators.flowable.FlowableDelay$DelaySubscriber$OnComplete.run(FlowableDelay.java:139)
        at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)
        at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:57)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:301)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1162)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:636)
        at java.lang.Thread.run(Thread.java:764)
09-10 16:39:29.878 7651-7678/research.android  D/OkHttp: <-- HTTP FAILED: java.io.IOException: Canceled

如何在等待静默登录完成时重试？

和

是什么导致了 NoSuchElementException？

Answer 1

据我所知，如果您的错误代码 > 300，则将使用 Throwable 调用 onError()，该 Throwable 可以转换为 HttpException，您可以从中获取服务器返回的错误代码，然后您可以调用其他函数来打个“无声的电话”

Answer 2

当你初始化客户端时：

Retrofit.Builder()
    .baseUrl(baseUrl)
    .client(createClient())
    .addConverterFactory(GsonConverterFactory.create())
    .addCallAdapterFactory(ApiHandler(Schedulers.io()))
    .build()

错误处理程序：

class ApiHandler(scheduler: Scheduler) : CallAdapter.Factory() {

private val original: RxJava2CallAdapterFactory
        = RxJava2CallAdapterFactory.createWithScheduler(scheduler)

override fun get(returnType: Type, annotations: Array<Annotation>, retrofit: Retrofit): CallAdapter<*, *>?
        = original.get(returnType, annotations, retrofit)?.let { Wrapper(it) }

private class Wrapper<R>(private val wrapped: CallAdapter<R, *>) : CallAdapter<R, Any> {
    override fun adapt(call: Call<R>?): Any? {
        call ?: return null
        val result = wrapped.adapt(call)

        return when (result) {
            is Maybe<*>      -> result.onErrorResumeNext(Function { Maybe.error(wrap(it)) })
            is Single<*>     -> result.onErrorResumeNext { Single.error(wrap(it)) }
            is Completable   -> result.onErrorResumeNext { Completable.error(wrap(it)) }
            is Flowable<*>   -> result.onErrorResumeNext(Function { Flowable.error(wrap(it)) })
            is Observable<*> -> result.onErrorResumeNext(Function { Observable.error(wrap(it)) })

            else             -> result
        }
    }

    override fun responseType(): Type = wrapped.responseType()

    private fun wrap(throwable: Throwable) = when (throwable) {
        is HttpException          -> {
            val exception = ApiException.http(throwable)
            toLog("ex - ${exception.message}")
            exception
        } // We had non-200 http error
        is JsonSyntaxException    -> ApiException.parse(throwable) // We had json parsing error
        is SocketTimeoutException -> ApiException.timeout(throwable) // A network error happened
        is IOException            -> ApiException.network(throwable) // A network error happened

        else                      -> ApiException.unknown(throwable) // We don't know what happened. We need to simply convert to an unknown error
    }
}
}

API异常类：

class ApiException internal constructor(message: String,
                                    /** Response object containing status code, headers, body, etc.  */
                                    val response: ErrorResponse?,
                                    /** The event kind which triggered this error.  */
                                    @ApiError val error: Int,
                                    exception: Throwable?) : RuntimeException(message, exception) {

companion object {
    fun http(exception: HttpException): ApiException {
        val response = exception.response()
        var errorResponse: ErrorResponse? = null
        val message = if (response == null) {
            if (exception.message().isEmpty()) exception.code().toString() else exception.message()

        } else {

            // here you can check error code and throw needed exception

            val errorBody = response.errorBody()?.string().toString()
            if (errorBody.isNotEmpty()) {
                toLog("ApiException: $errorBody")
            }
            try {
                errorResponse = GsonBuilder().create().fromJson(errorBody, ErrorResponse::class.java)
                errorResponse?.toString() ?: errorBody

            } catch (e: Exception) {
                e.printStackTrace()
                response.raw().message()
            }
        }
        return ApiException(message, errorResponse, ApiError.HTTP, exception)
    }

    fun network(exception: IOException): ApiException {
        return ApiException(exception.message ?: "network", null, ApiError.NETWORK, exception)
    }

    fun parse(exception: JsonSyntaxException): ApiException {
        return ApiException(exception.message ?: "parse", null, ApiError.CONVERSION, exception)
    }

    fun unknown(exception: Throwable): ApiException {
        return ApiException(exception.message ?: "unknown", null, ApiError.UNKNOWN, exception)
    }

    fun timeout(exception: SocketTimeoutException): ApiException {
        return ApiException("Connection timed out", null, ApiError.TIMEOUT_EXCEPTION, exception)
    }
}
}

以及调用请求时

yourRequest.compose { observable ->
                observable.retryWhen { flow ->
                    flow.ofType(ApiException::class.java).flatMap {
                        when {
                            it.error == ApiError.TIMEOUT_EXCEPTION -> Flowable.empty<T>()
                            it.error == ApiError.NETWORK           -> getSnackBarFlowable().flatMap { if (it) Flowable.just(it) else Flowable.empty<T>() }

                            else                                   -> Flowable.error(it)
                        }
                    }
                }
            }.subscribe({}, {})

getSnackBarFlowable() 是从片段中获取的。你可以用别的东西

fun getSnackBarFlowable(): Flowable<Boolean> = Flowable.create({ subscriber ->
    if (view == null) {
        subscriber.onNext(false)

    } else {
        val snackBar = Snackbar.make(activity!!.currentFocus, R.string.error_connection_fail, Snackbar.LENGTH_INDEFINITE)
        snackBar.setAction("Retry") { subscriber.onNext(true) }
        snackBar.show()
    }
}, LATEST)

我知道，代码已经够多了。但是这个解决方案对我在不同的项目中真的很有帮助

Answer 3

要解决 401 Unauthorized Error 尝试将 AuthInterceptor 实现到您的 OkHttpClient。

BasicAuthInterceptor interceptorAuth = new BasicAuthInterceptor(yourToken);

OkHttpClient client = new OkHttpClient.Builder()
        .addInterceptor(interceptorAuth)
        .build();

builder.client(client);

如果您的 authToken 已过期或错误，请尝试获取新的。

public class BasicAuthInterceptor implements Interceptor {

private String yourToken;

public BasicAuthInterceptor(String token) {
    this.yourToken = token;
}

@Override
public Response intercept(Chain chain) throws IOException {
    Request request = chain.request();
    Request authenticatedRequest = request.newBuilder()
        .header("Authorization", format("token %s", yourToken)).build();

    Response response = chain.proceed(authenticatedRequest);

    boolean unauthorized = response.code() == 401;
    if (unauthorized) {

        Request modifiedRequest = request.newBuilder()
                .header("Authorization", format("token %s", getNewToken())).build();
        response = chain.proceed(modifiedRequest);
        }

    return response;
    }

}