RxJava 2：在向订阅者发出重试通知时可重试完成答案

【问题标题】：RxJava 2: Retry Completable while emitting retry notifications to subscribersRxJava 2：在向订阅者发出重试通知时可重试完成
【发布时间】：2017-02-24 17:59:36
【问题描述】：

我是 RxJava 2 的新手，想重试 Completable 服务器 API 调用直到成功，同时发出重试尝试通知，以便我的 UI 可以向用户显示重试状态。

类似这样的：

public Observable<RetryAttempt> retryServerCall() {

    // execute Completable serverCall()

    // if an error is thrown, emit new RetryAttempt(++retryCount, error) to subscriber

    // retry until serverCall() is successful
}

public Completable serverCall();

public class RetryAttempt {
    public RetryAttempt(int retryCount, Throwable cause);
}

我尝试了几种不同的方法，但遇到了障碍。最接近的是这种方法，创建一个封闭的 Observable 并显式调用 onNext() / onComplete() / onError()。

public Observable<RetryAttempt> retryServerCall() {
    final int[] retryCount = {0};
    return Observable.create(e ->
        serverCall()
                .doOnError(throwable -> e.onNext(new RequestHelp.RetryAttempt(++retryCount[0], throwable)))
                .retry()
                .subscribe(() -> e.onComplete(), throwable -> e.onError(throwable)));
}

也许这是一个无关紧要的问题，但我不得不为retryCount 使用final 数组以避免出现variable used in lambda should be final or effectively final 错误。

我知道使用 Rx voodoo 必须有更好的方法来完成此任务。非常感谢任何指导！

【问题讨论】：

你不想完全像这样做，因为你会失去取消订阅的信号。
@Tassos 是的。我可以使用subscribeWith 为内部Observable 获取Disposable，然后通过setDisposable 处理，对吗？

标签： rx-java rx-java2

【解决方案1】：

public Single<List<Farmer>> getAllFarmers(long timestamp) {

    return  Observable.fromCallable(() -> mapiFactory.getAllFarmerAboveTime(timestamp))
            .doOnError(throwable -> Log.d(TAG, "Error calling getAllFarmers: "+throwable.getMessage()))
            .retryWhen(new RetryWithDelay(5,1000))
            .concatMap(farmersResponse -> Observable.fromIterable(farmersResponse.farmer))
            .filter(farmer -> !StringUtils.isBlank(farmer.cnic))
            .map(this::validateCnic)
            .distinct(farmer -> farmer.cnic)
            .toList();

}

当 fromCallable() 方法抛出异常 .retryWhen(new RetryWithDelay(5,1000)) 将在这里执行时，我们从 1000 开始以指数延迟重试 api 5 次

这里是 RetryWithDelay

public class RetryWithDelay implements Function<Observable<Throwable>,
  Observable<?>> {

private final int _maxRetries;
private final int _retryDelayMillis;
private int _retryCount;

public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
    _maxRetries = maxRetries;
    _retryDelayMillis = retryDelayMillis;
    _retryCount = 0;
}


@Override
public Observable<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {

    return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
        @Override
        public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {

              if (++_retryCount < _maxRetries) {

                // When this Observable calls onNext, the original
                // Observable will be retried (i.e. re-subscribed)

                Log.d(TAG, String.format("Retrying in %d ms", _retryCount * _retryDelayMillis));

                return Observable.timer(_retryCount * _retryDelayMillis, TimeUnit.MILLISECONDS);
            }

            // Max retries hit. Pass an error so the chain is forcibly completed
            // only onNext triggers a re-subscription (onError + onComplete kills it)
            return Observable.error(throwable);
        }

    });
}

}

【讨论】：