java - 如何构造具有自定义数字和不同延迟的 Observable？

Question

为了使用 rxscala 测试我的响应式(Reactive)程序，我需要构造这样一个 Observable:

val numberStream: Observable[Int] = Observable.???()

哪个

1. 发布数字1
2. 然后等待1s
3. 发布数字4
4. 然后等待3秒
5. 发布数字2
6. 然后等待2s

我有一个丑陋的解决方案，带有Thread和ReplaySubject:

val subject: Subject[Int] = ReplaySubject()
val numberStream: Observable = subject

new Thread(new Runnable {
    def run = {
        subject.onNext(1)
        Thread.sleep(1000)
        subject.onNext(4)
        Thread.sleep(3000)
        subject.onNext(2)
        Thread.sleep(2000)
    }
}).start()

有没有更好的解决办法？

Answer 1

您可以将多个带有延迟的 Observable 连接在一起，最终的 Observable 为空，并带有订阅延迟。

val numberStream = (
     Observable.just(1) ++ 
     Observable.just(4).delay(1.second) ++
     Observable.just(2).delay(3.second) ++
     Observable.empty.delaySubscription(2.second))

顺便说一句，如果您正在测试，则应该使用 TestScheduler，您可以将其作为第二个参数传递给 delay。