RxJava:当其中一个流不发出任何内容时如何处理 combineLatest() [英] RxJava: how to handle combineLatest() when one of the streams emits nothing

问题描述

我使用 combineLatest() 来组合 3 个 observables 流.所有这些组合在一起，以便同时显示 UI 中的所有数据.现在，有一种场景，其中一个 observable 不会发出任何内容，因为获取的数据可能为 null.

I use combineLatest() to combine 3 streams of observables. All these are combined so that all data in the UI is shown at the same time. Now, there is a scenario in which one of the observables won't emit anything, since the data that gets fetched, can be null.

是否有 RxJava 操作符让订阅者知道不会因为空数据而发出任何消息?

Is there a RxJava operator to let the subscriber know that there won't be any emits because of null data?

编辑

private fun retrieveData() {
    Observable.combineLatest(getCurrentUser.execute(), getLatestGoal.execute(), getLatestLog.execute(),
            Function3<User, Goal, Log, PersonalViewModel> { user, goal, log -> mapToViewModel(user, goal, log) })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe { /*todo: animation*/ }
            .doOnNext { view.setViewModel(it) }
            .doOnComplete { view.stopLoading() }
            .doOnError { /*todo: error message*/ }
            .subscribe()
}

第三个流:当用户有 nog 日志时，getLatestLog.execute() 不发出任何信息.当此流不发出时，整个视图将不可见.

The third stream: getLatestLog.execute() emits nothing when the user has nog log. When this stream doesn't emit, the whole view will not be visible.

数据取自 FireBase 实时数据库.ChildEventListener 有一个如下所示的方法:

The data is fetched from FireBase Realtime database. The ChildEventListener has a method that looks like this:

override fun onChildAdded(dataSnapshot: DataSnapshot?, p1: String?) {
                val log = dataSnapshot?.getValue(Log::class.java)
                log?.let { subscriber.onNext(it) }
                subscriber.onComplete()
                firebaseDatabase.reference.removeEventListener(this)
            }

推荐答案

您可以使用 public final Single first(T defaultItem) 方法.所以代码可能看起来像这样

You can use public final Single first(T defaultItem) method. So the code may look like this

getLatestLog.execute()
.first(someDefaultNonNullLog)
.toObservable()

这篇关于RxJava:当其中一个流不发出任何内容时如何处理 combineLatest()的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！