如何在另一个计算中重用队列？ fs2 流 Scala

Question

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

import cats.effect.{ExitCode, IO, IOApp, Timer}
import fs2.Stream
import fs2.concurrent.Queue

import scala.concurrent.duration._
import scala.util.Random

class StreamTypeIntToDouble(q1: Queue[IO, Int], q2: Queue[IO, String])(
  implicit timer: Timer[IO]
) {

  def storeInQueueFirst: Stream[IO, Unit] = {
    Stream(1, 2, 3)
      .covary[IO]
      .evalTap(n => IO.delay(println(s"Pushing $n to Queue First")))
      .metered(Random.between(1, 20).seconds)
      .through(q1.enqueue)

  }
  def getFromQueueFirst: Stream[IO, Unit] = {
    q1.dequeue
      .evalMap(n => IO.delay(println(s"Pulling from queue Second $n")))

  }
  def storeInQueueSecond(s: Stream[IO, Int]): Stream[IO, Unit] = {
    s.map { n =>
        n.toString
      }
      .metered(Random.between(1, 20).seconds)
      .through(q2.enqueue)
  }

  def getFromQueueSecond: Stream[IO, Unit] = {
    q2.dequeue
      .evalMap(n => IO.delay(println(s"Pulling from queue second $n")))
  }
}

object Five extends IOApp {
  override def run(args: List[String]): IO[ExitCode] = {
    val program = for {
      q1 <- Queue.bounded[IO, Int](10)
      q2 <- Queue.bounded[IO, String](10)

      b = new StreamTypeIntToDouble(q1, q2)
      _ <- b.storeInQueueFirst.compile.drain.start
      a <- b.getFromQueueFirst.compile.drain
      _ <- b.storeInQueueSecond(a).compile.drain
      _ <- b.getFromQueueSecond.compile.drain
    } yield ()
    program.as(ExitCode.Success)
  }
}

Answer 1

尝试更改getFromQueueFirst，使其生成Stream[IO, Int]，而不是Stream[IO, Unit]

def getFromQueueFirst: Stream[IO, Int] = {
  q1.dequeue
    evalTap(n => IO.delay(println(s"Pulling from queue Second $n")))
}

然后

val program = for {
  q1 <- Queue.bounded[IO, Int](10)
  q2 <- Queue.bounded[IO, String](10)

  b = new StreamTypeIntToDouble(q1, q2)
  _ <- b.storeInQueueFirst.compile.drain.start
  a <- b.getFromQueueFirst.compile.lastOrError
  _ <- b.storeInQueueSecond(Stream(a)).compile.drain
  _ <- b.getFromQueueSecond.compile.drain
} yield ()

编译。