/*
 * =========================================================================================
 * Copyright © 2013-2014 the kamon project <http://kamon.io/>
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
 * except in compliance with the License. You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the
 * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied. See the License for the specific language governing permissions
 * and limitations under the License.
 * =========================================================================================
 */

package kamon.trace

import akka.actor.{ActorLogging, Props, Actor, ActorRef}
import kamon.trace.Incubator.{CheckForCompletedTraces, IncubatingTrace}
import kamon.util.{NanoInterval, RelativeNanoTimestamp}
import scala.annotation.tailrec
import scala.collection.immutable.Queue
import kamon.util.ConfigTools.Syntax

class Incubator(subscriptions: ActorRef) extends Actor with ActorLogging {
  import context.dispatcher
  val config = context.system.settings.config.getConfig("kamon.trace.incubator")

  val minIncubationTime = new NanoInterval(config.getFiniteDuration("min-incubation-time").toNanos)
  val maxIncubationTime = new NanoInterval(config.getFiniteDuration("max-incubation-time").toNanos)
  val checkInterval = config.getFiniteDuration("check-interval")

  val checkSchedule = context.system.scheduler.schedule(checkInterval, checkInterval, self, CheckForCompletedTraces)
  var waitingForMinimumIncubation = Queue.empty[IncubatingTrace]
  var waitingForIncubationFinish = List.empty[IncubatingTrace]

  def receive = {
    case tc: TracingContext ⇒ incubate(tc)
    case CheckForCompletedTraces ⇒
      checkWaitingForMinimumIncubation()
      checkWaitingForIncubationFinish()
  }

  def incubate(tc: TracingContext): Unit =
    waitingForMinimumIncubation = waitingForMinimumIncubation.enqueue(IncubatingTrace(tc, RelativeNanoTimestamp.now))

  @tailrec private def checkWaitingForMinimumIncubation(): Unit = {
    if (waitingForMinimumIncubation.nonEmpty) {
      val it = waitingForMinimumIncubation.head
      if (NanoInterval.since(it.incubationStart) >= minIncubationTime) {
        waitingForMinimumIncubation = waitingForMinimumIncubation.tail

        if (it.tc.shouldIncubate)
          waitingForIncubationFinish = it :: waitingForIncubationFinish
        else
          dispatchTraceInfo(it.tc)

        checkWaitingForMinimumIncubation()
      }
    }
  }

  private def checkWaitingForIncubationFinish(): Unit = {
    waitingForIncubationFinish = waitingForIncubationFinish.filter {
      case IncubatingTrace(context, incubationStart) ⇒
        if (!context.shouldIncubate) {
          dispatchTraceInfo(context)
          false
        } else {
          if (NanoInterval.since(incubationStart) >= maxIncubationTime) {
            log.warning("Trace [{}] with token [{}] has reached the maximum incubation time, will be reported as is.", context.name, context.token)
            dispatchTraceInfo(context);
            false
          } else true
        }
    }
  }

  def dispatchTraceInfo(tc: TracingContext): Unit = subscriptions ! tc.generateTraceInfo

  override def postStop(): Unit = {
    super.postStop()
    checkSchedule.cancel()
  }
}

object Incubator {

  def props(subscriptions: ActorRef): Props = Props(new Incubator(subscriptions))

  case object CheckForCompletedTraces
  case class IncubatingTrace(tc: TracingContext, incubationStart: RelativeNanoTimestamp)
}