FPinScalaExercise7.10.scala

package FPinScalaExercise7

import java.util.concurrent.{Executors, ExecutorService, CountDownLatch, Callable}
import java.util.concurrent.atomic.AtomicReference

import fpinscala.parallelism.Actor

/**
 * Created by BPK_t on 2015/09/15.
 */
object Main {
  def throwException() : Int = {
    throw new Exception()
    10
  }
  def main(args: Array[String]) {

    println("start")
    val p = Par.fork(Par.unit(throwException()))
    val S = Executors.newFixedThreadPool(1)
    Par.run(S)(p) match {
      case Right(x) => println("get value=" + x)
      case Left(e) => println("catch Exception" + e.toString())
    }

    println("end")
  }

  trait Future[A] {
    private[FPinScalaExercise7] def apply(k : A => Unit) : Unit
  }
  type Par[A] = ExecutorService => Future[A]

  object Par {
    //7.10
    def run[A](es : ExecutorService)(p : Par[A]) : Either[Exception, A] = {
      val ref = new AtomicReference[A]
      val latch = new CountDownLatch(1)

      try {
        p(es) { a => ref.set(a); latch.countDown }
      } catch {
        case e:Exception => return Left(e)
      }
      latch.await
      Right(ref.get)
    }

    def unit[A](a : A) : Par[A] = {
      es => new Future[A] {
        def apply(cb : A => Unit) : Unit = cb(a)
      }
    }

    def fork[A](a: => Par[A]) : Par[A] =
      es => new Future[A] {
        def apply(cb : A => Unit) : Unit = eval(es)(a(es)(cb))
      }

    //7.10
    def eval(es : ExecutorService)(r: => Unit) : Unit =
      es.submit(new Callable[Unit] {def call = r}).get()

    def map2[A, B, C](p : Par[A], p2 : Par[B])(f : (A, B) => C) : Par[C] =
      es => new Future[C] {
        def apply(cb : C => Unit) : Unit = {
          var ar : Option[A] = None
          var br : Option[B] = None
          val combiner = Actor[Either[A, B]](es) {
            case Left(a) => br match {
              case None => ar = Some(a)
              case Some(b) => eval(es)(cb(f(a,b)))
            }
            case Right(b) => ar match {
              case None => br = Some(b)
              case Some(a) => eval(es)(cb(f(a,b)))
            }
          }
          p(es)(a => combiner ! Left(a))
          p2(es)(b => combiner ! Right(b))
        }
      }

    def map[A,B](pa: Par[A])(f: A => B): Par[B] =
      map2(pa, unit(()))((a, _) => f(a))

    //7.11
    def choiceN[A](n : Par[Int])(choices : List[Par[A]]): Par[A] =
      es => choices(run(es)(n).right.get)(es)

    //7.12
    def choiceMap[K,V](key: Par[K])(choices: Map[K, Par[V]]): Par[V] =
      es => choices(run(es)(key).right.get)(es)

    //7.13
    def chooser[A,B](pa: Par[A])(choices: A => Par[B]): Par[B] =
      es => choices(run(es)(pa).right.get)(es)

    //7.13
    def choice2[A](cond: Par[Boolean])(t: Par[A], f: Par[A]): Par[A] =
      chooser(cond)(if (_) t else f)

    //7.13
    def choiceN2[A](n : Par[Int])(choices : List[Par[A]]): Par[A] =
      chooser(n)(choices(_))

    //7.14
    def join[A](a: Par[Par[A]]): Par[A] =
      es => run(es)(a).right.get(es)

    //7.14
    def flatMap[A,B](a: Par[A])(f: A => Par[B]): Par[B] =
      join(map(a)(f))

    //def join2[A](a: Par[Par[A]]): Par[A] =
    //  flatMap(a)
  }
}