rx-java - 重复的可观察流

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

您需要repeatand retry，但由于您handleError在问题中有副作用，因此异常不会传播到下游以由retry操作员处理。

这里使用的do是操作符类。

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

Answer 2

根据@maxost 的评论，使用Repeat运算符。如果您也希望它在遇到错误时继续运行，那么您也需要在其中添加Retry操作员。这是您需要放置handleError逻辑的地方，因为最终订阅永远不会看到这些错误。

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();