How to retry Retrofit call on HTTP errors (401) when using RxJava?

Question

My current Android Application is employing Retrofit(2.4.0) and RxJava(2.1.16) to execute my Web Service calls.

Im using Google SignIn for my User Authentication.

I want my Retrofit calls to detect HTTP 401 (UNAUTHORIZED) and attempt to Silently Login with Google Signin then retry the Retrofit call.

My retrofit calls resemble this

@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));

From googling I can see that retry/retryWhen will only be triggered when an error occurs in my RxJava chain, however HTTP 401 errors are not going to raise this condition.

As a newbie to RxJava how can I detect my HTTP 401 code and..

a). Execute Google SignIn Silent login

b). Silent login completes OK, retry my API call?

UPDATE

Ive got closer with the following code

@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));

I do not like the Flowable.just("DELAY").delay() call and also even though I am now catching the exception and silently login in OK I get this exception

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

How can I get the retrywhen to wait for the silentLogin to complete?

and

Whats causing the NoSuchElementException?

Answer 1

As far as I remember if you have error code > 300 then onError() will be called with Throwable which can ba cast to HttpException from where you can get error code returned by server so then you can call other function to make some "silent call"

Answer 2

When you initialize client:

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

Error handler:

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 exception class:

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)
    }
}
}

And when calling request

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() is get from fragment. you can use something else

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)

I know, quite enough of code. But this solution is really helpful for me in different projects

Answer 3

To solve 401 Unauthorized Error try to implement AuthInterceptor to your OkHttpClient.

BasicAuthInterceptor interceptorAuth = new BasicAuthInterceptor(yourToken);

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

builder.client(client);

If your authToken is expired or bad try to gain new.

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;
    }

}