java - RxJava : How to repeat an observable chain when original condition is not met anymore?

Question

我想使用 repeat() 重新订阅一个 observable，但是当第一次订阅完成时，触发原始 observable 的条件不再满足。在这种情况下我该如何重新订阅观察者？

代码看起来像这样:

RxInput.onInput(grid)
    .flatMap(Grid::handle)
    .flatMap(Grid::check)
    .delay(300, TimeUnit.MILLISECONDS)
    .flatMap(Grid::clean)
    .repeat() // Does not work
    .subscribe(g -> {System.out.println("Finished");});

RxInput.onInput()的代码是这样的:

public static <T> Observable<T> onInput(T t) {
    return Observable.create(subscriber -> {
        if(InputSystem.isInputOn()) { // This is not true anymore when re-subscribing
            subscriber.onNext(t);
        }
    });
}

Answer 1

鉴于您似乎希望在 isInputOn 为 false 时阻止初始订阅的排放，然后像这样定义可观察值:

Observable.defer(
  () -> {
     if (!InputSystem.isInputOn())
       return Observable.empty();
     else 
       return yourInput
         .flatMap(Grid::handle)
         .flatMap(Grid::check)
         .delay(300, TimeUnit.MILLISECONDS)
         .flatMap(Grid::clean)
         .repeat();
    })
  .subscribe(...);

一个快速的额外说明，尽最大努力避免使用 Observable.create 就像上面的示例一样。如果您确实发出类似的东西，那么您需要将您的可观察值与 .onBackPressureXXX() 结合起来，因为您可能会从 flatMap 等运算符得到 MissingBackPressureException下游。