Async.scala

import scala.concurrent.ExecutionContext

case class Async[A, E](cb: (A => E) => E)

object Async {

  def run[A, E](async: Async[A, E])(
      implicit ec: ExecutionContext): (A => E) => E = {
    case Async(callback) =>
      f =>
        callback(f)
  }

  def >==>[A, E, B, C](fa: A => Async[B, E], fb: B => Async[C, E])(
      implicit ec: ExecutionContext): A => Async[C, E] = { input =>
    val pureA: Async[A, E] = pure(input)
    val pureB: Async[B, E] = flatMap(pureA)(fa)(ec)
    flatMap(pureB)(fb)(ec)
  }
  // we will do something similar to map:
  def flatMap[A, E, B](a: Async[A, E])(f: A => Async[B, E])(
      implicit ec: ExecutionContext): Async[B, E] = {
    Async { callback =>
      run(a)(ec) { input =>
        run(f(input))(ec)(callback)
      }
    }
  }

  // now express map in terms of flatMap:
  def map[A, E, B](a: Async[A, E])(f: A => B)(
      implicit ec: ExecutionContext): Async[B, E] = {
    flatMap(a) { input =>
      pure(f(input))
    }
  }

  def pure[A, E](a: A): Async[A, E] = Async(x => x(a))
}