android - retrofit with rxjava handling network exceptions globally

Question

I am trying to handle exceptions in app on global level, so that retrofit throws an error i catch it in some specific class with logic for handling those errors.

I have an interface

@POST("/token")
AuthToken refreshToken(@Field("grant_type") String grantType, @Field("refresh_token") String refreshToken);

and observables

/**
 * Refreshes auth token
 *
 * @param refreshToken
 * @return
 */
public Observable<AuthToken> refreshToken(String refreshToken) {
    return Observable.create((Subscriber<? super AuthToken> subscriber) -> {
        try {
            subscriber.onNext(apiManager.refreshToken(REFRESH_TOKEN, refreshToken));
            subscriber.onCompleted();
        } catch (Exception e) {
            subscriber.onError(e);
        }
    }).subscribeOn(Schedulers.io());
}

When i get 401 from server (invalid token or some other network related error) i want to refresh the token and repeat the rest call. Is there a way to do this with rxjava for all rest calls with some kind of observable that will catch this error globally, handle it and repeat the call that throw-ed it?

For now i am using subject to catch the error on .subscribe() like this

private static BehaviorSubject errorEvent = BehaviorSubject.create();

public static BehaviorSubject<RetrofitError> getErrorEvent() {
    return errorEvent;
}

and in some call

getCurrentUser = userApi.getCurrentUser().observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    (user) -> {
                        this.user = user;
                    },
                    errorEvent::onNext
            );

then in my main activity i subscribe to that behaviour subject and parse the error

SomeApi.getErrorEvent().subscribe(
            (e) -> {
                //parse the error
            }
    );

but i cant repeat the call for the observable that throw the error.

Answer 1

You need to use the operator onErrorResumeNext(Func1 resumeFunction), better explained in the official wiki:

The onErrorResumeNext(?) method returns an Observable that mirrors the behavior of the source Observable, unless that Observable invokes onError(?) in which case, rather than propagating that error to the Subscriber, onErrorResumeNext(?) will instead begin mirroring a second, backup Observable

In your case I would put something like this:

getCurrentUser = userApi.getCurrentUser()
.onErrorResumeNext(refreshTokenAndRetry(userApi.getCurrentUser()))
.observeOn(AndroidSchedulers.mainThread())
            .subscribe(...)

where:

    private <T> Func1<Throwable,? extends Observable<? extends T>> refreshTokenAndRetry(final Observable<T> toBeResumed) {
        return new Func1<Throwable, Observable<? extends T>>() {
            @Override
            public Observable<? extends T> call(Throwable throwable) {
                // Here check if the error thrown really is a 401
                if (isHttp401Error(throwable)) {
                    return refreshToken().flatMap(new Func1<AuthToken, Observable<? extends T>>() {
                        @Override
                        public Observable<? extends T> call(AuthToken token) {
                            return toBeResumed;
                        }
                    });
                }
                // re-throw this error because it's not recoverable from here
                return Observable.error(throwable);
            }
        };
    }

Note also that this function can be easily used in other cases, because it's not typed with the actual values emitted by the resumed Observable.