RxJava + GRPC 从错误中恢复并重试请求答案

【问题标题】：RxJava + GRPC recover from errors and retry requestsRxJava + GRPC 从错误中恢复并重试请求
【发布时间】：2018-05-24 01:31:11
【问题描述】：

我正在尝试使用 RxJava 和 GRPC 从错误中恢复。这是我观察到的：

Observable<Object> observable = Observable.fromCallable(() -> {
        try {
            Grpc.MyRequest request = Grpc.MyRequest.newBuilder()
                    .setToken(mToken)
                    .build();
            Grpc.MyResponse reply = mStub.mytest(request);
            return reply;
        } catch (Exception e) {
            ///
        }
    }).cache();

这是订阅：

observable.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> {
            })
            .subscribe((result) -> {
                MyResponse res = ((MyResponse) result);
                if (res.getCode()!=0) {
                    //Check error code and try to refresh token and repeat this request after.
                } 
            },throwable -> {
                throwable.printStackTrace();
            });

因此，当我从 GRPC 服务收到错误时，根据错误代码，我想尝试通过执行另一个请求来从中恢复，然后重复原始请求。我不确定如何使用 RxJavaretrywhen。做这样的事情最优雅的方式是什么？

【问题讨论】：

标签： android rx-java grpc grpc-java

【解决方案1】：

观察者链中的错误恢复确实需要一些踢踏舞，而且绝不是优雅的。但是，它可以包含在观察者链中。

boolean isRecoverable( Throwable t ) {
  // this test can be as sophisticated as you want
  if ( t instanceof StatusRuntimeException ) {
    return true;
  }
  return false;
}

...
.retryWhen( throwableObservable -> 
  throwableObservable.flatMap( t -> isRecoverable( t ) 
           ? Observable.just("") 
           : Observable.error( t ) )
...

这种方法允许您决定要如何处理错误。您可以添加延迟 just() 以便您不会立即重试。您可以返回获取新 API 令牌的 Observable，而不是 just()。

【讨论】：