1

为了用 rxscala 测试我的反应程序,我需要构造这样一个Observable

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

哪一个

  1. 公布号码1
  2. 然后等待1s
  3. 公布号码4
  4. 然后等待3s
  5. 公布号码2
  6. 然后等待2s

我有一个丑陋的解决方案,使用Threadand 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()

有没有更好的解决方案?

4

1 回答 1

1

您可以将多个具有延迟的 Observable 连接在一起,最终的 Observable 存在并且empty具有订阅延迟。

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

附带说明一下,如果您正在测试,您应该使用TestScheduler可以作为第二个参数传递给delay.

于 2015-11-13T16:29:06.033 回答