Scala/ZIO Concurrency challenge by @jdegoes | http://degoes.net/articles/zio-challenge

scala_concurrency_challenge.scala

import scalaz.zio._

trait Tap[-E1, +E2] {
  def apply[R, E >: E2 <: E1, A](effect: ZIO[R, E, A]): ZIO[R, E, A]
}

object Tap {
  type Percentage = Int

  def make[E1, E2](errBound: Percentage,
                   qualified: E1 => Boolean,
                   rejected: => E2): UIO[Tap[E1, E2]] =
    for {
      samples <- Ref.make[List[JobResult]](List.empty)
      errorsGauge = Rate.make(samples, 100)
    } yield new Tap[E1, E2] {

      def apply[R, E >: E2 <: E1, A](effect: ZIO[R, E, A]): ZIO[R, E, A] = for {
        errorRate <- errorsGauge.probe(Failing)
        result <-
          if (errorRate > errBound)
            errorsGauge.add(Successful) *> ZIO.fail(rejected)
          else
            exec(effect)
      } yield result

      private def exec[R, E >: E2 <: E1, A](eff: ZIO[R, E, A]): ZIO[R, E, A] =
        for {
          job <- eff.fork
          result <- job.join.catchSome({
            case e =>
              errorsGauge.add(
                if (qualified(e))
                  Failing
                else
                  Successful
              ) *> ZIO.fail(e)
          }).foldM(ZIO.fail, errorsGauge.add(Successful) *> ZIO.succeed(_))
        } yield result
    }
}

sealed trait JobResult
case object Successful extends JobResult
case object Failing extends JobResult

trait Rate[-A] {
  def probe[B <: A](b: B): UIO[Tap.Percentage]
  def add[B <: A](sample: B): UIO[Unit]
}

object Rate {
  def make[A](samples: Ref[List[A]],
              historySize: Int): Rate[A] =
    new Rate[A] {

      def probe[B <: A](b: B): UIO[Tap.Percentage] = samples.get.map { samples =>
        if (samples.isEmpty)
          0
        else
          (samples.count(_.equals(b)).toDouble / samples.size * 100).toInt  // 0..100
      }

      def add[B <: A](b: B): UIO[Unit] = for {
        _ <- samples.update(b :: _.take(historySize - 1))
      } yield ()
    }
}