如何在等待另一个 Flowable 发出时发出一个项目

Question

我正在使用 RxJava3，我有以下代码设置，我想在第一个和第二个 flowable 之间的中间发出一个项目。有办法吗？

firstFlowable.firstElement()
//I want to emit an item here
.flatMap { secondFlowable.firstElement() }

我想这样做的原因是因为 firstFlowable 初始化后还有很长一段时间 secondFlowable 初始化，我想用消息通知 UI那里的操作已经开始，我正在等待来自 secondFlowable.

的数据计算

我尝试使用 startWithItem，但这会在开始时初始化我的整个链，但我只想在 firstFlowable 产生第一个值后才发出。

Answer 1

也许您可以将 concatWith() 与 take() 和 skip() 一起用于 firstFlowable。

public static void main(String[] args) {
   Flowable<String> firstFlowable = Flowable.just("1st", "2nd", "3rd", "4th", "5th");
   Flowable<String> middleFlowable = Flowable.just("between");
   Flowable<String> secondFlowable = Flowable.just("A", "B", "C", "D");

   firstFlowable.take(1)
                .concatWith(middleFlowable)
                .concatWith(firstFlowable.skip(1))
                .concatWith(secondFlowable)
                .subscribe(System.out::println);

   Flowable.timer(10, SECONDS).blockingSubscribe();  // Just to block the main thread for a while
}

这给出了类似的东西：

1st
between
2nd
3rd
4th
5th
A
B
C
D

Answer 2

您可以使用合并来注入一个值，然后根据值类型进行操作。如果 secondFlowable 立即完成，您可能希望避免通过 takeUntil.

显示字符串

firstFlowable.firstElement()
.flatMap(v ->
    Maybe.<Object>merge(
        secondFlowable.firstElement(),
        Maybe.just("Second in progress")
    )
)
.takeUntil(v -> !(v instanceof String))
.observeOn(mainThread())
.subscribe(v -> {
   if (v instanceof String) {
       // display message here
   } else {
       // cast and display results of second
   }
});

如何在等待另一个 Flowable 发出时发出一个项目

How to emit an item while waiting another Flowable to emit

java

reactive-programming

observable

rx-java3