//设置package包名称以及导入依赖的类
package bmpattern


import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorLogging
import akka.actor.Props

import play.api.libs.json.Json.toJson

import bmmessages._
import akka.actor.Terminated

class RoutesActor extends Actor with ActorLogging {
	var originSender : ActorRef = null
	var next : ActorRef = null
	def receive = {
		case excute(msr)  => {
		    implicit val cm = msr.cm
		    
			originSender = sender
			msr.lst match {
				case Nil => originSender ! toJson("error")
				case head :: tail => {
					head match {
						case p : ParallelMessage => {
							next = context.actorOf(ScatterGatherActor.prop(self, MessageRoutes(tail, msr.rst)), "gate")
							next ! head
						}
						case c : CommonMessage => {
							next = context.actorOf(PipeFilterActor.prop(self, MessageRoutes(tail, msr.rst)), "gate")
							next ! head
						}
					}
					
					context.watch(next)
				}
			}
		}
		case result(rst) => {
			originSender ! rst
			cancelActor
		}
		case error(err) => {
			originSender ! err
			cancelActor
		}
		case bmmessages.timeout() => {
			originSender ! toJson("timeout")
			cancelActor
		}
		case Terminated(actorRef) => println("Actor {} terminated", actorRef)
		case _ => Unit
 	}
	
	def cancelActor = {
		context.stop(self)
	}
} 

//设置package包名称以及导入依赖的类
package actors

import akka.actor.Actor
import akka.actor.ActorLogging
import akka.event.LoggingReceive
import play.api.libs.json.JsValue
import play.api.libs.json.Json
import akka.actor.ActorRef
import akka.actor.Props
import scala.xml.Utility


class UserActor(uid: String, board: ActorRef, out: ActorRef) extends Actor with ActorLogging {


  override def preStart() = {
    BoardActor() ! Subscribe
  }

  def receive = LoggingReceive {
    case Message(muid, s) if sender == board => {
      val js = Json.obj("type" -> "message", "uid" -> muid, "msg" -> s)
      out ! js
    }
    case js: JsValue => (js \ "msg").validate[String] map { Utility.escape(_) }  map { board ! Message(uid, _ ) }
    case other => log.error("unhandled: " + other)
  }
}

object UserActor {
  def props(uid: String)(out: ActorRef) = Props(new UserActor(uid, BoardActor(), out))
} 

//设置package包名称以及导入依赖的类
package actors

import akka.actor.Actor
import akka.actor.ActorLogging
import akka.event.LoggingReceive
import akka.actor.ActorRef
import akka.actor.Terminated
import play.libs.Akka
import akka.actor.Props

class BoardActor extends Actor with ActorLogging {
  var users = Set[ActorRef]()

  def receive = LoggingReceive {
    case m:Message => users map { _ ! m}
    case Subscribe => {
      users += sender
      context watch sender
    }
    case Terminated(user) => users -= user
  }
}

object BoardActor {
  lazy val board = Akka.system().actorOf(Props[BoardActor])
  def apply() = board
}

case class Message(uuid: String, s: String)
object Subscribe 

//设置package包名称以及导入依赖的类
package com.example.actors

import akka.actor.Actor
import akka.actor.ActorLogging

// Note: Usually the message object (GreeterMessages) and the actor class (GreeterActor) will be called the same thing (eg. Greeter)
object GreeterMessages {
  case object Greet
  case object Done
}

class GreeterActor extends Actor with ActorLogging {

