rx-java - Retrofit2+RxJava2，无效令牌，retryWhen() 重新订阅时如何更新流

Question

我在下面有这个简单的代码，它模拟了我目前试图完成的场景

mApiService.api().postSomethingWithAccessToken(request, "some_invalid_access_token")
            .subscribeOn(Schedulers.io())
            .retryWhen(new Function<Observable<Throwable>, ObservableSource<AccessToken>>() {

                @Override
                public ObservableSource<AccessToken> apply(Observable<Throwable> throwableObservable) throws Exception {
                    return mApiService.api().getAccessToken();
                }
            })
            .subscribeOn(Schedulers.io())
            .subscribe(new Observer<Void>() {
                @Override
                public void onSubscribe(Disposable d) {
                }

                @Override
                public void onNext(Void value) {
                }

                @Override
                public void onError(Throwable e) {

                    e.printStackTrace();
                    onError(e);
                }

                @Override
                public void onComplete() {
                }
            });

我只是列举它以明确我的目标：

使用当前访问令牌执行POST调用
如果它收到适当的错误（404,403、401 等）
执行GET调用以获得新的访问令牌
使用新的访问令牌重试整个序列

基于上面的代码和到目前为止我对.retryWhen()的理解，如果原始Observable ( . postSomethingWithAccessToken() )发生错误，它将执行，并在必要时重试（根据您在重试中的条件），什么这里发生的是.retryWhen()在外部 Observable 之前首先执行，导致不需要的重复请求，根据我目前的理解（代码），我该如何实现上面提到的那些事情？任何帮助将不胜感激。:(

编辑：当前解决方法：

mApiService.api().postSomethingWithAccessToken(request, preferences.getString("access_token", ""))
            .subscribeOn(Schedulers.io())
            .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {

                @Override
                public ObservableSource<?> apply(final Observable<Throwable> throwableObservable) throws Exception {

                    return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {

                        @Override
                        public ObservableSource<?> apply(Throwable throwable) throws Exception {

                            if (throwable instanceof HttpException) {

                                HttpException httpException = (HttpException) throwable;

                                if (httpException.code() == 401) {

                                    return mApiService.api().getAccessToken()
                                            .doOnNext(new Consumer<Authentication>() {
                                                @Override
                                                public void accept(Authentication authentication) throws Exception {
                                                    update(authentication);
                                                }
                                            });
                                }
                            }

                            return Observable.error(throwable);
                        }
                    });
                }
            })
            .subscribe(new Observer<Void>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.e("subscribe", "TOKEN : " + preferences.getString("access_token", ""));
                }

                @Override
                public void onNext(Void value) {
                    Log.e("onNext", "TOKEN : " + preferences.getString("access_token", ""));
                }

                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                }

                @Override
                public void onComplete() {
                    Log.e("Complete", "____ COMPLETE");
                }
            });

通过共享偏好更新令牌的方法

public void update(Authentication authentication) {
    preferences.edit().putString("access_token", authentication.getAccessToken()).commit();
}

我注意到（我放了一个日志）外部 observable 的订阅和 retryWhen 是在主线程执行的，但是重试/重新订阅的流正在跳过不同的调度程序的线程，这似乎是一个竞争条件:(

    onSubscrbie_outer_observable: Thread[main,5,main]
    RetryWhen: Thread[main,5,main]
    Throwable_FlatMap: Thread[RxCachedThreadScheduler-1,5,main]
    doOnNext(Token_Refresh): Thread[RxCachedThreadScheduler-1,5,main]
    Throwable_FlatMap: Thread[RxCachedThreadScheduler-2,5,main]
    doOnNext(Token_Refresh): Thread[RxCachedThreadScheduler-2,5,main]
    Throwable_FlatMap: Thread[RxCachedThreadScheduler-1,5,main]
    doOnNext(Token_Refresh): Thread[RxCachedThreadScheduler-1,5,main]
    // and so on...

score 3 · Accepted Answer

这里有几个问题：

您需要在重试时将访问令牌传递回该postSomethingWithAccessToken方法，否则您将使用相同的旧无效访问令牌重试。
当逻辑不正确时重试，您必须响应Observable收到的错误并将重试逻辑放在那里。正如您所说，此方法首先执行，而不是在发生错误时执行，这throwableObservable是对错误的响应，它将错误反映为排放（onNext()），您可以flatMap()每个错误和响应都带有错误（用于将错误传递到源流）完成，或者用onNext()一些对象来表示它重试。
一篇很棒的博文禁止 Dan Lew讨论这个问题。

所以你需要：
1）将访问令牌存储在可以通过访问令牌刷新更改它的地方。
2) 修复逻辑正确响应错误时的重试

