fs2-jobs.scala

import cats.effect.{ExitCode, IO, IOApp}
import cats.instances.list._
import cats.syntax.all._
import fs2._

import scala.concurrent.duration._

object jobs extends IOApp {

  val largeStream: Stream[IO, Int] = Stream.range(0, 100).covary[IO]

  def listParallelProcessor(list: List[Int]): Stream[IO, Unit] =
    Stream.eval(list.parTraverse(x => IO(println(s"Processing: $x"))).void)

  val p1: Stream[IO, Unit] =
    largeStream
      .segmentN(10)
      .flatMap(s => listParallelProcessor(s.force.toList) ++ Stream.sleep_[IO](2.seconds))

  val sinkProcessor: Sink[IO, Int] = _.evalMap(x => IO(println(s"Processing: $x")))

  val p2: Stream[IO, Unit] =
    largeStream
      .segmentN(10)
      .map(s => Stream.segment(s).covary[IO].to(sinkProcessor) ++ Stream.sleep_[IO](2.seconds))
      .join(2)

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

}