It doesn't look like anything pre-exists within RxJava, but this seems to do the trick okay (a pretty direct translation from ReactiveCocoa's catch):
public class CatchObservable {
public interface CaughtErrorHandler {
public Observable<?> onError(Throwable throwable);
}
public static <T> Observable catchObserver(final Observable<T> observable, final CaughtErrorHandler errorHandler) {
return Observable.create(new Observable.OnSubscribeFunc<T>() {
@Override
public Subscription onSubscribe(final Observer<? super T> subscriber) {
return observable.subscribe(new Observer<T>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}
@Override
public void onError(Throwable e) {
Observable observable = errorHandler.onError(e);
observable.subscribe(subscriber);
}
@Override
public void onNext(T args) {
subscriber.onNext(args);
}
});
}
});
}
}
And used thusly:
private Observable refreshAuthIfNecessary(final Observable<?> request) {
return CatchObservable.catchObserver(request, new CatchObservable.CaughtErrorHandler() {
@Override
public Observable<?> onError(Throwable throwable) {
return Observable.concat(WebServices.this.refreshAuth(), request);
}
});
}