我对 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
import cats.effect.{Blocker, ExitCode, IO, IOApp, Timer}
import fs2.Stream
import fs2.io
import fs2.concurrent.Queue
import scala.concurrent.duration._
import scala.util.Random
class StreamTypeIntToDouble(q: Queue[IO, Int])(implicit timer: Timer[IO]) {
import core.Processing._
val blocker: Blocker =
Blocker.liftExecutionContext(
scala.concurrent.ExecutionContext.Implicits.global
)
def storeInQueue: Stream[IO, Unit] = {
Stream(1, 2, 3)
.covary[IO]
.evalTap(n => IO.delay(println(s"Pushing $n to Queue")))
.metered(Random.between(1, 20).seconds)
.through(q.enqueue)
}
def getFromQueue: Stream[IO, Unit] = {
q.dequeue
.evalMap(n => IO.delay(println(s"Pulling from queue $n")))
.through(
io.file
.writeAll(Paths.get("file.txt"), blocker)
)
}
}
object Five extends IOApp {
override def run(args: List[String]): IO[ExitCode] = {
val program = for {
q <- Queue.bounded[IO, Int](10)
b = new StreamTypeIntToDouble(q)
_ <- b.storeInQueue.compile.drain.start
_ <- b.getFromQueue.compile.drain
} yield ()
program.as(ExitCode.Success)
}
}