如何有条件地在RxJava流中添加异步操作？

Question

这里是我想要做的（使用科特林和RxJava）

makeServerCall() 
       .doOnNext { 
        doStuff(it) 
       } 
       //TODO: if it == 0, call asyncOperation() and wait for its callback to fire 
       //before running the rest of the stream. Otherwise immediately run the rest 
       //of the stream 
       .flatMap { 
        observable1(it) 
        observable2(it) 
        Observable.merge(
          getSpotSearchObservable(observable1), 
          getSpotSearchObservable(observable2) 
       } 
       .subscribeBy(onNext = { 
       allDone() 
        view? 
       }) 

如何在呼叫挤asyncOperation()，使流等待其余为它的回调火的简化版本，但只有当某个条件得到满足时？这似乎可能是Rx中的一个微不足道的操作，但没有想到明显的解决方案。

Answer 1

FlatMap it！

.flatMap { 
    if (it == 0) { 
     return@flatMap asyncOperation() 
      .ignoreElements() 
      .andThen(Observable.just(0)) 
    } 
    return@flatMap Observable.just(it) 
} 
.flatMap { 
    observable1(it) 
    observable2(it) 
    Observable.merge(
     getSpotSearchObservable(observable1), 
     getSpotSearchObservable(observable2) 
    ) 
}