问题标签 [rx-scala]

无法理解为什么以下rxscala代码无法按预期工作：

我按 分组projectEvents并projectName希望打印每个项目的项目。但是当我运行这段代码时，它只打印：

没有######### event in project印刷品。

我不明白为什么，我错过了什么吗？

我在使用 RxJava 背压时遇到问题。基本上，我有一个生产者生产的项目比消费者可以处理的要多，并且希望有一些缓冲队列来只处理我可以处理的项目，并在我完成其中一些项目时请求，就像在这个例子中一样：

我希望得到像这样的输出

但相反，我得到

有没有办法在一个项目中同时使用RxJava和RxScala？

我们有一个使用JavaObservable( RxJava ) 用 Ja​​va 编写的模块。然后我们有一个 Scala 模块，它应该使用 Java 模块，但它是用 Scala 编写的。

有没有方便的方法将它们转换成另一个？

给定基于前一个和传入事件无限生成新值的计算，也可能失败，这是获得两个Observables 的最佳方法：

• 可能是无限的，对于成功的转换，
• 用于错误报告？

细节

我正在尝试对由 ScalaFX GUI 中的事件更改的状态进行建模，并且第一次尝试使用 RXScala。简化事情：

这个想法是创建一个Observable[State]给定的初始状态并根据事件更改为下一个状态。很简单：

现在问题其实是计算下一个状态涉及到I/O，可能会失败，所以computeNext要返回Try[State]

现在很明显，我不能再使用scan了。通常我想要两个Observables：

有一个规则，如果Try[State]是失败，observableState就不会发出任何东西，也不会报告错误。下一个状态基于上一次成功的计算。

解决这种情况的最优雅的方法是什么？

我想Observable从列表的结果中实时生成Futures.

在最简单的情况下，假设我有一个我正在运行的期货列表Future.sequence，并且我只是在监视它们的进度，并在Observable每次完成时告诉我。我基本上是这样做的：

这在我的测试环境中运行良好。但我刚刚读到onNext不应该从不同的线程调用，至少没有注意没有重叠的调用。解决此问题的推荐方法是什么？似乎许多现实世界的使用Observables都需要onNext从这样的异步代码中调用，但我在文档中找不到类似的示例。

我尝试使用循环中发生的多重Observer订阅。Observable它onNext似乎不适用于每个观察者。

仅针对第一个观察者的响应

为什么第二个收不到订阅？谢谢

我使用 RxScala/Java 构造一个 Observable 来发出带有循环的套接字连接，结果是每个连接都作为一个事件。像这样，

之后，连接就可以开始阅读了。我需要订阅scoket连接事件并将其作为一个新的 Observable。
ConnectedSocket可以使用stratReading，返回Observable[CompletedProto]，创建一个 Observable 作为连接。它发出带有接收信息的事件。我想让这个语句返回Observable[CompletedProto]。如何构造它？

如果需要详细信息，我真的愿意更新它。

原始 observable ------a--------b--------c----------d-------->....
映射observable -----A--------B(finish)
简单代码如下：

如何在B此指定条件下完成映射的 observable？

takeUntil 方法
takeUntil的更新似乎是一个标准答案，但我的编辑器显示此方法Observable[Any]作为参数。请参阅定义，
def takeUntil(that: Observable[Any]): Observable[T]

如果我使用以下代码

发生编译错误

我们使用相同的包吗？我的 sbt 依赖是 "com.netflix.rxjava" % "rxjava-scala" % "0.20.7"

我是 Rx 的新手，我似乎可以观察到缓冲区事件数据。使用 cold observable或replay使数据流即使在很长一段时间后也会发送给后来的观察者。但是什么时候会被垃圾呢？此外，使用 Rx 时应该注意什么内存？

我尝试

输出错误

如何在中使用自定义调度程序RxScala？