定期的に観測可能なストリーム

Question

私はイベントで終了する観察可能なイベントのストリームを持っています。しかし、いったん終了すると、最初からストリームをやり直す必要があります。ストリームの表現は次のとおりです。

awaitStartEvent() 
    .switchMap(value -> awaitSecondEvent(value)) 
    .subscribe(result -> { 
    doSomethingWithResult(result); 
    // need to start at awaitStartEvent again here 
    }, error -> { 
    handleError(error); 
    // need to start at awaitStartEvent again here 
    } 

この状況を処理するにはどうすればよいでしょうか？

Answer 1

@ maxostのコメントによると、Repeat演算子を使用してください。エラーが発生してもそれを継続したい場合は、Retry演算子もそこに追加する必要があります。ここでは、最終的なサブスクリプションがこれらのエラーを見逃すことはありませんので、handleErrorロジックを置く必要があります。

Answer 2

あなたはrepeatとretryが必要ですが、あなたが質問にhandleError副作用を持っているので、例外がretryオペレータによって処理するために下流に伝播されることはありません。

ここでは、演算子のクラスdoを使用しています。

awaitStartEvent() 
    .switchMap(value -> awaitSecondEvent(value)) 
    .doOnNext(result -> { doSomethingWithResult(result); }) 
    .doOnError(error -> { handleError(error); }) 
    .repeat() 
    .retry() 
    .subscribe(); 

Answer 3

awaitStartEvent() 
    .switchMap(value -> awaitSecondEvent(value)) 
    .doOnNext(result -> doSomethingWithResult(result)) //handle result 
    .repeat() //resubsribe when onComplete called 
    .doOnError(error -> handleError(error)) //handle error 
    .retry() //resubscribe when onError called 
    .subscribe();