Bulkhead.scala

object Graduation extends DefaultRunnableSpec {
  def spec =
    suite("Graduation") {
      test("bulkhead") {

        type Task = Has[Console] with Has[Live]

        val task = Live.live(ZIO.sleep(100.millis)) *> Console.print(".")

        final case class Bulkhead(ref: TRef[Int], maxInProcess: Int) {
          def acquire: UIO[Unit] = (ref.get.retryUntil(_ < maxInProcess) *> ref.update(_ + 1)).commit
          def release: UIO[Unit] = ref.update(_ - 1).commit
        }

        for {
          ref       <- TRef.make(0).commit
          bulkhead  = Bulkhead(ref, 10)
          counter   <- Ref.make[Int](0)
          taskQueue <- Queue.dropping[RIO[Task, Unit]](100)
          _         <- ZIO.foreachParDiscard(1 to 100)(_ => taskQueue.offer(task))
          _ <- ZIO.foreachParDiscard(1 to 100) { _ =>
                (bulkhead.acquire *> taskQueue.take.flatten *> counter.update(_ + 1)).ensuring(bulkhead.release)
              }
          count <- counter.get
        } yield assertTrue(count == 100)
      }
    }
}