この記事は公開されてから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なのでテキストとして扱う場合は以下で述べるデコードが必要です。
データのデコード、エンコード
読み込んだデータのデコード、改行での分割はfs.textに定義されているPipeで行えます。ファイルからの読み込み時、書き込み時にそれぞれデコードとエンコードが必要です。プリセットではutf8、BASE64へのエンコード、デコードが提供されています。
テキストの操作
読み込んだテキストは組み込み型のStringなので、mapやflatMapなどの見慣れたコンビネーターを使って容易に変換が行えます。
まとめ
fs2でテキストファイルの読み込み、変換、書き込みをやってみました。 cats.effectなどのtypelevelのライブラリと組み合わせが容易なので扱いやすいですが、凝ったテキスト処理をする場合は自前でPipeを作ったりする必要があるかもしません。