Android 的 RxJava：公开异常并重试（有延迟）

Question

我有以下 Observable，它将每 30 秒执行一次带有 Retrofit 的 REST 调用：

 Subscription subscription = Observable.interval(0, REFRESH_INTERVAL, TimeUnit.SECONDS)
        .concatMap(new Func1<Long, Observable<Response>>() {
           @Override
           public Observable<Response> call(Long time) {
              return webservice.callRetrofitServiceWithRx(parameter);
           }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new UpdateSuccessAction(), new UpdateErrorAction());

可能会发生（尤其是 REST-Call）会抛出异常（例如，没有互联网连接）。

我想达到的目标：

Observable 应该 emit/expose 异常，以便我可以在 UI 上显示错误消息，但它应该继续发射项目（30 秒后重试）。

目前正在研究

如果我没有定义任何特殊行为，Observable 将发出异常并停止工作（= 30 秒内不重试）。
如果我尝试重试运算符，异常将被吞噬而不暴露，所以我无法在ui中显示错误。
如果我尝试 onErrorReturn 运算符，我可以处理异常，但据我所知无法重试。

解决方法

我目前的解决方法是重新订阅这个Observable，但我想知道是否有人有更优雅的解决方案。

Answer 1

我假设 doOnError 将满足您的需要（用于记录错误），并结合重试，例如：

Subscription subscription = Observable.interval(0, REFRESH_INTERVAL, TimeUnit.SECONDS)
    .concatMap(new Func1<Long, Observable<Response>>() {
       @Override
       public Observable<Response> call(Long time) {
          return webservice.callRetrofitServiceWithRx(parameter);
       }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .doOnError(new UpdateErrorAction())
    .retry()
    .subscribe(new UpdateSuccessAction());

Answer 2

在其他答案的帮助下找到解决方案。

起初我定义了一个 RetryWithDelay-Function，它在 30 秒后而不是立即开始重试。

 private static class RetryWithDelay
     implements Func1<Observable<? extends Throwable>, Observable<?>> {

  @Override
  public Observable<?> call(Observable<? extends Throwable> attempts) {
     return attempts.flatMap(new Func1<Throwable, Observable<?>>() {
        @Override
        public Observable<?> call(Throwable throwable) {
           return Observable.timer(CallBO.REFRESH_INTERVAL_IN_SEC,             }
     });
  }
}

然后我在这个 Observable-Chain 中使用了它：

Subscription subscription = Observable.interval(0, REFRESH_INTERVAL, TimeUnit.SECONDS)
.concatMap(new Func1<Long, Observable<Response>>() {
   @Override
   public Observable<Response> call(Long time) {
      return webservice.callRetrofitServiceWithRx(parameter);
   }
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnError(new UpdateErrorAction())
.retryWhen(new RetryWithDelay())
.subscribe(new UpdateSuccessAction());

Android 的 RxJava：公开异常并重试（有延迟）

RxJava for Android: Expose Exception and Retry (with delay)

android

rx-java

rx-android