问题标签 [fs2]

我对 fs2 流、处理元素进行了小测试，等待然后将它们写入文件。我收到一个类型错误，但我无法弄清楚它的含义：

错误：required: fs2.Stream[[x]cats.effect.IO[x],Unit] => fs2.Stream[[+A]cats.effect.IO[A],Unit], found : [F[_]]fs2.Pipe[F,Byte,Unit] 导入 java.nio.file.Paths

该程序允许将 Mapping Ints 推送到 Double 并识别队列的退出时间。该程序没有显示任何错误，但它没有打印任何内容。我错过了什么？

我收到了一个错误，Unit而不是Stream[IO, String]。我正在尝试在下一个队列中重用队列的结果

我正在尝试将流的元素从队列顺序传递到另一个队列，以使队列一次可以接收一个元素。推入队列的结果应该被拉出并传递给下一个队列进行计算。该代码没有显示任何执行错误，但它停止了第一个队列（拉和推）。谁能解释我错过了什么.??

输出如下：

预期的输出是：

我正在使用fs2-kafka从embedded-kafka读取。

我使用创建嵌入式 kafka withRunningKafkaOnFoundPort，创建主题并发布一些消息。但是，当我尝试使用 fs2-kafka 读回它时，我得到了 NullPointerException。我已经隔离了一个测试用例，代码如下。

这是我的代码：

build.sbt：

这里是堆栈跟踪：

我曾经fs2.concurrent.Queue通过队列尝试生产者/消费者。

我现在使用的解决方案是p停止生产Some(id)，其中 id 是客户的 id，openingSeconds然后更改为 Stream ofNone以终止cie q.dequeue.unNoneTerminate。

我可以使用单个None而不是重复的流None来终止c吗？

我遇到的问题None是当队列q已满时，单None将不会插入队列，因为writeToQueue当队列已满时不会接受新元素。p并c以不同的速度生产和消费。谢谢

我正在尝试在http4s服务器上的多个 http 请求之间共享状态。

这就是我尝试过的：

在我在 http 请求中更改状态后，状态保持不变。

Ref是否可以使用或来自的东西以纯 FP 风格共享“状态” Fs2？

更新：问题出在我的应用程序中。与我如何通过参考无关。我的错。

我有以下返回的特征fs2.Stream

返回fs2.Stream的 s 是无限的，因此应该能够阻止它们。给定一个实现trait Test是否有可能创建一个Resource[IO, Test]停止def action(arg: Int): fs2.Stream[IO, Unit]发布时返回的所有流？

我正在使用可以生成fs2.Stream[ConnectionIO, Int]数据库行的 doobie（Int在我的情况下）。这ConnectionIO[A]是 doobie 的内部类型。

给定一个HKT F[_]，val nt: ConnectionIO ~> F有没有办法fs2.Stream[F, Int]从fs2.Stream[ConnectionIO, Int]？

我没有找到某种mapK为fs2.Stream.

我有一个fs2.Stream由一些元素（可能是无限的）组成的，我想为流的所有元素同时安排一些计算。这是我尝试过的

程序输出看起来像

这不是预期的。我想为原始流的所有元素交错调度计算，但不要等到第一个流终止（由于无限调度而永远不会发生）。