RxJava无限重试与新的可观察性

Question

我正在为vertx使用RX-Ifified API，这个问题必须做一个潜在的无限重试直到成功循环，我想实现，但有困难。我是RXJava新手。

以下是我想做的：

null

vertx.eventBus().<JsonObject>sendObservable( "theAddress", aJsonObject );

vertx.eventBus().<JsonObject>sendObservable( "theAddress", aJsonObject )
     .retry()

vertx.eventBus().<JsonObject>sendObservable( "theAddress", aJsonObject )
     .retryWhen( error -> {
        return _vertx.eventBus().<JsonObject>sendObservable( "theAddress", aJsonObject )
      })

所以我想我可以使用RXJava的retryWhen（）运算符，它允许我在根observable发出错误时发出第二个observable。我认为，第二个可观察的代码可能与上面在步骤1中生成初始观察者的代码相同。

但是，retryWhen（）运算符（请参阅文档）不允许这第二个可观察到的对象发出错误，而不以错误结束订阅。

所以，我很难找到如何在这个链的第一部分设置一个潜在的无限重试循环。

vertx.eventBus().<JsonObject>sendObservable( "theAddress", aJsonObject )
     // imagine that retryWhen() accomplishes an infinite retry
     .retryWhen( error -> {
        return _vertx.eventBus().<JsonObject>sendObservable( ... )
      })
     .flatMap( response -> {
        // inspect response, if it has usable data,
        // return that data as an observable
        return Observable.from(response.data());

        // if the response has no usable data,
        // wait for some time, then start the whole process
        // all over again
        return Observable.timer(timeToWait).<WHAT GOES HERE?>;
     })

我创建的任何可观察到的东西似乎都需要链中的所有元素来达到这一点，这似乎是一种可能应该避免的递归。

在这一点上，任何帮助或建议都是非常欢迎的。

Answer 1

非常感谢RxJava Google小组的Ben Christensen指出defer（）操作符，它将在每次订阅时生成一个新的可观察的。然后，可以用标准的retry（）操作符组合它来获得有限重试。

因此，我的问题中第一个问题的最简单解决方案如下所示：

Observable.defer( () -> vertx.eventBus().<JsonObject>sendObservable( "theAddress", aJsonObject ) )
.retry()

...如果需要指数退避，可以在给defer（）运算符的工厂方法中使用适当的参数添加observable.timer（）。

我还在研究第二个问题。