问题标签 [rx-scala]

您好我想一起使用 rxscala 和 scalafx 所以我这样做是为了重新打包鼠标事件：

问题是在Subscription { ??? }块内我想删除注册的事件过滤器，filterEvent()但看着它我发现它filterEvent()没有保留对事件过滤器的引用，所以我不能使用stage.removeEventFilter(). 我也没有看到removeFilterEvent()等价物，那么如何使用 remove 注册事件过滤器filterEvent()？

我想一种解决方法是不使用filterEvent()，而是使用 javafx addEventFilter()，但语法更糟。

我的 oncreate 中有一个 Observable，我想在我的主线程上得到结果。但由于 AndroidScheduler.mainThread 仅适用于 Java Observable，因此我无法编译它。这是代码

这里的 'a' 是一个 rx.scala.lang Observable。这是我得到的编译时错误

请帮助解决这个问题。

所以我对 Scala 和 RX 都很陌生。最了解并实际编写此代码的人刚刚离开，我不确定发生了什么。这个结构遍布他的代码，我不太清楚它在做什么：

我主要得到do some stuff, 以及给订户的电话。我不明白的是，它subscriber来自哪里？是否subscriber => {实例化订阅者？是什么Observable { subscriber => { ... } }意思？

我想在类中添加一个方法Observable。到目前为止我所做的是：

但我得到：

怎么了？

我想知道是否有任何等效的 Scala fold ifEmpty 函数存在于集合和选项中：

这个函数很强大，因为它可以在不同的 monad 类型之间进行转换——例如，下面将 Option[Person] 转换为 Observable[Person]：

我目前的问题是我想要在 Reactive Extensions 领域有类似的东西。我的实际场景更多地需要“如果为空则切换”类型的功能 - 理想情况下：

所以这里的想法是，如果源 observable 完成并且什么都不发出，那么“切换”到另一个 observable。上面建议的一个实际示例 - 尝试从缓存中获取某些内容，如果没有返回任何内容，则从数据库中获取它。

我目前实现上述目标的解决方法如下：

换句话说 - 将结果包装在 Option 中（以便可以将空结果作为 None 值发出），然后将 flatMap 与模式匹配。

编辑：如果不存在，我将使用以下几个扩展函数：

有了以上内容，我现在可以做到：

我正在尝试递归定义一个可观察对象，它要么从主题发出项目，要么如果经过一定时间，则为默认值，在这种情况下，我使用计时器的默认值零。我正在使用 RxScala 并从以下代码开始：

这似乎应该工作，但输出令人困惑。每隔一个零序列包含两个而不是预期的四个。发出两个零，剩下的三秒过去了，但没有输出。

假设我们有这样的东西：

使用 rx-java/scala 的惯用解决方案是什么？

更新：再澄清一点，如果从代码中看不出来的话。

令ts _n和te _n分别为作业开始和结束的时间戳doSomething()。

然后下一个作业应该安排在ts _n+1 = max (te _n , ts _n + 1 秒)。

我正在使用 Vert.x 2.x ( http://vertx.io )，它广泛使用异步回调。由于典型的嵌套/回调地狱问题，这些很快就会变得笨拙。

我已经考虑过 Scala Futures/Promises（我认为这将是事实上的方法）和 Reactive Extensions (RxScala)。

从我的测试中，我发现了一些有趣的性能结果。

我的测试非常基础，我只是向 Vert.x verticle 发出一堆 HTTP 请求（通过 weighttp），它在 Vert.x 事件总线上进行异步调用，并处理一个响应，然后在 HTTP 200 中返回回复。

我发现如下（这里的性能是以每秒 HTTP 请求数来衡量的）：

• 异步回调性能 = 68,305 rps
• 接收性能 = 64,656 rps
• 未来/承诺性能 = 61,376 rps

测试条件为：

• Mac Pro OS X 优胜美地 10.10.2
• 甲骨文 JVM 1.8U25
• weighttp 版本 0.3
• 垂直 x 2.1.5
• 斯卡拉 2.10.4
• RxScala 0.23.0
• 4 个 Web 服务垂直实例
• 4 个后端服务 Verticle 实例

测试命令是

上面的数字是五次测试运行的平均值，最好和最差的结果。请注意，结果与呈现的平均值非常一致（不超过几百 rps 偏差）。

是否有任何已知原因导致上述情况发生 - 即每秒纯请求中的 Rx > Futures？

在我看来，响应式扩展是优越的，因为它们可以做更多的事情，但考虑到异步回调的标准方法通常似乎走上了 Futures/Promises 的轨道，我对性能下降感到惊讶。

编辑：这是 Web 服务垂直

编辑：这是后端垂直

我有需要解析字符串的应用程序，我使用它 RxScala。我的代码：

我有问题，因为我添加行的速度更快，然后我可以处理它们。

我想创建缓冲区，例如 200 行，如果缓冲区已满，则在缓冲区未满时忽略新记录，例如

RxScala 有办法做到这一点吗？

我正在尝试编写一个简单的 RxScala 程序：

当我运行这个程序时，我没有看到任何打印出来的东西。我怀疑这是因为产生数字的线程Observable.interval死了。我注意到waitFor(o)在RxScalaDemo中有一个调用，但我不知道它是从哪里导入的。

如何让这个程序永远运行，打印数字序列？