从另一个可观察对象调用可观察对象

Question

我是 RxJava 框架的新手。我的系统使用令牌身份验证，当我想从 http 请求中获得任何响应时，由于过期时间，我需要刷新我的身份验证令牌。

但是在从一个 observable 调用 subscribe 方法到另一个 observable 后我遇到了问题

主类（第一个可观察对象）

service.getPosts(offset,limit, PreferencesUtils.getAccessTokenKey())
        .subscribeOn(Schedulers.computation())
        .observeOn(AndroidSchedulers.mainThread())
        .doOnNext(response -> {
            if (!response.isSuccessful()) throw new RuntimeException();
        })
        .subscribe(
            feeds -> mView.showNewsFeedItems(feeds.body()),
            error -> {
                Helper.refreshToken();
                service.getPosts(offset,limit,PreferencesUtils.getAccessTokenKey()); 
            }
        );

我的响应不成功，我需要调用refreshToken()，我可以在其中刷新令牌并将其保存到PreferencesUtils

Helper 类（第二个 observable）

service.refreshToken(PreferencesUtils.getAccessTokenKey(), PreferencesUtils.getRefreshTokenKey())
        .subscribeOn(AndroidSchedulers.mainThread())
        .observeOn(Schedulers.newThread())
        .subscribe(


// doesnt proceed this part of code
                response -> {
                    PreferencesUtils.saveAccessToken(response.body().getAccessToken());
                    PreferencesUtils.saveRefreshToken(response.body().getRefreshToken());
                },

                error -> Log.d(TAG, "refresh token exception: " + `error.getMessage())`
//

            );

当我使用调试器时，来自 Helper 类的 Second observable 无法正常工作，它不想继续此服务，并且在订阅时立即完成而没有任何错误（肯定是在等待完成来自主线程的所有操作）。由于订阅方法没有继续（没有将访问令牌保存到首选项）我无法执行我的service.getPosts() 函数。

请帮忙！我将不胜感激！

Answer 1

与其像那样拆分观察者链，不如使用retryWhen() 运算符来响应失败的令牌。

Observable.just(Boolean.TRUE)
    .flatMap( ignored -> service
                           .getPosts(offset,limit,
                                      PreferencesUtils.getAccessTokenKey()) )
    .subscribeOn(Schedulers.computation())
    .doOnNext(response -> {
        if (!response.isSuccessful()) throw new TokenException();
    })
    .retryWhen( error -> error.flatMap( f ->
      {
        if ( f instanceof TokenException ) {
          return service.refreshToken(PreferencesUtils.getAccessTokenKey(), PreferencesUtils.getRefreshTokenKey())
            .doOnNext( response -> {
                PreferencesUtils.saveAccessToken(response.body().getAccessToken());
                PreferencesUtils.saveRefreshToken(response.body().getRefreshToken());
            });
        }
        return Observable.error(f);
      })
    )
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe( /* do stuff */ );

retryWhen() 运算符接受一个 observable，它将获取上游抛出的任何异常。如果异常是 TokenException，它将尝试通过调用 refreshToken() 返回的 observable 来刷新令牌，然后将结果保存到您的首选项中。

编辑：使用可以重新订阅而不会出现问题的 observable 启动观察者链。当必须刷新访问令牌时，将使用新令牌调用getPosts()。