不幸的是,据我所知,没有简单的方法将fs2流分成两部分。
您可以做的是通过将值推送到两个队列之一来拆分您的流(第一个用于小于 10 的值,第二个用于大于或等于 10 的值)。如果我们使用NoneTerminatedQueue
then 队列将不会被终止,直到我们放入None
它们。然后我们可以只使用dequeue
创建单独的流,直到队列没有关闭。
下面的示例解决方案。我将写入文件拆分为单独的方法:
import java.nio.file.Paths
import cats.effect.{Blocker, ExitCode, IO, IOApp}
import fs2.concurrent.{NoneTerminatedQueue, Queue}
import fs2.{Stream, io, text}
object FahrenheitToCelsius extends IOApp {
def fahrenheitToCelsius(f: Double): Double =
(f - 32.0) * (5.0 / 9.0)
//I split reading into separate method
def read(blocker: Blocker, over: NoneTerminatedQueue[IO, Double], under: NoneTerminatedQueue[IO, Double]) = io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"), blocker, 4096)
.through(text.utf8Decode)
.through(text.lines)
.filter(s => !s.trim.isEmpty && !s.startsWith("//"))
.map(line => fahrenheitToCelsius(line.toDouble))
.evalMap { value =>
if (value > 10) { //here we put values to one of queues
over.enqueue1(Some(value)) //until we put some queues are not close
} else {
under.enqueue1(Some(value))
}
}
.onFinalize(
over.enqueue1(None) *> under.enqueue1(None) //by putting None we terminate queues
)
//function write takes as argument source queue and target file
def write(s: Stream[IO, Double], blocker: Blocker, fileName: String): Stream[IO, Unit] = {
s.map(_.toString)
.intersperse("\n")
.through(text.utf8Encode)
.through(io.file.writeAll(Paths.get(fileName), blocker))
}
val converter: Stream[IO, Unit] = for {
over <- Stream.eval(Queue.noneTerminated[IO, Double]) //here we create 2 queues
under <- Stream.eval(Queue.noneTerminated[IO, Double])
blocker <- Stream.resource(Blocker[IO])
_ <- write(over.dequeue, blocker, "testdata/celsius-over.txt") //we run reading and writing to both
.concurrently(write(under.dequeue, blocker, "testdata/celsius-under.txt")) //files concurrently
.concurrently(read(blocker, over, under)) //stream runs until queue over is not terminated
} yield ()
override def run(args: List[String]): IO[ExitCode] =
converter
.compile
.drain
.as(ExitCode.Success)
}