将fs2流输出分成两个文件

问题描述

我刚刚开始使用fs2流进行冒险。我要实现的是读取文件（很大的文件，这就是我使用fs2的原因），将其转换并将结果写入两个不同的文件（基于某些谓词）。一些代码（来自https://github.com/typelevel/fs2），带有我的评论：

  val converter: Stream[IO,Unit] = Stream.resource(Blocker[IO]).flatMap { blocker =>
    def fahrenheitToCelsius(f: Double): Double =
      (f - 32.0) * (5.0/9.0)

    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).toString)
      .intersperse("\n")
      .through(text.utf8Encode)
      .through(io.file.writeAll(Paths.get("testdata/celsius.txt"),blocker))
      /* instead of the last line I want something like this:
      .through(<write temperatures higher than 10 to one file,the rest to the other one>)
      */
  }

最有效的方法是什么？显而易见的解决方案是让两个流具有不同的过滤器，但是效率低下（会有两次通过）。

解决方法

不幸的是，据我所知，没有简单的方法将 fs2 流分成两部分。

您可以做的是通过将值推入两个队列之一来拆分流（第一个表示小于10的值，第二个表示大于或等于10的值）。如果我们使用NoneTerminatedQueue，那么直到我们将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,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,"testdata/celsius-over.txt") //we run reading and writing to both
      .concurrently(write(under.dequeue,"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)

}

,

我设法找到另一种解决方案。在这里：

import cats.effect.{Blocker,IOApp,Resource}
import fs2.{io,text,Stream}
import fs2.io.file.WriteCursor
import java.nio.file.Paths

object Converter extends IOApp {

  val converter: Stream[IO,Unit] = Stream.resource(Blocker[IO]).flatMap { blocker =>
    def fahrenheitToCelsius(f: Double): Double =
      (f - 32.0) * (5.0/9.0)

    def saveFiltered(in: Stream[IO,blocker: cats.effect.Blocker,filename: String,filter: Double => Boolean) = {
      val processed = in.filter(filter).intersperse("\n").map(_.toString).through(text.utf8Encode)

      Stream.resource(WriteCursor.fromPath[IO](Paths.get(filename),blocker)).flatMap(_.writeAll(processed).void.stream)
    }

    io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"),4096)
      .through(text.utf8Decode)
      .through(text.lines)
      .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
      .map(line => fahrenheitToCelsius(line.toDouble))
      .observe( in => saveFiltered(in,"testdata/celsius_over.txt",{n => n >= 0}) )
      .through( in => saveFiltered(in,"testdata/celsius_below.txt",{n => n < 0}) )
  }

  def run(args: List[String]): IO[ExitCode] =
    converter.compile.drain.as(ExitCode.Success)
}

我认为它比涉及队列的答案要容易理解（尽管队列在类似情况下似乎是常见的解决方案）。