Classes to handle callback with Either[L, R]

EitherTask.scala

package x7c1.wheat.modern.callback.either

import x7c1.wheat.modern.callback.either.EitherTask.|
import x7c1.wheat.modern.patch.TaskAsync

import scala.concurrent.{Future, Promise}

class EitherTask [L, R] private (f: (Either[L, R] => Unit) => Unit){

  def flatMap[A](g: R => L | A): L | A =
    EitherTask(h => f {
      case Left(x) => h(Left(x))
      case Right(x) => g(x) run h
    })

  def map[A](g: R => A): L | A =
    EitherTask(h => f {
      case Left(x) => h(Left(x))
      case Right(x) => h(Right(g(x)))
    })

  def run(g: Either[L, R] => Unit): Unit = f(g)

  def toFuture: Future[Either[L, R]] = {
    val promise = Promise[Either[L, R]]()
    f { either =>
      try promise trySuccess either
      catch { case e: Throwable => promise tryFailure e }
    }
    promise.future
  }
}

object EitherTask {
  type | [A, B] = EitherTask[A, B]

  def apply[L, R](f: => Either[L, R]): L | R = apply { _(f) }
  def apply[L, R](f: (Either[L, R] => Unit) => Unit): L | R = new EitherTask(f)
  def fromEither[L, R](x: Either[L, R]): L | R = new EitherTask(_(x))
  def bindLeft[L]: LeftApplied[L] = new LeftApplied[L]
  def await[L](msec: Int): L | Unit = {
    EitherTask(g => TaskAsync.after(msec){
      g(Right({}))
    })
  }
  def async[L, R](f: => R): L | R = await(0) map (_ => f)
}

class LeftApplied[L]{
  def left(l: L): L | Nothing = EitherTask fromEither Left(l)
  def right[R](r: R): L | R = EitherTask fromEither Right(r)
  def await(msec: Int): L | Unit = EitherTask await msec
  def async[R](f: => R): L | R = EitherTask async f
}

EitherTaskTest.scala

package x7c1.wheat.modern.callback.either

import org.scalatest.{FlatSpecLike, Matchers}

import scala.collection.mutable.ArrayBuffer
import scala.concurrent.Await

class EitherTaskTest extends FlatSpecLike with Matchers {

  val provide = EitherTask.bindLeft[SampleError]

  it can "handle right value" in {
    val tasks = for {
      x1 <- provide right 1
      x2 <- provide right "foo"
    } yield {
      x1 + x2.length
    }
    tasks run {
      case Right(result) => result shouldBe 4
      case Left(error) => fail("unexpected error")
    }
  }
  it should "stop if error occurred" in {
    val signs = ArrayBuffer[Int]()
    val tasks = for {
      x1 <- {
        signs += 111
        provide right 1
      }
      _ <- {
        signs += 222
        provide left new SampleSubError("oops!")
      }
      x2 <- {
        signs += 333
        provide right "foo"
      }
    } yield x1 + x2.length

    tasks run {
      case Left(error) =>
        signs shouldBe Seq(111, 222)
        error.message shouldBe "oops!"
      case Right(result) =>
        fail(s"unexpected result: $result")
    }
    tasks run {
      case Left(error: SampleSubError) =>
        error.decorated shouldBe "[oops!]"
      case result =>
        fail(s"unexpected result: $result")
    }
  }
  it can "await given msec" in {
    val tasks = for {
      start <- provide right System.currentTimeMillis()
      x1 <- provide right 22
      _ <- provide await (msec = 111)
      x2 <- provide right 33
      _ <- provide await (msec = 222)
    } yield {
      val elapsed = System.currentTimeMillis() - start
      elapsed.toInt -> (x1 + x2)
    }
    val either = {
      import concurrent.duration._
      Await.result(tasks.toFuture, atMost = 5.seconds)
    }
    either match {
      case Right((elapsed, sum)) =>
        sum shouldBe 55
        elapsed should be >= 333
      case Left(error) => fail("unexpected error")
    }
  }
}

class SampleError(val message: String)

class SampleSubError(x: String) extends SampleError(x) {
  def decorated = s"[$message]"
}

TaskAsync.scala

package x7c1.wheat.modern.patch

import java.util.{Timer, TimerTask}

object TaskAsync {
  def after[A](msec: Long)(f: => A): Unit = {
    val task = new TimerTask { override def run() = f }
    new Timer().schedule(task, msec)
  }
  def async[A](f: => A): Unit = {
    after(0)(f)
  }
}