  def receive = {
    case GreeterMessages.Greet => {
      var greetMsg = "Hello World!"

      println(greetMsg)
      log.info(greetMsg)

      sender() ! GreeterMessages.Done // Send the 'Done' message back to the sender
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.github.unknownnpc.remotedebugtool.actor

import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem}
import com.github.unknownnpc.remotedebugtool.message._

import scala.concurrent.ExecutionContext

class MainAppActor()(implicit actorSystem: ActorSystem) extends Actor with ActorLogging {

  private implicit val dispatcher: ExecutionContext = actorSystem.dispatcher

  private var jdiVmServiceActor: ActorRef = _
  private var reportServiceActor: ActorRef = _

  override def preStart(): Unit = {
    log.info("Main app actor starts")
    reportServiceActor = createReportServiceActor()
    jdiVmServiceActor = createJdiVmServiceActor()
  }

  override def postStop() {
    log.info("Main app actor stops")
  }

  override def receive: Receive = {

    case MainAppActorStart =>
      startServices()

    case MainAppActorStop =>
      stopServices()
      context.stop(self)
      context.system.terminate()

  }

  private def startServices() = {
    jdiVmServiceActor ! JdiVmServiceStart
  }

  private def stopServices() = {
    log.info("Received command to stop all services")
    jdiVmServiceActor ! JdiVmServiceStop
    reportServiceActor ! ReportServiceStop
  }

  private def createJdiVmServiceActor() = {
    context.actorOf(JdiVmServiceActor.props(reportServiceActor), "jdi-vm-service")
  }

  private def createReportServiceActor() = {
    context.actorOf(ReportServiceActor.props(self), "report-service")
  }

} 

//设置package包名称以及导入依赖的类
package sample.cluster.factorial

import scala.concurrent.duration._
import com.typesafe.config.ConfigFactory
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.ActorSystem
import akka.actor.Props
import akka.cluster.Cluster
import akka.routing.FromConfig
import akka.actor.ReceiveTimeout

//#frontend
class FactorialFrontend(upToN: Int, repeat: Boolean) extends Actor with ActorLogging {

  val backend = context.actorOf(FromConfig.props(),
    name = "factorialBackendRouter")

  override def preStart(): Unit = {
    sendJobs()
    if (repeat) {
      context.setReceiveTimeout(10.seconds)
    }
  }

  def receive = {
    case (n: Int, factorial: BigInt) =>
      if (n == upToN) {
        log.debug("{}! = {}", n, factorial)
        if (repeat) sendJobs()
        else context.stop(self)
      }
    case ReceiveTimeout =>
      log.info("Timeout")
      sendJobs()
  }

  def sendJobs(): Unit = {
    log.info("Starting batch of factorials up to [{}]", upToN)
    1 to upToN foreach { backend ! _ }
  }
}
//#frontend

object FactorialFrontend {
  def main(args: Array[String]): Unit = {
    val upToN = 200

    val config = ConfigFactory.parseString("akka.cluster.roles = [frontend]").
      withFallback(ConfigFactory.load("factorial"))

    val system = ActorSystem("ClusterSystem", config)
    system.log.info("Factorials will start when 2 backend members in the cluster.")
    //#registerOnUp
    Cluster(system) registerOnMemberUp {
      system.actorOf(Props(classOf[FactorialFrontend], upToN, true),
        name = "factorialFrontend")
    }
    //#registerOnUp
  }
} 

//设置package包名称以及导入依赖的类
package sample.cluster.simple

import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import akka.actor.ActorLogging
import akka.actor.Actor

class SimpleClusterListener extends Actor with ActorLogging {

  val cluster = Cluster(context.system)

  // subscribe to cluster changes, re-subscribe when restart 
  override def preStart(): Unit = {
    //#subscribe
    cluster.subscribe(self, initialStateMode = InitialStateAsEvents,
      classOf[MemberEvent], classOf[UnreachableMember])
    //#subscribe
  }
  override def postStop(): Unit = cluster.unsubscribe(self)

  def receive = {
    case MemberUp(member) =>
      log.info("Member is Up: {}", member.address)
    case UnreachableMember(member) =>
      log.info("Member detected as unreachable: {}", member)
    case MemberRemoved(member, previousStatus) =>
      log.info("Member is Removed: {} after {}",
        member.address, previousStatus)
    case _: MemberEvent => // ignore
  }
} 

//设置package包名称以及导入依赖的类
package sample.cluster.simple

import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import akka.actor.ActorLogging
import akka.actor.Actor

class SimpleClusterListener2 extends Actor with ActorLogging {

  val cluster = Cluster(context.system)

  // subscribe to cluster changes, re-subscribe when restart 
  override def preStart(): Unit = {
    //#subscribe
    cluster.subscribe(self, classOf[MemberEvent], classOf[UnreachableMember])
    //#subscribe
  }
  override def postStop(): Unit = cluster.unsubscribe(self)

  def receive = {
    case state: CurrentClusterState =>
      log.info("Current members: {}", state.members.mkString(", "))
    case MemberUp(member) =>
      log.info("Member is Up: {}", member.address)
    case UnreachableMember(member) =>
      log.info("Member detected as unreachable: {}", member)
    case MemberRemoved(member, previousStatus) =>
      log.info("Member is Removed: {} after {}",
        member.address, previousStatus)
    case _: MemberEvent => // ignore
  }
} 

//设置package包名称以及导入依赖的类
package sample.hello

import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.ActorRef
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.Terminated

object Main2 {

  def main(args: Array[String]): Unit = {
    val system = ActorSystem("Hello")
    val a = system.actorOf(Props[HelloWorld], "helloWorld")
    system.actorOf(Props(classOf[Terminator], a), "terminator")
  }

  class Terminator(ref: ActorRef) extends Actor with ActorLogging {
    context watch ref
    def receive = {
      case Terminated(_) =>
        log.info("{} has terminated, shutting down system", ref.path)
        context.system.shutdown()
    }
  }

} 

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

import akka.actor.ActorSystem
import akka.actor.ExtendedActorSystem
import akka.actor.Actor
import akka.actor.Terminated
import akka.actor.ActorLogging
import akka.actor.Props
import akka.actor.ActorRef
import scala.util.control.NonFatal


  def main(args: Array[String]): Unit = {
    if (args.length != 1) {
      println("you need to provide exactly one argument: the class of the application supervisor actor")
    } else {
      val system = ActorSystem("Main")
      try {
        val appClass = system.asInstanceOf[ExtendedActorSystem].dynamicAccess.getClassFor[Actor](args(0)).get
        val app = system.actorOf(Props(appClass), "app")
        val terminator = system.actorOf(Props(classOf[Terminator], app), "app-terminator")
      } catch {
        case NonFatal(e) ? system.shutdown(); throw e
      }
    }
  }

  class Terminator(app: ActorRef) extends Actor with ActorLogging {
    context watch app
    def receive = {
      case Terminated(_) ?
        log.info("application supervisor has terminated, shutting down")
        context.system.shutdown()
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.example.supervisor

import akka.actor.SupervisorStrategy.{Restart, Stop}
import akka.actor.{Actor, ActorLogging, ActorRef, OneForOneStrategy}


class CustomOneForOneUser extends Actor with ActorLogging {

  // Map that keeps track of how often a given child has been restarted
  var restarts = Map.empty[ActorRef, Int].withDefaultValue(0)

  // Using the default parameters for a SupervisorStrategy mean "arbitarily
  // often over forever"
  override val supervisorStrategy = OneForOneStrategy() {

    case _: ArithmeticException =>

      restarts(sender) match {
        case tooManyRestarts if tooManyRestarts > 15 =>
          restarts -= sender
          Stop
        case n =>
          restarts = restarts.updated(sender, n+1)
          Restart
      }
  }

  override def receive: Receive = ???
} 

//设置package包名称以及导入依赖的类
package sample.kamon

import akka.actor.{Actor, ActorLogging}
import sample.kamon.KafkaActorConsumer.Consume

class LongConsumer extends Actor with ActorLogging {

  val telemetryDumperSupervisor = context.actorOf(
    TelemetryDumperSupervisor.props(),
    "telemetry-dumper-supervisor"
  )

  lazy val topicStream: Iterable[Int] = Stream.from(1)

  override def preStart(): Unit = {
    log.info("######## LongConsumer preStart!!!")
    self ! Consume
  }

  def receive: Receive = {
    case Consume =>
      log.info("######## LongConsumer Consume!!!")
      topicStream foreach { kafkaMessage =>
        consume(kafkaMessage)
      }
  }

  def consume(kafkaMessage: Int): Unit = {
    telemetryDumperSupervisor ! kafkaMessage
  }
}

object KafkaActorConsumer {
  object Consume
  object Start
} 

//设置package包名称以及导入依赖的类
package sample.kamon

import akka.actor.SupervisorStrategy.Restart
import akka.actor.{Actor, ActorLogging, OneForOneStrategy, Props}
import sample.kamon.TelemetryDumperSupervisor.Saved

import scala.util.Random

class TelemetryDumperSupervisor extends Actor with ActorLogging {

  log.info("TelemetryDumperSupervisor started!")

  override val supervisorStrategy = OneForOneStrategy(loggingEnabled = false) {
    case e: Exception =>
      log.error(e, "There was an error when trying to save telemetry data, restarting.")
      Restart
  }

  def receive: Receive = {
    case telemetry: Int =>
      Thread.sleep(40 * Random.nextInt(10))
      context.actorOf(TelemetryDumper.props(new SimpleBackoffStrategy, telemetry))
    case Saved(telemetry) =>
    //Make some post process
  }
}

object TelemetryDumperSupervisor {
  def props(): Props = Props(classOf[TelemetryDumperSupervisor])

  case class Saved(telemetry: Int)
} 

//设置package包名称以及导入依赖的类
package lib.tools

import akka.actor.ActorLogging
import akka.persistence.PersistentActor
import lib.models.PayLoad

import scala.concurrent.duration._


trait Monitor extends PersistentActor with ActorLogging {
  context.setReceiveTimeout(120.seconds)

  final val persistenceId: String = self.path.name
  def commandHandler: Receive
  def recoverHandler:Receive
  final def receiveRecover: Receive = defaultRecoverHandler orElse recoverHandler
  final def receiveCommand: Receive = defaultCommandHandler orElse commandHandler
  final def defaultCommandHandler: Receive = {
    case payLoad: PayLoad => log.warning(s"No command handler for Payload [$payLoad] for the monitor $persistenceId")
  }
  final def defaultRecoverHandler: Receive = {
    case msg => log.warning(s"No recover handler for msg [$msg] for the monitor $persistenceId")
  }
} 

//设置package包名称以及导入依赖的类
package proton.game.hermes

import java.util.UUID

import akka.actor.ActorLogging
import akka.event.LoggingReceive
import akka.persistence.PersistentActor

import scala.collection.mutable.ListBuffer

object ChunkedHermesGameFileEntries {
  trait EntriesMessage
  case class AppendEntries(entries: Seq[HermesGameFileEntry]) extends EntriesMessage
  case class GetEntries() extends EntriesMessage

  trait EntriesEvent
  case class EntriesAppended(entries: Seq[HermesGameFileEntry]) extends EntriesEvent

  trait EntriesResult
  case class EntriesAppendedResult(id: UUID, count: Int) extends EntriesResult
  case class GetEntriesResult(id: UUID, entries: Seq[HermesGameFileEntry]) extends EntriesResult

  val gameFileEntriesRegionName = "hermesGameFileEntries"
}

class ChunkedHermesGameFileEntries(moduleSettings: HermesGameTickerModuleSettings) extends PersistentActor with ActorLogging {
  import context._
  import ChunkedHermesGameFileEntries._

  private val _id: UUID = UUID.fromString(self.path.name)
  private val _entries = new ListBuffer[HermesGameFileEntry]()

  setReceiveTimeout(moduleSettings.chunkedTimeout)

  override def receiveRecover: Receive = {
    case event: EntriesEvent => updateState(event)
  }

  def updateState(e: EntriesEvent) = e match {
    case EntriesAppended(entries) => _entries ++= entries
  }

  override def receiveCommand: Receive = LoggingReceive {
    case AppendEntries(entries) =>
      if (entries.nonEmpty) {
        persist(EntriesAppended(entries))(e => {
          updateState(e)
          sender ! EntriesAppendedResult(_id, entries.size)
        })
      } else {
        sender ! EntriesAppendedResult(_id, 0)
      }
    case GetEntries() => sender ! GetEntriesResult(_id, _entries)
  }

  override def persistenceId: String = "hermes-game-file-entries-" + _id.toString
} 

//设置package包名称以及导入依赖的类
package engine.executors

import akka.actor.{Actor, ActorLogging}

class ReducerActor extends Actor with ActorLogging {
  override def receive: Receive = {
    case ReducerDescription(jobName, reducer, chunk) =>
      log.debug(s"reduction phase of Job $jobName")
      val key = chunk.head._1
      val values = chunk.map { c =>
        c._2
      }
      val r = reducer.runReduce(key, values)
      sender ! ReductionResult(jobName, r)
  }
} 

//设置package包名称以及导入依赖的类
package engine.executors

import akka.actor.{Actor, ActorLogging}

class SorterActor extends Actor with ActorLogging {

  private val items = scala.collection.mutable.ListBuffer[(Any, Any)]()

  override def receive: Receive = {
    case PreSort(jobName, a) =>
      log.debug(s"caching for preSorter in Mapper for job $jobName items size to cache ${a.size}")
      items.append(a: _*)
    case Sort(jobName) =>
      log.debug(s"sorting <key , values> phase initiated for job $jobName - items in size ${items.size}")
      //TODO check sorting
      sender ! Sorted(jobName, items)
  }
} 

//设置package包名称以及导入依赖的类
package com.ulasakdeniz.kmeans

import akka.actor.{ActorLogging, Actor}
import com.sksamuel.scrimage.RGBColor
import com.ulasakdeniz.image.PixelData

class ReducerActor extends Actor with ActorLogging{

  def receive = idle

  def idle: Receive = {
    case initialCentroid: RGBColor => {
      context become collecting(initialCentroid, List.empty)
    }
  }

  def collecting(currentCentroid: RGBColor, clusterPoints: List[PixelData]): Receive = {
    case point: PixelData => {
      context become collecting(currentCentroid, point :: clusterPoints)
    }
    case EndOfIteration => {
      val total = clusterPoints.foldLeft(0, 0, 0){
        (acc: (Int, Int, Int), p: PixelData) => (acc._1 + p.rgb.red, acc._2 + p.rgb.green, acc._3 + p.rgb.blue)
      }
      val clusterSize = clusterPoints.size
      val newCentroid = RGBColor(total._1 / clusterSize, total._2 / clusterSize, total._3 / clusterSize)
      log.info(s"ReducerActor: NewCentroid = $newCentroid")

      context.parent ! newCentroid
      context become collecting(newCentroid, clusterPoints)
    }
    case PrepareForNextIteration => {
      context become collecting(currentCentroid, List.empty)
    }
    case SendClusterData => {
      val clusterCoordinates = clusterPoints.map(p => (p.x, p.y))
      sender ! ClusterData(currentCentroid, clusterCoordinates)
    }
  }
} 

//设置package包名称以及导入依赖的类
package info.unterstein.akka.persistence

import java.util.UUID

import akka.actor.{Actor, ActorLogging, Props}
import info.unterstein.akka.persistence.api.PersistentActorMessage
import info.unterstein.akka.persistence.client.ElasticSearchClientWrapper
import scala.util.{Failure, Success}
import ElasticSearchStoreActor._
import com.sksamuel.elastic4s.ElasticDsl._
import scala.concurrent.ExecutionContext.Implicits.global


class ElasticSearchStoreActor extends Actor with ActorLogging {

  val client = ElasticSearchClientWrapper.getByConfiguration

  def receive = {
  	case message: InitializedMessage =>
      sender ! (client.client != null)
    case message: StoreMessage =>
      val indexRequest = index into ElasticSearchClientWrapper.messageIndex / message.messageType id UUID.randomUUID.toString.replace("-", "") fields (
        ElasticSearchClientWrapper.messageFieldName -> PersistentActorMessage.mapToJson(message.originalMessage)
        )
      log.debug(s"Executing index request: $indexRequest")
      val indexResult = client.scalaClient.execute { indexRequest }
      val originalSender = sender
      indexResult onComplete {
        case Success(result) => originalSender ! StoreSuccessMessage(result.getId)
        case Failure(exception) => originalSender ! StoreFailMessage(exception)
      }
    case other => sender ! NotUnderstandable()
  }
}

object ElasticSearchStoreActor {

  case class InitializedMessage()

  case class StoreMessage(messageType: String, scheduleDate: Long = System.currentTimeMillis(), originalMessage: Map[String, String])

  case class NotUnderstandable()

  case class StoreSuccessMessage(id: String)

  case class StoreFailMessage(exception: Throwable)

  def props = Props[ElasticSearchStoreActor]
} 

//设置package包名称以及导入依赖的类
package com.scalaio.tcp.server

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.io.Tcp

object ServerActor {
  def props(bindAddress: InetSocketAddress, tcp: ActorRef, handler:ActorRef) =
    Props(classOf[ServerActor], bindAddress, tcp, handler)
}

class ServerActor(bindAddress: InetSocketAddress, tcp: ActorRef, handler: ActorRef) extends Actor with ActorLogging {
  import Tcp._

  // TODO: verify why we bind from within the actor
 tcp ! Bind(self, bindAddress)

  def receive = {
    case b @ Bound(localAddress) =>
      log.info(s"Tcp Server bound to <$localAddress>")

    case CommandFailed(_: Bind) =>
      log.warning("Tcp ServerActor failed to bind. Stopping...")
      context stop self

    case c @ Connected(remote, local) =>
      log.info(s"Tcp Server Connected. remote=<$remote>, local=<$local>. Registering handler...")
      val connection = sender()
      connection ! Register(handler)
  }
}