该程序允许将 Mapping Ints 推送到 Double 并识别队列的退出时间。该程序没有显示任何错误,但它没有打印任何内容。我错过了什么?
import cats.effect.{ExitCode, IO, IOApp, Timer}
import fs2._
import fs2.concurrent.Queue
import scala.concurrent.duration._
import scala.util.Random
class Tst(q1: Queue[IO, (Double, IO[Long])])(implicit timer: Timer[IO]) {
val streamData = Stream.emit(1)
val scheduledStream = Stream.fixedDelay[IO](10.seconds) >> streamData
def storeInQueue: Stream[IO, Unit] = {
scheduledStream
.map { n =>
val entryTime =
timer.clock.realTime(java.util.concurrent.TimeUnit.SECONDS)
(n.toDouble, entryTime)
}
.through(q1.enqueue)
.evalTap(n => IO.delay(println(s"Pushing $n to Queue")))
q1.dequeue
.evalTap(_ => timer.sleep(Random.between(10, 30).seconds))
.map { n =>
val exitTime =
timer.clock.realTime(java.util.concurrent.TimeUnit.SECONDS)
(n._1, exitTime)
}
.evalMap(n => IO.delay(println(s"Pulling from queue $n")))
}
}
object Five2 extends IOApp {
override def run(args: List[String]): IO[ExitCode] = {
val program = for {
q <- Queue.bounded[IO, (Double, IO[Long])](1)
b = new Tst(q)
_ <- b.storeInQueue.compile.drain
} yield ()
program.as(ExitCode.Success)
}
}