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.

Answer 2

@Override
public Observable<List<MessageEntity>> messages(String accountId, int messageType) {
    return mMessageService.getLikeMessages(messageType)
            .onErrorResumeNext(mTokenTrick.
                    refreshTokenAndRetry(mMessageService.getLikeMessages(messageType)));
}