Как создать Traversable с помощью ввода-вывода с эффектом кошки с учетом асинхронного вызова, который будет вызываться несколько раз

Question

Что я действительно пытаюсь сделать, так это отслеживать несколько файлов, и когда любой из них изменяется, я хотел бы обновить какое-то состояние и создать побочный эффект, используя это состояние. Я представляю, что я хочу это scan через Traversable который производит Traversable[IO[_]], Но я не вижу пути там.

в качестве минимальной попытки произвести это я написал

package example

import better.files.{File, FileMonitor}
import cats.implicits._
import com.monovore.decline._
import cats.effect.IO
import java.nio.file.{Files, Path}
import scala.concurrent.ExecutionContext.Implicits.global

object Hello extends CommandApp(
  name = "cats-effects-playground",
  header = "welcome",
  main = {
    val filesOpts = Opts.options[Path]("input", help = "input files") 
    filesOpts.map { files =>
      IO.async[File] { cb  =>
        val watchers = files.map { path =>
          new FileMonitor(path, recursive = false) {
            override def onModify(file: File, count: Int) = cb(Right(file))
          }
        }
        watchers.toList.foreach(_.start)
      }
        .flatMap(f => IO { println(f) })
        .unsafeRunSync
    }
  }
)

но у этого есть два главных недостатка. Один он создает поток для каждого файла, который я смотрю, что немного тяжело. Но что еще более важно программа завершается, как только один файл изменяется, хотя onModify будет вызван больше раз, если программа останется запущенной.

Я не женат на том, чтобы использовать лучшие файлы, это просто казалось наименьшим сопротивлением. Но я требую использования Cats IO.

Answer 1

Это решение не решает проблему создания группы потоков, и оно не создает строго Traversable, но оно решает основной вариант использования. Я очень открыт для того, чтобы меня критиковали и предлагали лучшее решение.

package example

import better.files.{File, FileMonitor}
import cats.implicits._
import com.monovore.decline._
import cats.effect.IO
import java.nio.file.{Files, Path}
import java.util.concurrent.LinkedBlockingQueue
import scala.concurrent.ExecutionContext.Implicits.global

object Hello extends CommandApp(
  name = "cats-effects-playground",
  header = "welcome",
  main = {
    val filesOpts = Opts.options[Path]("input", help = "input files")
    filesOpts.map { files =>
      val bq: LinkedBlockingQueue[IO[File]] = new LinkedBlockingQueue()
      val watchers = files.map { path =>
        new FileMonitor(path, recursive = false) {
          override def onModify(file: File, count: Int) = bq.put(IO(file))
        }
      }
      def ioLoop(): IO[Unit] = bq.take()
          .flatMap(f => IO(println(f)))
          .flatMap(_ => ioLoop())

      watchers.toList.foreach(_.start)
      ioLoop.unsafeRunSync
    }

  }
)