如何创建连续发射的RxJava可观察？

Question

我正在尝试创建一个RxJava BlockingObservable，它将每隔X毫秒发出一个变量的值，直到（条件==true）或超时发生。

下面的代码似乎与我想要的很接近，但它总是发出一次，然后退出。奇怪的是，我在testolin（）中有一个永远不会正确的条件——我希望这个可观察到的持续发出并最终超时，但事实并非如此。

我错过了什么/做错了什么？

Observable.fromCallable(() -> getSendWindow())
        .sample(10, TimeUnit.MILLISECONDS)
        .timeout(30, TimeUnit.SECONDS)
        .takeUntil(sendWindow -> 1==2)
        .doOnError(throwable -> log.warn("Timed out waiting for send window to clear. Giving up."))
        .doOnCompleted(() -> {
            log.info("Send window cleared");
        })
        .toBlocking().forEach(sendWindow -> log.info("sendWindow={}, getSendWindow());

Answer 1

.示例并不像您认为的那样。采样率将上述可观测值限制为（最多）每10秒一次。

Observable.fromCallable（）只发出一次事件，然后完成。

<代码>。sample（）等待10秒，并每隔10秒发出最后一个事件（如果有）。因此，当您将其附加到只有一个事件的可观察对象时，它只发出一个事件。然后就完成了。

你可能真正想要的（我是一个. net程序员，所以原谅我的外壳等）是这个。

编辑：感谢@akanokd告诉我java对重复事件使用间隔。

Observable.interval(10, timeUnit.MILLISECONDS)
    .map(x -> getSendWindow())
    .takeUntil(sendWindow -> 1==2)
    .doOnError(throwable -> log.warn("Timed out waiting for send window to clear. Giving up."))
     .doOnCompleted(() -> {
            log.info("Send window cleared");
        })
    .toBlocking().forEach(sendWindow -> log.info("sendWindow={}, getSendWindow());

可以通过对JAVA特定版本的API调用来编辑此答案。。。