package calculatePi

import akka.actor.{ Actor, ActorLogging, ActorRef, Props, ActorSystem }
import akka.routing._

case object Start

case class CalculatePi(val precision: Int)
case class ResultPi(val result: BigDecimal)

case class Calcul(startValue: Int, nbOfElts: Int)
case class Result(value: BigDecimal)

class Calculator extends Actor with ActorLogging {

  def calculatePiFor(start: Int, nbOfElts: Int) = {
    def iter(k: Int) = {
      val r1 = BigDecimal(4) / (8 * k + 1)
      val r2 = BigDecimal(2) / (8 * k + 4)
      val r3 = BigDecimal(1) / (8 * k + 5)
      val r4 = BigDecimal(1) / (8 * k + 6)
      val r5 = (BigDecimal(1) / 16).pow(k)
      r5 * (r1 - r2 - r3 - r4)
    }

    (for {
      i <- start until (nbOfElts + start)
      n = iter(i)
    } yield n).sum
  }

  def receive = {
    case Calcul(startValue, nbOfElts) =>
      log.info(s"calculate $nbOfElts elts of the series begining from $startValue")
      sender ! Result(calculatePiFor(startValue, nbOfElts))
    case _ => ()
  }

}

class TeamLeader extends Actor with ActorLogging {

  val numberOfWorker = 10
  val calculator = context.actorOf(
    Props[Calculator].withRouter(RoundRobinPool(numberOfWorker)), name = "calculator")

  val initReceive: PartialFunction[Any, Unit] = {
    case CalculatePi(precision) =>
      val p = precision / numberOfWorker
      log.info("starting for calculating pi")
      (0 to numberOfWorker).foreach { x =>
        calculator ! Calcul(x * p, p - 1)
        context.become(waitingReceive(numberOfWorker, 0, sender), true)
      }
  }

  def waitingReceive(nbOfResponses: Int, sum: BigDecimal, initialDemandor: ActorRef): PartialFunction[Any, Unit] = {
    case Result(decimalPart) =>
      if (nbOfResponses > 0) {
        log.info(s"one partial result found. waiting for ${nbOfResponses} again")
        context.become(waitingReceive(nbOfResponses - 1, sum + decimalPart, initialDemandor), true)
      } else {
        log.info(s"All partial result received. sending computed result to asker")
        initialDemandor ! ResultPi(sum)
      }
  }

  def receive = initReceive

}

class Master extends Actor with ActorLogging {

  def receive = {
    case Start =>
      val leader = context.actorOf(Props[TeamLeader], "leader")
      leader ! CalculatePi(1000000)

    case ResultPi(result) =>
      println(result)
      context.system.shutdown()
  }
}

object Main extends App {
  val system = ActorSystem("SuperMario")
  val master = system.actorOf(Props[Master], "master")
  master ! Start
  println(s"waiting for something happens...")

}