问题标签 [fs2]

给定一个要被多次调用的函数A => IO[B]（又名Kleisli[IO, A, B]），并且有副作用，比如更新数据库，如何将这样的多次调用委托给一个流（我猜Pipe[IO, A, B]）（fs2，monix observable/iterant）？这样做的原因是能够在一个时间窗口内累积状态、批量调用等。

更具体地说，http4s 服务器需要一个Request => IO[Response]，所以我正在研究如何对流进行操作（为了上述好处），但最终为 http4s 提供了这样的功能。

我怀疑它在幕后需要一些相关 ID，我对此很好，我对如何从 FP 的角度安全和正确地进行操作更感兴趣。

最终，我期望的签名可能是这样的：

Pipe[IO, A, B] => (A => IO[B]), 这样对 Kleisli 的调用就会通过管道传输。

作为事后的想法，是否有可能背压？

我有如下事件流：

我想对在一分钟内收到的这些事件进行分组（即从每分钟的 0 秒到 59 秒）。这听起来很简单fs2

问题是分组功能不纯。它使用currentTimeMillis. 我可以按如下方式解决这个问题：

问题是我想避免使用带有元组的笨拙样板。还有其他解决方案吗？

或者，对于这种情况，使用不纯函数可能不是那么糟糕？

使用 fs2 (ver. 1.0.4) 和 cats-effect IO，我可以将 URL 流式传输到本地文件，

但是，此代码段在完成时不会返回有关该进程的任何信息。最重要的是，除了知道操作是成功还是失败之外，我还想知道如果操作成功，读取了多少字节。我不想检查新文件大小来获取此信息。另一方面，如果操作失败，我想知道导致失败的原因。

我尝试过attempt，但无法解决将原始字节写入新文件的后续步骤。请指教。谢谢

我想使用类似于take(n: Int)但在时间维度上的函数： consume(period: Duration. 因此，如果发生超时，我希望流终止。我知道我可以将流编译成类似的东西IO[List[T]]并取消它，但是我会丢失结果。实际上，我想将无尽的流转换为有限的流并保留结果。

更多关于更广泛的问题。我有来自消息传递代理的无穷无尽的事件流，但我也有轮换凭据来连接到代理。所以我想要的是消耗事件流一段时间，然后停止，获取新凭据，再次连接到创建新流的代理并将两个流连接为一个。

我正在尝试编译流，但不知何故Compiler不在范围内，需要什么上下文绑定才能将其纳入范围？

我想并行运行 N 个嵌套流/管道，并将每个元素仅发送到一个嵌套流。平衡允许我这样做，但我想将具有相同“键”的元素路由到相同的嵌套流或管道。

我看不到任何功能可以做到这一点，所以我编写了一个基本的 POC，它将每个元素广播到每个流。然后流/管道只过滤它应该处理的元素（见下文）。这似乎效率很低，有没有更好的方法将元素路由到特定的嵌套流？

具有相同路由键的消息应由相同的流/管道处理

Monix 的作者说Monix 与 FS2 的比较

FS2更好的地方：

• 生产者和消费者之间的通信模型是基于拉的，有时更容易实现新的运营商

Monix 更好的地方：

• 生产者和消费者之间的通信模型是基于推送的（带有背压），这使得它本质上更高效

几个问题出现：

• 在基于拉的模型中哪些操作更容易实现？
• 是否有更难以这种方式实施的操作？
• 为什么基于拉的方法本质上较慢？

我想实现一个http4s服务器，它从另一个服务接收内容，处理它并返回响应。

原始服务使用重定向，所以我添加了Follow 重定向中间件。我还添加了 Logger 中间件来检查生成的日志。

服务的骨架是：

当我尝试使用 curl 访问服务时：

响应包含原始内容的第一部分，并以：

查看日志，它们包含以下行：

这表明接收过早的 EOF 时出错。

我在这个问题上找到了一个可能的答案：但答案建议使用不推荐使用的方法，如 tohttpService。

我想我需要使用流重写代码，但我不确定更惯用的方法是什么。一些建议？

我已经定义了存储库：

它的实现：

在应用程序中，我正在执行以下操作：

问题是我怎样才能将它打印到控制台？（通常对提取的数据进行一些操作）

我试过了：

但不成功

我有一个流列表，列表的List[Stream[_]]大小在函数的开头是已知的，每个流的大小等于n或n+1。我想获得交错流，例如

我正在使用fs2.Stream. 我的第一个观点：

我正在寻找基于基本操作的解决方案（map, fold,...）