Task.scala

sealed trait Task[A] {
  import Task._

  final def flatMap[B](f: A => Task[B]): Task[B] = Suspend(this, f)

  final def map[B](f: A => B): Task[B] = this match {
    case Suspend(inner, suspension) =>
      Suspend(inner, suspension andThen { _ map f })

    case Async(body) =>
      Async { cb =>
        body { e =>
          cb(e rightMap f)
        }
      }

    case Now(value) => Now(f(value))
  }

  final def unsafeRunAsync(cb: Throwable \/ A => Unit): Unit = this match {
    case Suspend(inner, suspension) =>
      inner unsafeRunAsync {
        case -\/(t) => cb(t)
        case \/-(e) => suspension(e) unsafeRunAsync cb
      }

    case Async(body) => body(cb)
    case Now(value) => cb(\/-(value))
  }

  final def unsafeRun: A = {
    @volatile
    var value: A = null.asInstanceOf[A]
    val latch = new CountDownLatch(1)

    unsafeRunAsync {
      case -\/(t) => throw t
      case \/-(a) =>
        value = a
        latch.countDown()
    }

    /*
     * this latch is *already* at 0 if you're running a
     * chain of delays.  it will only result in a blocked
     * thread if there is some async computation somewhere
     * in the chain
     */
    latch.await()

    value
  }
}

object Task {

  def now[A](value: A): Task[A] = Now(a)

  def delay[A](body: => A): Task[A] = Async { _(\/-(body)) }

  def async[A](body: ((Throwable \/ A) => Unit) => Unit): Task[A] = Async(body)

  final case class Suspend[E, A](inner: Task[E], suspension: E => Task[A]) extends Task[A]
  final case class Async[A](body: ((Throwable \/ A) => Unit) => Unit) extends Task[A]
  final case class Now[A](value: A) extends Task[A]
}