问题标签 [rx-scala]

我原以为这段代码会打印

但它不打印最初的“订阅”。

我想从这样的sObservable数组中创建一个：Observable

该程序无法编译，因为Observable有几个重载方法都被调用merge：

然后我想在另一个 Java 类的帮助下消除重载：

Scala 代码变为（此处仅列出相关部分）：

该程序仍然无法编译：

我有三个问题：

1. 如何merge像第一种情况一样用纯 Scala 调用方法？
2. 为什么第二个程序无法编译？
3. 如何在Scala中调用merge上述类中的方法？RxUtils

我正在尝试自己切片可观察流，例如：

输出为：

sourceboundaries可能在到达之前在线迭代，result所以它只创建边界和结果缓冲区，但没有什么可填充的。

我的方法是使用publish/ connect：

这可以产生输出：

现在我只需connect要从外部世界隐藏connect它，当它result被订阅时（我在一个类中这样做，我不想暴露它）。就像是：

但是现在，doOnSubscribe动作永远不会被调用，所以发布source永远不会连接......

怎么了？

我怎样才能拥有一个单一订阅者ReplaySubject：

1. 缓冲所有接收到的事件，onNext()直到有人订阅它，
2. 一旦有人订阅它，所有缓冲的事件都会转发给订阅者，从ReplaySubject缓冲区中删除它们（出于性能原因），后续事件仅转发给订阅者（不再缓冲），
3. 如果多个观察者订阅它，则抛出异常，
4. 给主题的所有缓冲事件都根据每个事件的生成时间排序，并在订阅期间按该顺序转发给订阅者？

另外，这有意义吗？我认为这个主题会有很好的用例，例如它从文件系统转发事件的情况......

是否有可能有一种BehaviorSubject暂停和恢复开关？像PausableBehaviorSubject.pause()和PausableBehaviorSubject.resume()？那怎么可能呢？

这个想法是，当暂停时，主题不会转发通过 接收到的事件PausableBehaviorSubject.onNext()，而是只存储最后一个。在恢复时（或在订阅时，如果未暂停），PausableBehaviorSubject将转发存储的最后一个事件（如果有）并将后续事件转发给其订阅者。

我正在查看在给定时间间隔创建的 RxScala 可观察对象：

如果我把它放在工作表中，我会得到这个结果：

这让我很困惑： 的元素periodic实际上包含什么？

它们是否包含一些索引？它们是否包含创建它们的时间间隔？

如果我们有可观察的：

如何构造新的 observable：

也许问题很短，但我真的找不到如何实现。谢谢

谢谢大家，我找到了一种方法并考虑删除var

我根本不喜欢 var，再次感谢！

最后 我得到了一个轻松的方式，希望可以帮助别人。

其他答案也很有帮助~

假设我们有 Ints 的源 Observable：

我想创建另一个 Observable，产生的值与源中首次出现的值的差异大于 10：

这里我使用 var start来保存源Observable的第一个值。

有没有办法简化这段代码？我不喜欢这种为 var 赋值的方法

我需要编写一些 RxScala 代码，Observable从 Scala Swing 中的文本字段创建一个。在查看了 Github 之后，我编写了这段代码，它似乎可以工作：

Wherefield.text给出了该字段中文本的当前状态。

但是我不明白这一行：

这有什么Subscription用？它不带任何参数，所以我假设它不能用于取消订阅我创建的 observable，因为它甚至不引用这个 observable。

那么有什么Subscription()好处呢？

为什么它不带任何参数？

我有一个名为responsestype的变量，Observable[Try[Int]]它发出：

Success(n), 其中n是自然数或 Failure(Exception)

我正在总结这个 observable 的值，如下所示：

打印语句显示了这一点：

紧接着，我试图从 observable 中得到总和，如下所示：

但是，这里total永远不会更新，并且 print 语句永远不会显示任何内容。

为什么会这样？

为什么事件不再传递？