У меня есть поток кусков String. Теперь я хочу записать эти куски в файлы. В каждом файле я хочу написать N кусков. Я написал следующий код, который решает проблему. Но мой подход, безусловно, не идиоматический c.
object Diff extends App{
implicit val cs: ContextShift[IO] = IO.contextShift(ExecutionContext.global)
val l: Stream[Pure, (Chunk[String], Int)] =
Stream("Jakob", "Leo", "Tom", "Anton", "Lavinia", "Daniel")
.intersperse("\r\n")
.chunkN(4)
.zip(Stream(1,2,3,4))
val writeJob: Stream[Pure, ExitCode] = l.map {
case (ch, i) =>
val path = Paths.get(s"file_$i")
Stream.resource(Blocker[IO]).flatMap { blocker =>
Stream.chunk(ch)
.covary[IO]
.through(text.utf8Encode)
.through(file.writeAll(path, blocker))
}.compile.drain.as(ExitCode.Success).unsafeRunSync()
}
writeJob.compile.drain.as(ExitCode.Success)
}
1) Поэтому я называю unsafeRunSync()
более одного раза, что является недостатком (поскольку карта вызывается много раз). Но я не знаю, как решить это по-другому. Я новичок в FS2 и Cats-Effects.
2) Также я звоню chunkN(4)
здесь. Который в моем примере выглядит безобидным. Но в моем случае я бы использовал chunkN(10 ^ 6)
. Правильно ли я понимаю Chunks
из документации, что FS2 накапливает 10 ^ 6 порцию в памяти и затем отправляет их в нисходящий поток?
Я знаю, что есть функция fileRotate, которая работает и байты как limit
Но я думаю, что это не очень помогает в моем случае использования.
Спасибо