RxJava：结合冷热observable相互等待答案

【问题标题】：RxJava: Combining hot and cold observable to wait for each otherRxJava：结合冷热observable相互等待
【发布时间】：2018-01-23 09:35:22
【问题描述】：

我的 observables 是这样定义的

    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 在按钮单击时执行。我有两个案例，第一个有效，第二个无效。

案例1

活动开始并在 WAIT_TIME 秒后单击按钮。输出：

   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 期间结束。输出

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

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

我怎样才能使用 rx 操作符来实现它。我试过merge，但它只是结合了两者，点击事件不等待加载。

我也尝试了reply, cache, publish, share，但无法让它们的正确组合按我的意愿工作。

【问题讨论】：

标签： android rx-java rx-java2 rx-kotlin rx-kotlin2

【解决方案1】：

concatWith 运算符非常适合您的用例，但第二个 observable 应该在创建后立即开始存储点击事件，以便在订阅 observable 时发出存储的事件（当 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 }
        )

【讨论】：

非常好的组合和解决方案。谢谢

【解决方案2】：

尝试使用 combineLatest 而不是 concatWith。 “结果加载”总是一样的

【讨论】：