来自队列的 fs2 流不消耗元素

问题描述

我想从队列创建流，将所有元素打印到控制台。 当前的一个片段不打印任何内容：

object TestApp extends App {

  implicit val contextShift: ContextShift[IO] = IO.contextShift(ExecutionContext.global)

  private val value: IO[(fs2.Stream[IO,Unit],String => IO[Unit],() => IO[Unit])] = for {
    queue <- Queue.noneTerminated[IO,String]
  } yield {
    val stream: fs2.Stream[IO,Unit] = queue.dequeue.map(println)

    def send(msg: String): IO[Unit] = queue.enqueue1(Some(msg))

    def close(): IO[Unit] = queue.enqueue1(None)

    (stream,send _,close _)
  }

  val (stream,send,close) = value.unsafeRunSync()

  send("msg1").unsafeRunSync()
  send("msg2").unsafeRunSync()

}

流创建有什么问题？

解决方法

在您的示例中，您只是创建了一个队列并创建了流的描述。为了运行流，您需要调用 compile，它将公开多个方法，这将允许运行和使用流的值，如 toList、fold 或 {{1} }

在您的情况下，您对值并不真正感兴趣，因为您只想打印它们，因此您应该使用 drain:

drain