在创建 rx Observable 后添加元素

Question

我如何实现这样的情景：在创建Observable之后添加元素，是否能够做到？在观察者模式中，我只需触发事件等操作。你有什么想法吗？

import rx.lang.scala._

val target  = Observable(1,2,3,4)

val subscription1 = target subscribe(println(_))
val subscription2 = target subscribe(println(_))


def addToObservable(toAdd: Int, target: Observable[Int]): Observable[Int] = {
  target/*.addElementAndNotifyObservers(toAdd)*/
}

addToObservable(4, target) //should print 4 on all subscriptions
addToObservable(6, target) //should print 6 on all subscriptions

- Eddie Jamsession

Answer 1

你不能这样做-至少不是对你创建的可观测对象。你需要使用一个Subject，通过它你可以发射值。Subject基本上是一个同时具有Observable和Observer的对象。

例如：

import rx.lang.scala._
import rx.lang.scala.subjects._

val subject = ReplaySubject[Int]()             
val initial = Observable(1,2,3,4)     
val target = initial ++ subject   // concat the observables

val subscription1 = target subscribe(println(_))
val subscription2 = target subscribe(println(_))

subject.onNext(4)    // emit '4'
subject.onNext(6)    // emit '6'   

- PeterParameter