android – retrofit with rxjava handling network exceptions globally-ThrowExceptions

Exception or error:

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

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));
        } catch (Exception e) {

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())
                    (user) -> {
                        this.user = user;

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

            (e) -> {
                //parse the error

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

How to solve:

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


    private <T> Func1<Throwable,? extends Observable<? extends T>> refreshTokenAndRetry(final Observable<T> toBeResumed) {
        return new Func1<Throwable, Observable<? extends T>>() {
            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>>() {
                        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.


public Observable<List<MessageEntity>> messages(String accountId, int messageType) {
    return mMessageService.getLikeMessages(messageType)

Leave a Reply

Your email address will not be published. Required fields are marked *