RxJava 2：再試行ユーザーの再試行通知を送信中に完了

Question

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

おそらく、そのやや周辺の問題が、私はエラーvariable used in lambda should be final or effectively finalを避けるためにretryCountためfinal配列を使用する必要がありました。

これを達成するには、Rxブードーを使用する方がよいことがわかっています。どんな指導も大歓迎です！

Answer 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（新しいRetryWithDelay（5,1000））は、ここで実行されます、我々は1000年

から始まる指数遅れで5回のAPIを再試行し、ここ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); 
     } 

    }); 
} 

}