使用RXJava进行缓存处理

Question

我正在尝试用rxJava实现这个工作流程,但我确定我是在滥用还是做错了.

• 用户要求登录
• 如果loginResult在缓存中可用,则"发出"缓存的LoginResult
• 实际上,如果一切都成功,则实际执行对Web服务的请求并缓存结果
• 如果发生错误,则最多重试3次,如果有第4次,则清除缓存.

这是我完整的代码片段.

public class LoginTask extends BaseBackground<LoginResult> {
  private static CachedLoginResult cachedLoginResult = new CachedLoginResult();
  private XMLRPCClient xmlrpcClient;
  private UserCredentialsHolder userCredentialsHolder;

  @Inject
  public LoginTask(XMLRPCClient client, UserCredentialsHolder userCredentialsHolder) {
    this.xmlrpcClient = client;
    this.userCredentialsHolder = userCredentialsHolder;
  }

  @Override
  public LoginResult performRequest() throws Exception {
    return UserApi.login(
        xmlrpcClient,
        userCredentialsHolder.getUserName(),
        userCredentialsHolder.getPlainPassword());


  }

  @Override
  public Observable<LoginResult> getObservable() {
    return cachedLoginResult.getObservable()
        .onErrorResumeNext(
            Observable.create(
                ((Observable.OnSubscribe<LoginResult>) subscriber -> {
                  try {
                    if (!subscriber.isUnsubscribed()) {
                      subscriber.onNext(performRequest()); // actually performRequest
                    }
                    subscriber.onCompleted();
                  } catch (Exception e) {
                    subscriber.onError(e);
                  }
                })
            )
                .doOnNext(cachedLoginResult::setLoginResult)
                .retry((attempts, t) -> attempts < 3)
                .doOnError(throwable -> cachedLoginResult.purgeCache())
        );
  }


  private static class CachedLoginResult {
    private LoginResult lr = null;
    private long when = 0;

    private CachedLoginResult() {
    }

    public boolean hasCache() {
      return lr != null && when + TimeUnit.MILLISECONDS.convert(30, TimeUnit.MINUTES) > System.currentTimeMillis();
    }

    public void setLoginResult(LoginResult lr) {
      if (lr != null) {
          this.lr = lr;
          this.when = System.currentTimeMillis();
      }
    }

    public void purgeCache() {
      this.lr = null;
      this.when = 0;
    }

    public Observable<LoginResult> getObservable() {
      return Observable.create(new Observable.OnSubscribe<LoginResult>() {
        @Override
        public void call(Subscriber<? super LoginResult> subscriber) {
          if (!subscriber.isUnsubscribed()) {
            if (hasCache()) {
              subscriber.onNext(lr);
              subscriber.onCompleted();
            } else {
              subscriber.onError(new RuntimeException("No cache"));
            }
          }
        }
      });
    }
  }
}

由于我无法找到任何类似的例子,我开始在1天前"玩"rxjava,我不确定我的实现.

感谢您的时间.

Answer 1

我认为这段代码没问题，干得好:)

Observable.create您在您的中使用是正确的LoginTask，因为否则调用的结果可能会在内部缓存，然后就retry没有多大帮助......

CachedLoginResult这对于我来说是没有必要的Observable。Observable.just在这里，您可以通过使用实用方法来简化代码Observable.error，例如：

public Observable<LoginResult> getObservable() {
  if (hasCache()) {
      return Observable.just(lr);
  } else {
      return Observable.error(new RuntimeException("No cache"));
  }
}

注意：just存储您告诉它在内部发出的值，以便重新订阅将始终产生该值。Observable.just(performRequest()).retry(3)这就是我上面暗示的，例如，您不应该这样做，因为performRequest只会被调用一次。