0

我试图replay在 RxScala 中理解。我创建了一个像这样的可观察对象：

lazy val toyObservable : Observable[Int] = {
    val coldObservable : Observable[Int] = intPerSecond
    val hotObservable : ConnectableObservable[Int] = coldObservable.publish
    val cachedObservable = hotObservable //.replay(3)   //<<<<<<<<< ODD THING 
    cachedObservable.connect
    cachedObservable
}

其中intPerSecond每秒发射一个整数，从 0 开始。第一个订阅的观察者确实每秒看到一个整数。如果第二个观察者在 t=6 秒时加入，那么从那时起，他们都会以一秒的间隔看到匹配的流 6...7...8...9...。正如预期的那样。

现在，如果我添加，.replay(3)我希望当第二个观察者加入时，他会看到 3456...7...8...9，即他会立即从缓存中获取 3 个整数，然后接收它们以每秒一个的速度生产。但相反，现在两个观察者都看不到任何东西。我有语法错误吗？

4

1 回答 1

1

你忘记打电话了hotObservable.connect。以下代码完全输出您想要的内容：

import rx.lang.scala._
import rx.lang.scala.observables._
import scala.concurrent.duration._
val intPerSecond = Observable.interval(1.seconds).map(_.toInt)
val coldObservable : Observable[Int] = intPerSecond
val hotObservable : ConnectableObservable[Int] = coldObservable.publish
val cachedObservable = hotObservable.replay(3)
cachedObservable.connect
hotObservable.connect
cachedObservable.foreach(i => println(s"1: $i"))
Thread.sleep(6000)
cachedObservable.foreach(i => println(s"2: $i"))
于 2015-06-05T03:47:03.123 回答