//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.core.cluster

import java.util.concurrent.TimeoutException
import akka.actor._
import akka.pattern.gracefulStop
import akka.util.Timeout
import scala.concurrent.duration._
import it.agilelab.bigdata.wasp.core.WaspEvent.OutputStreamInitialized
import it.agilelab.bigdata.wasp.core.WaspEvent.NodeInitialized

abstract class ClusterAwareNodeGuardian extends ClusterAware {

  import akka.actor.SupervisorStrategy._

  // customize
  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1.minute) {
      case _: ActorInitializationException => Stop
      case _: IllegalArgumentException => Stop
      case _: IllegalStateException => Restart
      case _: TimeoutException => Escalate
      case _: Exception => Escalate
    }

  override def preStart(): Unit = {
    super.preStart()
    log.info("Starting at {}", cluster.selfAddress)
  }

  override def postStop(): Unit = {
    super.postStop()
    log.info("Node {} shutting down.", cluster.selfAddress)
    cluster.leave(self.path.address)
    gracefulShutdown()
  }

  override def receive: Actor.Receive = uninitialized orElse initialized orElse super.receive


  def uninitialized: Actor.Receive = {
    case OutputStreamInitialized => initialize()
  }

  def initialize(): Unit = {
    log.info(s"Node is transitioning from 'uninitialized' to 'initialized'")
    context.system.eventStream.publish(NodeInitialized)
  }

  def initialized: Actor.Receive

  def gracefulShutdown(): Unit = {
    val timeout = Timeout(5.seconds)
    context.children foreach (gracefulStop(_, timeout.duration))
    log.info(s"Graceful shutdown completed.")
  }
} 

//设置package包名称以及导入依赖的类
package ooyala.common.akka

import akka.actor.{ActorSystem, ActorRef}
import akka.pattern.gracefulStop
import scala.concurrent.Await

object AkkaTestUtils {
  import scala.concurrent.duration._

  // This is a var for now because we need to let people change it, and we can't pass this in as a param
  // because then we would change the API.  If we have it as a default param, we can't have multiple methods
  // with the same name.
  var timeout = 15 seconds

  def shutdownAndWait(actor: ActorRef) {
    if (actor != null) {
      val stopped = gracefulStop(actor, timeout)
      Await.result(stopped, timeout + (1 seconds))
    }
  }

  def shutdownAndWait(system: ActorSystem) {
    if (system != null) {
      system.shutdown()
      system.awaitTermination(timeout)
    }
  }
}