RxJava:结合观察到的冷热等待彼此 [英] RxJava: Combining hot and cold observable to wait for each other

问题描述

我的观测值是这样定义的

I have my observables defined like this

    val initLoading = Observable.fromCallable { println("${System.currentTimeMillis()}") }
            .subscribeOn(Schedulers.computation())
            .delay(WAIT_TIME, TimeUnit.SECONDS)
            .map { "loading ${System.currentTimeMillis()}" }
            .observeOn(AndroidSchedulers.mainThread())

    val click = RxView.clicks(button).map { "click ${System.currentTimeMillis()}" }
    initLoading.concatWith(click)
            .subscribeBy(
                    onNext = { println("result $it") },
                    onError = { throw it }
            )

initialLoading开始以Activity的onCreate方法运行.单击按钮执行click.我有两种情况，第一种有效，第二种无效.

initialLoading starts running at Activity's onCreate method. click is executed on button click. I have two cases and first is working, second isn't.

案例1

活动开始，并且在秒后秒后单击按钮.输出:

activity starts and button is clicked after WAIT_TIME seconds. Output:

   01-23 13:08:07.170  I/System.out: 1516698487170
   01-23 13:08:17.174  I/System.out: result loading 1516698497172
   01-23 13:08:29.258  I/System.out: result click 1516698509258

案例2

活动开始并在 WAIT_TIME结束之前单击按钮.输出

activity starts and button is clicked before WAIT_TIME period is over. Output

   01-23 13:09:07.392 I/System.out: 1516698547392
   01-23 13:09:17.398 I/System.out: result loading 1516698557395

因此，问题在于单击事件丢失了.我希望click事件等待加载，然后继续工作.简而言之，情况2的输出应与情况1相同.

so, the problem is that the click event is lost. I want the click event to wait for the loading, and then continue working. in short, case 2 output should be the same as case 1.

我如何使用rx运算符来实现这一目的.我尝试了merge，但它只是将两者结合在一起，而click事件并不等待加载.

How can i chive this using rx operators. I tried merge but it just combines both and click event doesn't wait for loading.

我也尝试过reply, cache, publish, share，但是无法按我的要求正确地组合它们.

I also tried reply, cache, publish, share but couldn't get the right combination of them to work as I want.

推荐答案

concatWith运算符非常适合您的用例，但是第二个可观察符应在创建后立即开始存储点击事件，以便可以存储已存储的事件.订阅了observable时发出(在initLoading完成时发生).这可以通过用replay()和connect()修改您的click观察值来实现.

The concatWith operator is good for your use case but the second observable should start to store click events immediately after it is created so that the stored events can be emitted when the observable is subscribed to (which happens when initLoading completes). This can be achieved by modifying your click observable with replay() and connect().

val replayedClicks = click.replay();
replayedClicks.connect(); // The original click observable is subscribed to at this point

现在您可以在concatWith中使用replayedClicks，并且在initLoading完成后将重播其存储的事件:

Now you can use replayedClicks in the concatWith, and its stored events will be replayed after initLoading finishes:

initLoading.concatWith(replayedClicks)
        .subscribeBy(
                onNext = { println("result $it") },
                onError = { throw it }
        )

这篇关于RxJava:结合观察到的冷热等待彼此的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！