RxJava：当其中一个流什么都不发出时，如何处理组合Latest（）

Question

我使用combineLatest（）组合了3个可观测数据流。所有这些都结合在一起，以便同时显示UI中的所有数据。现在，有一个场景，其中一个可观察对象不会发出任何东西，因为获取的数据可以为null。

是否有RxJava操作符让订阅者知道由于空数据，不会有任何发射？

编辑

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()
}

第三个流：getLatestLog。当用户有nog日志时，execute（）不会发出任何消息。当此流不发射时，整个视图将不可见。

数据从FireBase实时数据库中获取。ChildEventListener的方法如下所示：

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)
            }

Answer 1

您可以使用public final Single first（T defaultItem）方法。所以代码可能如下所示

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

Answer 2

如果您有Java8或一些Optionals，您可以使用此构造：

  @Test
  void name() {
    TestScheduler scheduler = new TestScheduler();

    Observable<Optional<Integer>> o1$ =
        Observable.just(Optional.ofNullable(4)).mergeWith(Observable.never());
    Observable<Optional<Integer>> o2$ =
        Observable.just(Optional.ofNullable(2)).mergeWith(Observable.never());

    Observable<Optional<Integer>> o3$ =
        Observable.<Optional<Integer>>never()
            .timeout(1000, TimeUnit.MILLISECONDS, scheduler)
            .onErrorResumeNext(
                throwable -> {
                  return Observable.<Optional<Integer>>never()
                      .mergeWith(Observable.just(Optional.empty()));
                });

    Observable<Tuple3<Optional<Integer>, Optional<Integer>, Optional<Integer>>> result =
        Observable.combineLatest(
                o1$,
                o2$,
                o3$,
                (integer, integer2, integer3) -> Tuple.of(integer, integer2, integer3))
            .filter(t -> t._1.isPresent() && t._2.isPresent() && t._3.isPresent());

    TestObserver<Tuple3<Optional<Integer>, Optional<Integer>, Optional<Integer>>> test =
        result.test();

    scheduler.advanceTimeTo(10000, TimeUnit.SECONDS);

    test.assertNotComplete().assertNoErrors().assertNoValues();
  }

如您所知，不允许通过可观察管道发出空值。因此，我们需要一些其他的构造来表示null。在Java8中有一个称为Optional的构造（vavr称之为Option-

在这个例子中，o3$-可观察不会发出任何东西。它也可能出错，也许这更像你的情况。我们将捕获错误（在这种情况下：超时异常）并返回一个带有Optional.empty.的可观察

在组合回调中，我们组合所有三个值。在后面的步骤中，我们过滤掉所有元组，这些元组都有有效值（可选值为Value）。

只有当三个值都已发出一个值时，才会发出一个值。

当不能使用可选类时，还可以定义无效对象，如以下示例所示：

class So51217041 {
  private static Integer INVALID_VALUE = 42;

  @Test
  void name() {
    Observable<Integer> o1$ = Observable.just(4).mergeWith(Observable.never());
    Observable<Integer> o2$ = Observable.just(2).mergeWith(Observable.never());

    Observable<Integer> o3$ =
        Observable.<Integer>never()
            .onErrorResumeNext(
                throwable -> {
                  return Observable.<Integer>never().mergeWith(Observable.just(INVALID_VALUE));
                });

    Observable<Tuple3<Integer, Integer, Integer>> result =
        Observable.combineLatest(
                o1$,
                o2$,
                o3$,
                (integer, integer2, integer3) -> Tuple.of(integer, integer2, integer3))
            .filter(t -> t._3 != INVALID_VALUE); // yeah I know, I want to compare reference, not the content

    TestObserver<Tuple3<Integer, Integer, Integer>> test = result.test();

    test.assertNotComplete().assertNoErrors().assertNoValues();
  }
}

此外，当您希望流以INVALID或NULL开头时，组合最新至少会发出一个值，您可以使用可观察的#start To（INVALID）oder可观察的#start To（Optional.empty（））。这将保证可观察的至少会发出一个值。