Rxjava Retry With Exponential Backoff

RetryWithExponentialBackOff.java

import java.io.IOException;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

/**
 * Created by jimmy on 30/05/2017.
 */

public class RetryWithExponentialBackOff implements Func1<Observable<? extends Throwable>, Observable<?>> {

  private final int maxRetries;
  private final int retryDelayMillis;
  private long currentRetryCount;

  public RetryWithExponentialBackOff(final int maxRetries, final int retryDelayMillis) {
    this.maxRetries = maxRetries;
    this.retryDelayMillis = retryDelayMillis;
    this.currentRetryCount = 0;
  }


  @Override
  public Observable<?> call(Observable<? extends Throwable> errors) {
   return errors
            .zipWith(Observable.range(1, maxRetries > 0 ? maxRetries : Integer.MAX_VALUE), (Func2<Throwable, Integer, Throwable>) (throwable, attemptNumber) -> throwable)
            .flatMap(new Func1<Throwable, Observable<Long>>() {
              @Override
              public Observable<Long> call(Throwable throwable) {
                if (throwable instanceof IOException) {
                  long newInterval = retryDelayMillis * (currentRetryCount * currentRetryCount);
                  if (newInterval < 0) {
                    return Observable.error(new Throwable("Interval out of bounds!"));
                  }

                  return Observable.timer(newInterval, TimeUnit.MILLISECONDS, Schedulers.immediate());
                } else {
                  return Observable.error(throwable);
                }
              }
            });
  }
}

.java 가 빠져있어서 신택스 하이라이팅이 안되는군요.