scala - 如何手动更新 observable？

Question

我是 reactivex 和rxscala的新手，可以创建Observable这样的：

val observable = Observable[String] { subscriber => 
    subscriber.onNext("something")
}

我可以将新字符串subscriber放入Observable.apply.

observable外面可以更新吗？我的意思是，有没有类似的方法putNext：

observable.putNext("another string")

让我把新东西放到现有的 observable 上？

score 2 · Accepted Answer

如果要手动控制 Observable，则需要使用 Subject。

Subject 是一种桥接器或代理，在 ReactiveX 的某些实现中可用，它既充当观察者又充当 Observable。因为它是一个观察者，所以它可以订阅一个或多个 Observable，而且因为它是一个 Observable，它可以通过重新发射它观察到的项目来传递它们，它也可以发射新的项目。

您可以订阅一个主题，但您也可以将新元素传递给它，实际上是手动控制它，这就是您所要求的。

例子：

val subject = PublishSubject[String]()
subject.onNext("one")
subject.onNext("two")
subject.onNext("three")
subject.onCompleted()

1 回答 1