package su.hil.tesseract.internal

import akka.actor.ActorRef
import akka.cluster.Cluster
import akka.cluster.ddata.Replicator._
import akka.cluster.ddata._
import akka.dispatch.Futures
import akka.pattern.ask
import su.hil.tesseract.utils.{ Utils, Complete }

import scala.concurrent.Future
import Context._

/** Creation date: 27.06.2015
  * Copyright (c) harati
  */

/** Асинхронная статистика тессеракта
  */
object Statistics extends Module {

    private val countersKey = PNCounterMapKey("stats.counters")

    private var replicator: ActorRef = _

    def push(f: ServerData)(implicit context: Context) = (replicator ? Update(countersKey, f.counters, WriteLocal)(x ⇒ x)).mapTo[UpdateSuccess[_]]

    import Context._

    def pull(implicit context: Context): Future[ServerData] = (for (
        f ← ask(replicator, Get(countersKey, ReadLocal));
        result = f
    ) yield result).map {
        case f @ GetSuccess(firstKey: PNCounterMapKey, firstData) ⇒ new ServerData(f.get(firstKey))
        case NotFound(key, q) ⇒ throw new StatsRetrieveException
    }

    private def empty(implicit context: Context): ServerData = new ServerData(PNCounterMap().increment("online", 0))

    private[internal] class ServerData(
            private[internal] var counters: PNCounterMap)(implicit context: Context) {

        def online = counters.get("online").get
        def onlineInc(): Unit = counters = counters.increment("online")
        def onlineDec(): Unit = counters = counters.decrement("online")

        def push() = Statistics.push(this)

    }

    override def name: String = "Statistics"

    override def stop(implicit context: Context): Future[Complete] = Futures.successful(Complete())

    override def priority: Int = 0

    override def start(implicit creator: Core): Future[Complete] = {
        replicator = DistributedData(creator.context.system).replicator
        push(empty)
        Futures.successful(Complete())
    }
}

class StatsRetrieveException extends RuntimeException