rx-java - 使用 repeatWhen() 的动态延迟值

Question

现在我正在用 RxJava 实现一些轮询逻辑。我应该多次轮询端点，直到它告诉我停止。此外，每个响应都会返回一个我应该在再次轮询之前延迟的时间。我的逻辑现在看起来像这样:

service.pollEndpoint()
    .repeatWhen(observable -> observable.delay(5000, TimeUnit.MILLISECONDS))
    .takeUntil(Blah::shouldStopPolling);

现在我将延迟值硬编码为 5000，但我希望它取决于轮询响应中的值。我尝试使用返回 Observable.just(pollResponse).repeatWhen(observable -> observable.delay(pollResponse.getDelay(), TimeUnit.MILLISECONDS)) 的平面 map ，但这似乎不是正确的想法，因为它与源 Observable 混淆了。我觉得这是我忽略的简单事情。谢谢!

Answer 1

正如@JohnWowUs 提到的，您需要带外通信，但如果您多次订阅该序列，则可以使用 defer拥有每个订阅者的状态:

Observable.defer(() -> {
    int[] pollDelay = { 0 };
    return service.pollEndpoint()
    .doOnNext(response -> pollDelay[0] = response.getDelay())
    .repeatWhen(o -> o.flatMap(v -> Observable.timer(pollDelay[0], MILLISECONDS)))
    .takeUntil(Blah::shouldStopPolling);
});