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：重试完成，同时向用户发出重试通知

回答

相关问题