android - RxJava : Combining hot and cold observable to wait for each other

Question

我的观察对象是这样定义的

    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 在按钮点击时执行。我有两个案例，第一个是工作，第二个不是。

案例一

WAIT_TIME 秒后， Activity 开始并单击按钮。输出:

   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

Activity 开始并且按钮被点击 before WAIT_TIME 时间结束。输出

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

所以，问题是点击事件丢失了。我希望点击事件等待加载，然后继续工作。简而言之，案例 2 的输出应该与案例 1 相同。

我如何使用 rx 运算符来解决这个问题。我试过 merge 但它只是结合了两者并且点击事件不等待加载。

我也试过reply, cache, publish, share 但无法按照我的意愿正确组合它们。

Answer 1

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

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

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

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