合并处理两个不同类型的 RxJava Observable

Question

我的目标

我想检查服务器的令牌是否仍然有效，假设我通过调用 getter : preferenceHelper.isTokenValid() 就知道该信息。然后，如果令牌无效，调用请求获取新令牌并在本地更新令牌，THEN，继续下一个请求post指向服务器的点.那是因为我需要一个有效的令牌才能发出任何进一步的服务器请求。

假设我有两个服务器请求 returns Observable:

此请求旨在获取服务器令牌，然后在收到后更新它。

Observable<Response<EntityToken>> updateServerToken = retrofitApi.authenticate(username,password);

这个请求是为了post当前位置到服务器，如果成功，return保存的点

Observable<Response<EntityPoint>> updateServerToken = retrofitApi.postPoint(point);

我目前面临的问题：

需要合并的两个 observable 来自不同的类型
仅在需要时执行令牌更新请求
等待令牌更新请求完成，然后再执行到 post 点的请求

我应该如何编写我的 RxJava Observable 来满足所有这些条件？

Answer 1

由于三个调用之间存在依赖关系，merge没有任何意义。相反，使用 flatMap:

Observable<Response<EntityPoint>> response = 
retrofitApi.isTokenValid()
.flatMap(isValid ->
     isValid
     ? Observable.just("")
     : retrofitApi.authenticate(username,password)
                  .doOnNext(token -> doSomethingWithTheToken(token)
)
.flatMap(dummy -> retrofitApi.postPoint(point));

Answer 2

首先，我将创建一个方法来检查 entityToken 是否有效。如果有效，请使用 Observable.just() 但您必须以某种方式创建 Response 的实例。如果无效，则在您的要求 retrofitApi.authenticate() 中使用 API 调用服务器。无论采用哪种路径，方法 getTokenObservable() 都会发出 Observable>。

public Observable<Response<EntityToken>> getTokenObservable(EntityToken entityToken, String username, String password) {    
  boolean isTokenValid = preferenceHelper.isTokenValid(entityToken);
  if (isTokenValid) {
    //my assumption that you have something like this
    Response<EntityToken> responseToken = new Response<EntityToken>(); 
    responseToken.setEntityToken(entityToken);
    return Observable.just(new Response<EntityToken>(entityToken.class));
  } else {
    Observable<Response<EntityToken>> updateServerToken = retrofitApi.authenticate(username, password);
    return updateServerToken;
  }
}

然后在调用它时，使用 flatMap() 来接收 Observable> 的发射和 returns Observable> 的发射。订阅并照常进行。

Observable<Response<EntityToken>> updatePointObservable = getTokenObservable(entityToken, username, password);
updatePointObservable
.flatMap(new Func1<Response<EntityToken>, Observable<Response<EntityPoint>>>() {
    @Override
    public Observable<Response<EntityPoint>> call(Response<EntityToken> responseToken) { 
        EntityToken entityToken = responseToken.getEntityToken(); //my assumption
        saveTokenLocally(entityToken); //this is where you save your token locally, change to the right method that you have
        Observable<Response<EntityPoint>> updateServerTokenObservable = retrofitApi.postPoint(point, entityToken); //pass your entityToken to the call?
        return updateServerTokenObservable;     
    }
})
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Response<EntityPoint>>() {
    @Override
    public void onCompleted() {
        //your own logic
    }

    @Override
    public void onError(Throwable e) {
        //your own logic
    }

    @Override
    public void onNext(Response<EntityPoint> entityPoint) {
        //your own logic
    }
});

合并处理两个不同类型的 RxJava Observable

Merge and handle two RxJava Observable of different types

android

rx-java

retrofit2