RxJava: retryWhen with retry limit

Question

I am new to ReactiveX and reactive programming in general. I need to implement a retry mechanism for Couchbase CAS operations, but the example on the Couchbase website shows a retryWhen which seems to retry indefinitely. I need to have a retry limit and retry count somewhere in there.

The simple retry() would work, since it accepts a retryLimit, but I don't want it to retry on every exception, only on CASMismatchException.

Any ideas? I'm using the RxJava library.

Answer 1

In addition to what Simon Basle said, here is a quick version with linear backoff:

.retryWhen(notification ->
    notification
    .zipWith(Observable.range(1, 5), Tuple::create)
    .flatMap(att ->
            att.value2() == 3 ? Observable.error(att.value1()) : Observable.timer(att.value2(), TimeUnit.SECONDS)
    )
)

Note that "att" here is a tuple which consists of both the throwable and the number of retries, so you can very specifically implement a return logic based on those two params.

If you want to learn even more, you can peek at the resilient doc I'm currently writing: https://gist.github.com/daschl/db9fcc9d2b932115b679#retry-with-delay

Answer 2

reviving this thread since in the Couchbase Java SDK 2.1.2 there's a new simpler way to do that: use the RetryBuilder:

Observable<Something> retryingObservable =
sourceObservable.retryWhen(
  RetryBuilder
    //will limit to the relevant exception
    .anyOf(CASMismatchException.class)
    //will retry only 5 times
    .max(5)
    //delay doubling each time, from 100ms to 2s
    .delay(Delay.linear(TimeUnit.MILLISECONDS, 2000, 100, 2.0))
  .build()
);