队列中的 fs2 流不消耗元素答案

【问题标题】：fs2 Stream from queue doesn't consume elements队列中的 fs2 流不消耗元素
【发布时间】：2021-06-24 16:04:07
【问题描述】：

我想从队列中创建将所有元素打印到控制台的流。当前的 sn-p 不打印任何内容：

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()

}

流创建有什么问题？

【问题讨论】：

标签： scala cats-effect fs2

【解决方案1】：

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

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

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

private val value: IO[(fs2.Stream[IO, Unit], String => IO[Unit], () => IO[Unit])] = for {
  queue <- Queue.noneTerminated[IO, String]
} yield {
  //I changed map to evalMap, since printing is effect and should be wrapped in IO
  val stream: fs2.Stream[IO, Unit] = queue.dequeue.evalMap(v => IO(println(v)))

  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()
//Closing of stream will be delayed by 5s and run in separate fiber
close().delayBy(5.seconds).start.unsafeRunSync()

//steam would block here until it's closed
stream.compile.drain.unsafeRunSync()

【讨论】：