RxJava retryWhen resubscribe propagation

Question

I'm using Retrofit with RxJava in an Android app for communications and have to handle error on parsing the response from a seemly ok HTTP response (status 200 code).

I have also implemented a way of handling the error using retryWhen operator which is connected to user's input to decide whether to retry it or not. This works by resubscribing to the original Observable.

The first approach I have tried was to have something like this:

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

With this the service is not called again. It seems the retryWhen operator cannot resubscribe to the service's Observable.

What end up working was implementing another operator which doesn't send the onCompleted forward and use it with lift like the following:

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()));
                }
            }
        };
    }
}

Using it like:

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

Is this the correct way of dealing with it or is there a simpler, better way?

Answer 1

It's looks like a bug in rx-java implementation. Anyway, throwing an exception from map function is a bad thing since the function is supposed to be pure (e.g. without side effects). You should use a flatMap operator in your case:

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);

The code above works as expected and really retries the request if error occurs.