java - RxJava棘手的startWith(Observable)

以下代码仅在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立即发出的情况下存在竞争

最佳答案

方向是好的，但是你需要publish(Function)来共享observable1的信号，并且当开关出现时concatEager不会丢失其中的元素：

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