RxJava2 同时执行 Completable 和 Observable 并处理两者的结果答案

【问题标题】：RxJava2 execute Completable and Observable simultaneously and handle result from bothRxJava2 同时执行 Completable 和 Observable 并处理两者的结果
【发布时间】：2020-01-06 13:43:23
【问题描述】：

我有一个 Completable 和 Observable 发射项目。我必须把它们结合起来。某种 Observable.combineLatest 但具有 Completable 和 Observable。所以逻辑是：如果 Observable 先发出 item，当 Completable 没有完成时，它应该等待 Completable（而不是忽略），如果 Completable 完成了，则应该处理来自 Observable 的项目。

我尝试将 Completable 转换为 Observable，但 Observable 不会发出 onNext，因此 Observable.zip、Observable.combineLatest 等不起作用。

适合我的代码：

myCompletable.andThen(myObservable).subscribe()

但我不会异步调用 completable 和 observable 。我只是等到 Completable 完成。是否可以异步运行 Completable 和 Observable 并在 Complitable 完成时处理 observable 项？

【问题讨论】：

如果Observable 在Completable 完成之前发出多个项目会发生什么？您需要保留所有之前的物品吗？
不，我不知道。对于这种情况，我不需要 flowable。

标签： android kotlin observable rx-java2

【解决方案1】：

您可以将myCompletable 转换为Observable 并使用.defaultIfEmpty 来保证至少会发出一项。检查这个例子：

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

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

请注意，只有myObservable 发出的最后一个项目将使用此方法保留，如评论中所建议的那样。

【讨论】：

在这种情况下，我得到了默认值。我认为这是因为 Completable（在我的情况下是 Hot Completable - CompletableSubject）没有发出项目，因为 onComplete() 调用了。