RxJava 2：重試完成，同時向用戶發出重試通知

我是RxJava 2的新手，想重試Completable服務器API調用，直到成功，同時發出重試嘗試的通知，以便我的UI可以顯示重試狀態爲用戶。RxJava 2：重試完成，同時向用戶發出重試通知

事情是這樣的：

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使用final陣列retryCount。

我知道必須有更好的使用Rx voodoo來完成這個任務。任何指導，非常感謝！

來源

2017-02-24 HolySamosa

你並不想這樣做，正是這樣，因爲你將失去退訂信號。 –

@Tassos True。我可以使用'subscribeWith'來獲取'Observable'的'Disposable'，然後通過'setDisposable'處理，對嗎？ – HolySamosa

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））將得到執行我們在這裏重試API在指數延遲5倍，從1000

開始，這裏是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); 
     } 

    }); 
}

}

來源

2017-02-27 19:03:38

RxJava 2：重試完成，同時向用戶發出重試通知

回答

相關問題