rx-java - 使用 RxJava 刷新透明身份验证 token

Question

我希望在我的 Java 应用程序中使用 Observable 包装经过身份验证的调用，如果需要，它将在发出请求之前重新授权我的用户。我使用来自 Shape Code Blog 的这种方法在 Objective-C 中使用 ReactiveCocoa 做了类似的事情。 :

- (RACSignal *)doRequestAndRefreshTokenIfNecessary:(RACSignal *)requestSignal {
    return [requestSignal catch:^(NSError *error) {
        // Catch the error, refresh the token, and then do the request again.
        BOOL hasRefreshToken = [UserManager sharedInstance].refreshToken != nil;
        BOOL httpCode401AccessDenied = error.code == -1011;
        if (httpCode401AccessDenied && hasRefreshToken) {
            NSLog(@"Will attempt to refresh access token.");
            return [[[self refreshToken] ignoreValues] concat:requestSignal];
        }
        return requestSignal;
    }];
}

RACSignal 是对 Observable 的 ReactiveCocoa 模拟。此方法创建一个 RACSignal，它将捕获由初始请求发出的任何错误信号，如果该错误符合给定的条件(在这种情况下，访问被拒绝并且有一个可用的身份验证 token )，则会透明地触发新的 RACSignal 以使用刷新 token 然后重新发送原始信号。

RxJava 是否提供类似的功能？我找不到与 catch: 类似的内容在 Combining Observables文档。

Answer 1

看起来 RxJava 中没有任何预先存在的东西，但这似乎可以解决问题(从 ReactiveCocoa 捕获的非常直接的翻译):

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);
                    }
                });
            }
        });
    }
}

并因此使用:

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);
        }
    });
}