RxJava retryWhen 重新订阅传播

Question

我在 Android 应用程序中使用 Retrofit 和 RxJava 进行通信，并且必须处理解析来自看似正常的 HTTP 响应（状态 200 代码）的响应时的错误。

我还使用 retryWhen 运算符实现了一种处理错误的方法，该运算符连接到用户的输入以决定是否重试。这通过重新订阅原始 Observable.

来实现

我尝试的第一种方法是这样的：

services.getSomething()
  .map(response -> {
    if (checkBadResponse(response)) {
      throw new RuntimeException("Error on service");
    } else {
      return parseResponse(response);
    }
  }).retryWhen(this::shouldRetry);

这样就不会再调用服务了。 retryWhen 操作员似乎无法重新订阅服务的 Observable.

最终的工作是实施另一个不发送 onCompleted 转发并将其与 lift 一起使用的运算符，如下所示：

public class CheckResponseStatus<T> implements Observable.Operator<ResponsePayload<T>, ResponsePayload<T>> {
    @Override
    public Subscriber<? super ResponsePayload<T>> call(Subscriber<? super ResponsePayload<T>> subscriber) {
        return new Subscriber<ResponsePayload<T>>() {
            private boolean hasError = false;

            @Override
            public void onCompleted() {
                if (!hasError)
                    subscriber.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                hasError = true;
                subscriber.onError(e);
            }

            @Override
            public void onNext(ResponsePayload<T> response) {
                if (response.isOk()) {
                    subscriber.onNext(response);
                } else {
                    hasError = true;
                    subscriber.onError(new RuntimeException(response.getMessage()));
                }
            }
        };
    }
}

像这样使用它：

services.getSomething()
  .lift(new CheckResponseStatus())
  .map(response -> parseResponse(response))
  .retryWhen(this::shouldRetry);

这是处理它的正确方法还是有更简单、更好的方法？

Answer 1

这看起来像是 rx-java 实现中的错误。无论如何，从 map 函数抛出异常是一件坏事，因为该函数应该是纯函数（例如，没有副作用）。您应该在您的情况下使用 flatMap 运算符：

services.getSomething()
  .flatMap(response -> {
    if (checkBadResponse(response)) {
      return Observable.<ResponseType>error(new RuntimeException("Error on service"));
    } else {
      return Observable.<ResponseType>just(parseResponse(response);
    }
  }).retryWhen(this::shouldRetry);

上面的代码按预期工作，如果发生错误会重试请求。

RxJava retryWhen 重新订阅传播

RxJava retryWhen resubscribe propagation

java

android

java-8

rx-java

retrofit