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

Question

这是我正在尝试做的简化版本（使用 Kotlin 和 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 {
    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)
    )
}