RxJava tricky startWith(Observable)

2
votes

The following code emits items from observable1 only after observable2 is completed. 

observable1.startWith(observable2)
           .subscribe()

I need to achieve another behavior

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

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

The second observable emits items only while first observable is empty and then items from first one are emited.

I could not find correct solution using only basic operators, what is correct RxJava 2 implementation of custom operator startWithDefault should look like?

P.S.

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

is not correct solution because of race in case of immediate emit from observable1

1
javaandroidrx-javareactive-programmingrx-java2

1 Answers

4
votes

The direction was good, but you need publish(Function) to share observable1's signals and concatEager to not lose elements from it when the switch appens:

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