问题标签 [rx-scala]

我正在使用 rx-scala，它是 rx-java 的子项目。我将使用 Scala 语法，希望每个人都能理解。

我遇到了奇怪的行为，我不知道这是错误还是代表我滥用 rx 运算符。

问题陈述

我有一个ox: Observable[X]和一个触发器 observable tr: Observable[()]。我想要一个 observable oy，它是oxusing function的转换f: Function[X,Y]，但只有在触发时才会发生，因为f它可能很昂贵。

如果 的最后一个值没有转换后的值ox，则oy应该是null。

一些备注：

• ox很热，因为它是 UI 事件的结果。
• oxprintln正如我通过调试检查的那样，行为正确（值和时间） 。
• oy在正确的时间开火；它只是使用过时的值ox，当它是一个非空值时。

当前代码

上面代码的问题是：它最初可以工作，但一段时间后，当触发 时，tr它oy会应用于. 不改变时，如果我反复触发，结果会更新并最终赶上。foxoxtr

如果我删除mergeto not reset to null，那么一切正常（可能是因为效果看起来不确定）。

问题

我提供的代码有问题。

1. 我想知道我是否做错了什么。
2. 我欢迎实现我需要的其他方法。

对于 Jave 人

• 泛型/类型注释：ox: Observable[X]意味着Observable<X> ox
• lambdas：x => null意味着x -> null

具有以下内容build.sbt：

这个代码：

当我运行它时，所有打印的是Ready?; 我看完全没有n = ...。

我使用sbt run; 它是使用 Scala 2.6.11 和 RxScala 0.24.1 以及 sbt 0.13 构建的。有任何想法吗？

我刚刚开始使用 RxScala 并遇到了一些示例。我知道有一个 Observable 合同，其中 Observer 特征定义如下：

我可以使用 Observer 订阅 Observable 事件流。因此，观察者是为观察者特征中的 3 种方法提供实现的任何东西。

现在，在以下示例中：

我正在创建一个 Long 类型的 Observable 事件，并且在 subscribe 调用中，我只覆盖了 onNext 。onError 和 onCompleted 呢？

编辑：如果我不提供 onError 和 onCompleted 处理程序会怎样？我会违反任何线程语义吗？

来自 Scala 中 Rx Observable 的 API 文档：

http://reactivex.io/rxscala/scaladoc/#rx.lang.scala.Observable

有两种应用方法，一种接受订阅者，另一种接受观察者。除了它们采用不同的类型来订阅 Observable 之外，这两种 apply 方法之间有什么区别？

TableEntries 我从 RxScala 开始，我正在尝试提出一种轮询机制，以检查每个间隔（比如 20 秒）的数据库，以检查表中的某些行是否有任何变化。

我需要从一个会发出 List[MyTableEntries] 的 Observable 开始。所以我从以下开始：

在我传递给函数的观察者中，我实现了 onNext、onError 和 onCompleted！但是有几个问题：

1. 如果我的数据库响应时间超过 30 秒会发生什么
2. 如果我的数据库完全关闭会怎样？

这是我所做的有效方法吗？建议？

我试图replay在 RxScala 中理解。我创建了一个像这样的可观察对象：

其中intPerSecond每秒发射一个整数，从 0 开始。第一个订阅的观察者确实每秒看到一个整数。如果第二个观察者在 t=6 秒时加入，那么从那时起，他们都会以一秒的间隔看到匹配的流 6...7...8...9...。正如预期的那样。

现在，如果我添加，.replay(3)我希望当第二个观察者加入时，他会看到 3456...7...8...9，即他会立即从缓存中获取 3 个整数，然后接收它们以每秒一个的速度生产。但相反，现在两个观察者都看不到任何东西。我有语法错误吗？

我最近刚刚开始使用 Scala 进行编码，我想开始使用 RxScala，但我无法弄清楚如何将它导入 Intellij。有谁知道如何做到这一点，我已经花了最后几个小时试图弄清楚这一点，但我并不比开始时更接近？如果这有什么影响，我也在使用 Ubuntu。

使用 Play Frameworks Enumerators、Iteratees 和 Enumeratees 与使用 RxScalas Observables、Subscriptions 等进行异步数据流相比如何？

在什么类型的场景下你会选择使用 RxScala，什么时候会选择 Play？

如果您有大数据流过您的流，那会影响您的决定吗？

假设我有一个类型为元素的事件流In：

还有一个用于将 type 对象转换为 typeIn对象的函数Out，但“在未来”：

此时我想observableIn根据我的功能转换我的元素futureInToFutureOut。也就是说，我想要一个类型为元素的事件流Out，与原始流的元素匹配，但通过 function 转换futureInToFutureOut。

我认为这应该有效：

这是正确的吗？有一个更好的方法吗？

我有一个通过 Socket 协议连接到服务的 Observable。与套接字的连接是通过客户端库发生的。我使用的客户端库有 java.util.Observer，我可以注册推送到其中的事件

我有两个我不明白的悬而未决的问题。

如何在我的订阅者中获得第 3 步的结果？

每次当我收到 MyEvent 时，订阅者如下所示，我看到正在创建一个新连接。最终，为每个传入事件运行第 1 步、第 2 步和第 3 步。