RxJava 棘手的 startWith(Observable)

Question

以下代码仅在 observable2 完成后才从 observable1 发出项目。

observable1.startWith(observable2)
           .subscribe()

我需要实现另一种行为

observable1 ->       0   1   2   3
observable2 -> 1   2   3   4   5   6

observable1.startWithDefault(observable2)
            -> 1   2 0   1   2   3

第二个可观察对象仅在第一个可观察对象为空时发出项目，然后发出第一个可观察对象中的项目。

我无法仅使用基本运算符找到正确的解决方案，自定义运算符 startWithDefault 的正确 RxJava 2 实现应该是什么样子？

聚苯乙烯

observable1.subscribe()
observable2.takeUntil(observable1).subscribe()

不是正确的解决方案，因为在 observable1 立即发出时存在竞争

Answer 1

方向很好，但是您需要publish(Function)共享observable1的信号，并且concatEager在附加开关时不要丢失其中的元素：

observable1.publish(o -> 
    Observable.concatEager(observable2.takeUntil(o), o)
)
.subscribe(...)