fs2でテキストファイルの読み書き

scalaのストリームプロセッシングライブラリであるfs2でテキストファイルの読み込み、変換、書き込みをやってみました。

#Scala

佐々木一洋

2020.12.26

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

最近scalaのストリームプロセッシングライブラリであるfs2を試してみたのですが、テキストファイルの読み書きをするサンプルコードが見つけられず少し試行錯誤をしたのでそのメモです。

コード

出落ちですが、以下はテキストファイルを1行ずつ読んで、行を逆にしてテキストファイルに書き出すサンプルです。

package example

import cats.effect.{Blocker, ContextShift, ExitCode, IO, IOApp, Resource, Sync}
import fs2.{io, text}

import java.nio.file.Paths
import java.util.concurrent.Executors
import scala.concurrent.ExecutionContext

class TextOpsExample[F[_]: Sync: ContextShift] {

  def readLines(inputFile: String,
                outputFile: String,
                blocker: Blocker): F[Unit] =
    io.file
      .readAll(Paths.get(inputFile), blocker, 4096) //テキストファイル読み込み
      .through(text.utf8Decode) //UTF8でデコード
      .through(text.lines) //改行ごとに分割
      .map(_.reverse) //テキストを逆に
      .intersperse("\n") //改行コードで結合
      .through(text.utf8Encode) //UTF8でエンコード
      .through(io.file.writeAll(Paths.get(outputFile), blocker)) //ファイル書き出し
      .compile
      .drain
}

object TextOpsExample extends IOApp {
  override def run(args: List[String]): IO[ExitCode] =
    for {
      _ <- Resource
        .make(IO(Executors.newFixedThreadPool(5)))(es => IO(es.shutdown()))
        .map(ExecutionContext.fromExecutor(_))
        .map(Blocker.liftExecutionContext)
        .use { blocker =>
          new TextOpsExample[IO].readLines("input.txt", "output.txt", blocker)
        }
    } yield ExitCode.Success
}

ファイルの読み書き

ファイルの読み書きにはfs2.io.file.readAllを使用します。ファイルI/Oはブロッキング処理なので、読み込みに使用するcats.effect.Blockerを指定します。サンプルでは、コンパニオンオブジェクトのrunでBlockerに必要なスレッドプールの生成と破棄を行っています。Blockerの生成はグローバルなExecutionContextからも行えますが、ブロッキング処理で使用するためスレッドプールを使用しています。スレッドプールの使い分けについてはcats effectのドキュメントが詳しいです。

readAllで読んだデータはByteなのでテキストとして扱う場合は以下で述べるデコードが必要です。