fs2

fs2.scala

object Main extends App {
  import cats.effect.IO
  import fs2.{Stream, Scheduler}
  import fs2.async.signalOf
  import fs2.async.mutable.Signal
  import scala.concurrent.duration._
  import scala.concurrent.ExecutionContext.Implicits.global

  def stopAfter[A](f: FiniteDuration): Stream[IO, A] => Stream[IO, A] = in => {
    def close(s: Signal[IO, Boolean]): Stream[IO, Unit] =
      Scheduler[IO](2).flatMap { scheduler =>
        scheduler.sleep_[IO](f) ++ Stream.eval(s.set(true))
      }

    Stream.eval(signalOf[IO, Boolean](false)).flatMap { end =>
      in.interruptWhen(end).concurrently(close(end))
    }
  }

  val s =
    Stream.repeatEval(IO(println("hello")))
      .through(stopAfter(1.millis))

  s.compile.drain.unsafeRunSync()
}