如何动态更新 RX Observable？

Question

（使用 RxKotlin 和 RxJava，但为了简单起见使用元代码）

许多Reactive Extensions 指南首先从现有数据创建Observable。从The introduction to Reactive Programming you've been missing，它是从单个字符串创建的

var soureStream= Rx.Observable.just('https://api.github.com/users');

类似地，来自RxKotlin 的首页，来自填充列表

val list = listOf(1,2,3,4,5)
list.toObservable()     

现在考虑一个产生outStream的简单过滤器，

var outStream = sourceStream.filter({x > 3})

在这两个指南中，源事件都被声明为先验。这意味着事件的时间线有某种形式

source: ----1,2,3,4,5-------
out:    --------------4,5---

如何修改sourceStream 使其更像一个管道？换句话说，sourceStream 创建期间没有可用的输入数据？当源事件可用时，立即由 out 处理：

source: ---1--2--3-4---5-------
out:    ------------4---5-------

我希望找到Observable.add() 进行动态更新

var sourceStream = Observable.empty()
var outStream = sourceStream.filter({x>3})

//print each element as its added 
sourceStream .subscribe({println(it)})
outStream.subscribe({println(it)})

for i in range(5):
    sourceStream.add(i)

这可能吗？

Answer 1

我是新手，但是没有主题我怎么能解决我的问题呢？如果我是 测试一个应用程序，我希望它每 5 次“弹出”一次更新 秒，除了这个发布订阅我还能怎么做 商业？有人可以发布这个问题的答案吗？ 涉及订阅者？

如果你想每五秒弹出一次更新，那么用interval 操作符创建一个 Observable，不要使用 Subject。构建 Observable 有几十种不同的运算符，因此您很少需要主题。

也就是说，有时您确实需要一个，它们在测试代码时非常方便。我在单元测试中广泛使用它们。

To Use Subject Or Not To Use Subject? 是关于主题主题的优秀文章。