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.