Retry with Predicate

[benjchristensen]

Several people have expressed need for conditional retry behavior. I suggest a new retry overload that simply takes a predicate that receives the Throwable.

public final Observable<T> retry(Func1<Throwable, Boolean> predicate)

This would allow deciding whether to retry based on the Throwable and/or count which can be maintained by the Func1.

Or we could use a Func2 and pass in the count:

public final Observable<T> retry(Func2<Integer, Throwable, Boolean> predicate)

[benjchristensen]

Here is the type of thing that has to be done for conditional retries without this overload:

import java.util.concurrent.atomic.AtomicInteger;

import rx.Observable;
import rx.Subscriber;

public class ConditionalRetry {

    public static void main(String[] args) {

        final AtomicInteger c = new AtomicInteger();
        Observable<String> oWithRuntimeException = Observable.create((Subscriber<? super String> s) -> {
            System.out.println("Execution: " + c.get());
            if (c.incrementAndGet() < 3) {
                s.onError(new RuntimeException("retryable"));
            } else {
                s.onNext("hello");
                s.onCompleted();
            }
        });

        final AtomicInteger c2 = new AtomicInteger();
        Observable<String> oWithIllegalStateException = Observable.create((Subscriber<? super String> s) -> {
            System.out.println("Execution: " + c2.get());
            if (c2.incrementAndGet() < 3) {
                s.onError(new RuntimeException("retryable"));
            } else {
                s.onError(new IllegalStateException());
            }
        });

        subscribe(oWithRuntimeException);
        subscribe(oWithIllegalStateException);
    }

    public static void subscribe(Observable<String> o) {
        o = o.materialize().flatMap(n -> {
            if (n.isOnError()) {
                if (n.getThrowable() instanceof IllegalStateException) {
                    return Observable.just(n);
                } else {
                    return Observable.error(n.getThrowable());
                }
            } else {
                return Observable.just(n);
            }
        }).retry().dematerialize();

        o.subscribe(System.out::println, t -> t.printStackTrace());
    }
}

[benjchristensen]

Completed.