RxJava2 execute Completable and Observable simultaneously and handle result from both

Question

I have one Completable and Observable that emits items. I have to combine them. Some kind of Observable.combineLatest but with Completable and Observable. So logic is: if Observable emit item first, when Completable is not complete, it should wait for Completable (not ignore), if Completable is complete, then items from Observable should be handled.

I tried convert Completable into Observable but that Observable does not emit onNext, so Observable.zip, Observable.combineLatest, etc does not worked.

Code that works for me:

myCompletable.andThen(myObservable).subscribe()

But I don't call completable and observable asynchronously. I just wait until completable completes.Is it possible to run Completable and Observable asynchronously and handle observable items when Complitable complites?

What should happen if Observable emits more than one item before the Completable completes? Do you need to keep all prior items? — Sanlok Lee

Sanlok Lee Sanlok Lee · Accepted Answer · 2019-09-04T17:24:22

You can convert myCompletable to an Observable and use .defaultIfEmpty to guarantee there will be at least one item emitted. Check this example:

val myObservable: Observable<Long> = ...
val myCompletable: Completable = ...

Observable.combineLatest(
    myObservable
    myCompletable.toObservable<Long>().defaultIfEmpty(0L),
    BiFunction<Long, Long, Long> { a, b -> a}
)
.subscribe()

Note that only the last item emitted by myObservable will be kept with this method, as suggested in the comment.

RxJava2 execute Completable and Observable simultaneously and handle result from both

1 Answers