这是一个建议代码：

postSomethingWithAccessToken(request, accessToken)
        .subscribeOn(Schedulers.io())
        .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
                   @Override
                   public ObservableSource<?> apply(
                           @NonNull Observable<Throwable> throwableObservable) throws Exception {
                       return throwableObservable.flatMap(
                               new Function<Throwable, ObservableSource<? extends R>>() {
                                   @Override
                                   public ObservableSource<? extends R> apply(
                                           @NonNull Throwable throwable) throws Exception {
                                       if (throwable.code == 401) { //or 404/403, just a pseudo-code, put your real error comparing logic here
                                           return getAccessToken()
                                                           .doOnNext(refreshedToken -> accessToken.updateToken(refreshedToken));
                                                   //or keep accessToken on some field, the point to have mutable
                                                   //var that you can change and postSomethingWithAccessToken can see
                                       }
                                       return Observable.error(throwable);
                                   }
                               });
                       }
                   }
        )
        .subscribeOn(Schedulers.io())
        .subscribe(new Consumer<Result>() {
                       @Override
                       public void accept(@NonNull Result result) throws Exception {
                           //handle result
                       }
                   }
        );

score 1 · Accepted Answer

非常感谢 yosriz，因为他为我指出了正确的方向来解决我的磨牙问题，我必须使用defer. 所以我在 GitHub 中遇到了这个问题，为什么当我使用 retryWhen 运算符时重新订阅源 observable 会发出相同的输出？

这与我现在遇到的问题完全相同，对于任何遇到相同问题的人来说，这里都是我的解决方案。

Observable
    .defer(new Callable<ObservableSource<?>>() {
        @Override
        public ObservableSource<?> call() throws Exception {
            // return an observable source here, the observable that will be the source of the entire stream;
        }
    })
    .subscribeOn( /*target thread to run*/ )
    .retryWhen( {
        // return a throwable observable here that will perform the logic when an error occurred
    })
    .subscribe( /*subscription here*/ )

或者这是我的解决方案的完整非 lambda

Observable
    .defer(new Callable<ObservableSource<?>>() {
        @Override
        public ObservableSource<?> call() throws Exception {
            return mApiService.api().postSomethingWithAccessToken(
                request, preferences.getString("access_token", ""));
        }
    })
    .subscribeOn(Schedulers.io())
    .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
        @Override
        public ObservableSource<?> apply(final Observable<Throwable> throwableObservable) throws Exception {
            return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                @Override
                public ObservableSource<?> apply(Throwable throwable) throws Exception {
                    if (throwable instanceof HttpException) {
                        HttpException httpException = (HttpException) throwable;
                        if (httpException.code() == 401) {
                            return mApiService.api().getAccessToken().doOnNext(new Consumer<Authentication>() {
                                    @Override
                                    public void accept(Authentication authentication) throws Exception {
                                        update(authentication);
                                    }
                                });
                        }
                    }
                    return Observable.error(throwable);
                }
            });
        }
    })
    .subscribe(new Observer<Void>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.e("subscribe", "TOKEN : " + preferences.getString("access_token", ""));
        }

        @Override
        public void onNext(Void value) {
            Log.e("onNext", "TOKEN : " + preferences.getString("access_token", ""));
        }

        @Override
        public void onError(Throwable e) {
            e.printStackTrace();
        }

        @Override
        public void onComplete() {
            Log.e("Complete", "____ COMPLETE");
        }
    });

.retryWhen()这里的关键点是“当操作员重新订阅源 observable 时如何修改/更新现有的源 observable”

rx-java - Retrofit2+RxJava2，无效令牌，retryWhen() 重新订阅时如何更新流

2 回答 2

Related

Reference