如何在RxJava中的重试请求错误后更改参数

Question

我使用Retrofit 2.0向服务器发送登录请求，服务器返回客户端会话令牌，我必须在其他请求中使用该令牌。但是这个令牌有限的生命周期，当它过期时，服务器会返回HTTP错误401。

当我遇到这个错误时，我尝试使用以下代码重新登录：

    holder.getApi(GuideProfileApi.class)
      .getProfile(String.valueOf(holder.getServerId()), holder.getServerToken())
      .subscribeOn(Schedulers.io())
      .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
         @Override
         public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
           return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
             @Override
             public ObservableSource<?> apply(Throwable throwable) throws Exception {
               if (throwable instanceof HttpException &&  ((HttpException)throwable).code() == 401) {
                 RegistryLoginResult loginResult = holder.login().blockingSingle();
                 return holder.getApi(GuideProfileApi.class)
                    .getProfile(String.valueOf(loginResult.getUserId()), loginResult.getSessionToken());
               }
               return Observable.error(throwable);
            }
        });
     }
  })
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Consumer<ProfileResult>() {
    @Override
    public void accept(ProfileResult profileResult) throws Exception {
      Log.d("Result", profileResult.toString());
    }
  }, new Consumer<Throwable>() {
      @Override
      public void accept(Throwable throwable) throws Exception {
        Log.e("Result", throwable.getLocalizedMessage());
      }
});

我发送了一个重试请求，但是请求参数和之前登录失败的请求一样。在再次发送请求之前，我如何更改请求参数？

- Pavel

Answer 1

您可以使用retryWhen，但问题是您的retryWhen会重试您在惰性时刻创建的相同可观察对象。您的解决方案是使用defer运算符来获取host()，因为defer不是在定义时创建可观察对象，而是在订阅时消耗它。

Observable.defer(()-> holder.getApi(GuideProfileApi.class)
          .getProfile(String.valueOf(holder.getServerId()),holder.getServerToken()))
  .subscribeOn(Schedulers.io())
  .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
     @Override
     public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
       return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
         @Override
         public ObservableSource<?> apply(Throwable throwable) throws Exception {
           if (throwable instanceof HttpException &&  ((HttpException)throwable).code() == 401) {
             RegistryLoginResult loginResult = holder.login().blockingSingle();
             return holder.getApi(GuideProfileApi.class)
                .getProfile(String.valueOf(loginResult.getUserId()), loginResult.getSessionToken());
           }
           return Observable.error(throwable);
        }
    });
 }
  })
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Consumer<ProfileResult>() {
    @Override
    public void accept(ProfileResult profileResult) throws Exception {
      Log.d("Result", profileResult.toString());
    }
      }, new Consumer<Throwable>() {
          @Override
          public void accept(Throwable throwable) throws Exception {
            Log.e("Result", throwable.getLocalizedMessage());
          }
    });

您可以在此处查看重试的一些示例：https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/errors/ObservableExceptions.java

该链接提供了有关如何实现重试的示例代码。

- paul

Answer 2

您使用了错误的操作符。 retryWhen 会在遇到错误时重试您的原始可观测对象。您需要的是 onErrorResumeNext。可以尝试以下方式：

 holder.getApi(GuideProfileApi.class)
      .getProfile(String.valueOf(holder.getServerId()), holder.getServerToken())
      .subscribeOn(Schedulers.io())
      .onErrorResumeNext(new Function<Throwable, ObservableSource<?>>() {
        @Override
         public ObservableSource<?> apply(Throwable throwable) {
               if (throwable instanceof HttpException &&  ((HttpException)throwable).code() == 401) {
                 RegistryLoginResult loginResult = holder.login().blockingSingle();
                 return holder.getApi(GuideProfileApi.class)
                    .getProfile(String.valueOf(loginResult.getUserId()), loginResult.getSessionToken());
               }
               return Observable.error(throwable);
        }
    }) 
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Consumer<ProfileResult>() {
    @Override
    public void accept(ProfileResult profileResult) throws Exception {
      Log.d("Result", profileResult.toString());
    }
  }, new Consumer<Throwable>() {
      @Override
      public void accept(Throwable throwable) throws Exception {
        Log.e("Result", throwable.getLocalizedMessage());
      }
});

- JohnWowUs

---

如何排除 return holder.getApi(GuideProfileApi.class) .getProfile(String.valueOf(loginResult.getUserId()), loginResult.getSessionToken()); 并更改 原始可观察对象 的参数？ - Pavel

我相信在RxJava中的Observables是不可变的。一旦你创建了一个observable，就不能回头去改变它。 - JohnWowUs

感谢您的咨询。 - Pavel

但是如果第二个可观察到一个错误并且您想要重试怎么办？看起来保罗提供的推迟解决方案更加干净。 - Scott Cooper