継続的に放射するRxJava Observableの作成方法は？

Question

RxJava BlockingObservableを作成しようとしていますが、（条件== true）またはタイムアウトが発生するまで、Xミリ秒ごとに変数の値を出力します。

以下のコードは私が望むものに近いようですが、常にONCEを出して終了します。奇妙なのは、私がtakeUntil()の状態になっていることです。これは決して真実ではありません。この観測可能なものが継続的に放出され、最終的にはタイムアウトすると期待していますが、そうではありません。

私はここで間違っている/何をしていますか？

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

。サンプルは、あなたが思う通りのことをしません。サンプルレートは上記のObservableを10秒に1回（最大で）制限します。

Observable.fromCallable()は、イベントを1回だけ発行して完了します。

.sample()は10秒待機し、最後のイベント（存在する場合）を10秒ごとに送信します。したがって、イベントを1つしか持たないObservableにアタッチすると、1つのイベントが発生します。それから完了です。

あなたはおそらく（私は.netプログラマですので、私のケーシングなどを言い訳します）これはです。

編集：Javaが繰り返しイベントの間隔を使用することを伝えてくれた@akanokdに感謝します。

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呼び出しでこの答えを編集すること自由に感じ...