问题标签 [monix]

我正在使用一个库（特别是） ScalaJS，react在那里我遇到了一个有趣的情况，我认为这对于经验丰富的反应式程序员来说是非常常规的。我有一个ComponentwithState和一个 callback shouldComponentUpdate(State)。这里的基本思想是，如果回调被触发但State与上次没有变化render，则返回false。true否则，如果State更改很重要，也许会返回。

我正在使用一个库monix ，但它似乎与其他反应式库相同，所以我想这是一个与上下文无关的问题。

我想做类似的事情：有一些反映State自上次以来的增量的状态render。在每个 上render，清除缓冲区。或者，有一个renderedState将所有渲染状态作为序列反映的receivedState主题，一个反映所有接收到 State的更新的needsUpdate主题，以及一个反映最新是否与最新receivedState匹配的主题renderedState。不过，我在实际执行这些想法时遇到了麻烦。这是我被困在的地方：

这是我为其他回调所做的：

因此，当componentWillUpdate触发回调时，处理程序会onNext在channel（主题）上触发。

shouldComponentUpdate不过是不同的。它返回一个值，因此需要采用不同的结构。我很难考虑正确的调整。

总结一下：

• react在视图生命周期的不同阶段有回调，例如componentDidMount,componentDidUpdate等。
• 我以相同的方式处理除了一个阶段之外的所有阶段 - 回调的形状是State -> Callback<Void>所以我所要做的就是对每种类型的生命周期事件使用 a并在触发回调时Subject提交它。onNext
• State -> Boolean但是一种类型的事件具有或的形状State -> Callback<Boolean>。
• 我觉得我应该能够使用代表最后呈现/接收状态之间的增量的主题来对此进行建模。
• 但是，我不知道这如何适应反应式风格。

假设我有多个有序的迭代器。如果我想合并这些迭代器，同时[(1,3,4), (2,4,5)] -> [1,2,3,4,4,5]使用monix对它们进行全局排序（例如） ，我会怎么做？

我实际上使用的是 Scala，但这个问题对所有 Rx 和流框架都是通用的。

我的用例是我有一个生成的 observable（因此很冷），我希望多个消费者并行使用完全相同的值，但我希望它们具有显着不同的吞吐量。

我需要通过重放广播一个可观察的来完成，但我看到使用最大缓冲区大小重放的常见策略是在溢出时从缓冲区中删除元素（然后对于最慢的消费者丢失）而不是返回 -给生产者施压。如果您将所有广播的 observables 都视为热的，这是有道理的，但是，就我而言，我知道它实际上是冷的并且可以被背压。

在任何 JVM 反应流兼容框架中是否有某种方法可以实现这一点？

非常感谢！

Monix observables 有 apis.onErrorRestartIf(f: Throwable => Boolean)和.onErrorRestart(times: Int). 如何指定重试执行.onErrorRestartIf 的最大次数？

我有一个分页资源，我想用 Monix 递归地使用它。我想要一个 Observable，它会发出下载的元素并递归地使用页面。这是一个简单的例子。它当然行不通。它发出第一页，然后是第一页+第二页，然后是第一页+第二页+第三页。我希望它首先发射，然后是第二个，然后是第三个，依此类推。

有任何想法吗？

UPD 对不起，它似乎在工作，我只是有一个错误size + 5。所以看起来问题已经解决了，但是如果你发现我做错了什么，请告诉我。

我正在使用AKKA Http、Monix和Datastax Java 驱动程序为 Apache Cassandra构建一个 REST API ，我在尝试从 cassandra 获取一些项目时遇到了一些麻烦，等待查询完成并返回结果。

我能够轻松打印所有结果，但无法等待查询完成并返回所有项目。我的休息点只是返回一个空的项目数组，因为它不等待查询完成。

我有一个executeQuery方法：

• queryString: String表示 cassandra 查询
• page: Int对分页有用
• parameters: Any*表示参数，如果查询需要

并返回一个Observable[Row].

然后，为了执行这样的查询，检索它的结果，解析它们并将它们发回，我使用Monix Observable 和 Subscription。

假设我想通过一个名为的公共字段检索一些项目pid：

WhererowToItem只是将一行解析为Itemand Items: List[Item]。我正在查看Task，但我不太确定它在寻找什么。

编辑

使用@Alexandru Nedelcu 解决方案，我可以items在itemList将它们插入其中后立即打印所有内容，但该调用仍然得到空响应：{ "items" : [] }.

这是编辑后的代码：

如何等待结果全部解析并插入到项目中，然后将它们发回？

我已经用 monix 和 akka-streams 对 List[ClassA] 到 List[ClassB] 的映射进行了基准测试，但我不明白为什么它这么慢。

我尝试了不同的映射方式，这是 JMH 的结果：

这是代码：

当初始集合更大时，工作台结果会变得更糟。

我想知道我的错误是什么。

感谢您分享您的知识！

最好的祝福

我正在为我要完成的任务编写一个简单的示例。假设有一个tasks我想每 1 秒触发一次的任务列表 () 。

这可以通过 ascheduler或其他方式来完成。

现在这个流有两个消费者，但是

• C1应在完成所有任务时触发
• C2应该在所有任务的第 n 次完成时触发。（也可以每隔 n 秒）

这是一些示例代码。目前，它没有计划重复 - 因为我不知道Observable.repeat使用Scheduler.

我正在使用 monix 任务，我正在尝试捕获 Throwable，然后转换为自定义错误。我已将代码删除/更改为简单且相关。这是代码（代码片段之后的问题）：

我可以看到打印语句，但Task.now(Response(...没有返回预期的语句。相反，调用处理程序方法的方法会引发错误。我如何让它返回Task[Response]？

成功案例有效，失败案例无效。

编辑 #1：修复 scala 代码中的错误。

编辑#2 这就是我修复它的方式。

我在思考未来，忘记了lazy eval任务的性质。此外，我了解CancellableFuture在失败任务中该值是如何被丢弃的。

使用monix我试图通过构建 Observable[Node] 并使用广度优先算法来遍历图形。但是，我有一点递归问题。这是一个说明我的问题的片段：

第一次迭代后的可观察停止。谁能提示我为什么？