//设置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包名称以及导入依赖的类
import akka.actor.ActorSystem
import akka.routing.FromConfig
import akka.stream.ActorMaterializer
import core.authentication._
import core.authentication.tokenGenerators._
import core.db.{DatabaseContext, FlywayService}
import core.entities.Timer
import messages.parser.MessageGenerator
import messages.{MessageProcessor, Messenger}
import restapi.http.HttpService
import restapi.http.routes.HttpRouter
import utils.Configuration
import websocket.{ConnectedClientsStore, WebSocketHandler}

import scala.concurrent.ExecutionContext

class Bootstrap {

  def startup(): Unit = {
    implicit val actorSystem: ActorSystem        = ActorSystem()
    implicit val materializer: ActorMaterializer = ActorMaterializer()

    val config        = new Configuration()
    val dbc           = new DatabaseContext()
    val flywayService = new FlywayService(config)
    flywayService.migrateDatabaseSchema()

    val bearerTokenGenerator = new JwtBearerTokenGenerator(SecuredTokenGenerator.generate, Timer(config.tokenExpiration))
    val userAuthenticator    = new UserAuthenticator(UserSecretUtils.validate, bearerTokenGenerator, dbc.credentialsDao)
    val connectedClients     = actorSystem.actorOf(ConnectedClientsStore.props(), "connected-clients-store")

    val messageGenerator = new MessageGenerator()
    val messagesRouter   = actorSystem.actorOf(FromConfig.props(Messenger.props(dbc.usersDao)), "messagesRouter")
    val processingRouter = actorSystem.actorOf(FromConfig.props(MessageProcessor.props(messageGenerator, messagesRouter)), "processingRouter")

    val apiDispatcher: ExecutionContext = actorSystem.dispatchers.lookup("akka.blocking-api-dispatcher")
    val webSocketHandler = new WebSocketHandler(connectedClients, processingRouter, config.messageTimeout)(actorSystem, materializer, apiDispatcher)
    val authService = new AuthenticationService(userAuthenticator, dbc.usersDao, connectedClients)
    val httpRouter  = new HttpRouter(dbc, authService, webSocketHandler, processingRouter, config)(apiDispatcher)
    val httpService = new HttpService(httpRouter, config)
    httpService.start()
  }
} 

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

import java.util.concurrent.TimeUnit

import akka.actor.{ ActorLogging, Actor, Deploy, Props }
import akka.dispatch.{ RequiresMessageQueue, UnboundedMessageQueueSemantics }
import akka.routing.FromConfig

import scala.concurrent.duration.Duration

class SimpleDnsManager(val ext: DnsExt) extends Actor with RequiresMessageQueue[UnboundedMessageQueueSemantics] with ActorLogging {

  import context._

  private val resolver = actorOf(FromConfig.props(Props(ext.provider.actorClass, ext.cache, ext.Settings.ResolverConfig).withDeploy(Deploy.local).withDispatcher(ext.Settings.Dispatcher)), ext.Settings.Resolver)
  private val cacheCleanup = ext.cache match {
    case cleanup: PeriodicCacheCleanup ? Some(cleanup)
    case _ ? None
  }

  private val cleanupTimer = cacheCleanup map { _ ?
    val interval = Duration(ext.Settings.ResolverConfig.getDuration("cache-cleanup-interval", TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS)
    system.scheduler.schedule(interval, interval, self, SimpleDnsManager.CacheCleanup)
  }

  override def receive = {
    case r @ Dns.Resolve(name) ?
      log.debug("Resolution request for {} from {}", name, sender())
      resolver.forward(r)
    case SimpleDnsManager.CacheCleanup ?
      for (c ? cacheCleanup)
        c.cleanup()
  }

  override def postStop(): Unit = {
    for (t ? cleanupTimer) t.cancel()
  }
}

object SimpleDnsManager {
  private case object CacheCleanup
} 

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

import java.util.concurrent.TimeUnit

import akka.actor.{ ActorLogging, Actor, Deploy, Props }
import akka.dispatch.{ RequiresMessageQueue, UnboundedMessageQueueSemantics }
import akka.routing.FromConfig

import scala.concurrent.duration.Duration

class SimpleDnsManager(val ext: DnsExt) extends Actor with RequiresMessageQueue[UnboundedMessageQueueSemantics] with ActorLogging {

  import context._

  private val resolver = actorOf(FromConfig.props(Props(ext.provider.actorClass, ext.cache, ext.Settings.ResolverConfig).withDeploy(Deploy.local).withDispatcher(ext.Settings.Dispatcher)), ext.Settings.Resolver)
  private val cacheCleanup = ext.cache match {
    case cleanup: PeriodicCacheCleanup ? Some(cleanup)
    case _ ? None
  }

  private val cleanupTimer = cacheCleanup map { _ ?
    val interval = Duration(ext.Settings.ResolverConfig.getDuration("cache-cleanup-interval", TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS)
    system.scheduler.schedule(interval, interval, self, SimpleDnsManager.CacheCleanup)
  }

  override def receive: PartialFunction[Any, Unit] = {
    case r @ Dns.Resolve(name) ?
      log.debug("Resolution request for {} from {}", name, sender())
      resolver.forward(r)
    case SimpleDnsManager.CacheCleanup ?
      for (c ? cacheCleanup)
        c.cleanup()
  }

  override def postStop(): Unit = {
    for (t ? cleanupTimer) t.cancel()
  }
}

object SimpleDnsManager {
  private case object CacheCleanup
} 

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

import akka.actor.{ActorSystem, Props}
import akka.routing.FromConfig
import com.mikemunhall.simpletwitterstats.Settings
import com.mikemunhall.simpletwitterstats.api.APIService
import com.mikemunhall.simpletwitterstats.server.twitter.client.BasicTwitterClient
import com.typesafe.scalalogging.StrictLogging

object Main extends App with StrictLogging {

  logger.info("Starting application")

  // Typical actor setup
  implicit val system = ActorSystem("simple-twitter-stats")
  val settings = Settings(system)

  val parsingActor = system.actorOf(FromConfig.props(Props(classOf[ParsingActor])), "parsingActor")
  val twitterClient = BasicTwitterClient(parsingActor, settings)
  val apiService = new APIService()

  twitterClient.start
  apiService.start

  sys addShutdownHook {
    twitterClient.stop
    apiService.stop

    logger.info("Stopping application")
  }
} 

//设置package包名称以及导入依赖的类
package jug.workshops.reactive.akka.routing

import java.util.concurrent.TimeUnit

import akka.actor.{Actor, ActorSystem, Props}
import akka.routing.{FromConfig, GetRoutees, Routees}
import com.typesafe.config.ConfigFactory

object RoutersPart4DynamicRouterDemo {

  def main(args: Array[String]): Unit = {
      val config=ConfigFactory.load("routers/routersdemo")
      val system=ActorSystem("dynamicRouting",config)

      val sender=system.actorOf(Props[Sender],"dynamicDemo")

      sender ! "START"

      TimeUnit.SECONDS.sleep(20)
      system.terminate()

  }

  //change messages-per-resize = 1 to different values
  class Sender extends Actor{

    val workers=context.actorOf(FromConfig.props(Props[DynamicWorker]),"dynamicRouter")

    val tasks=List(10,10,10,10,10,10,10,100,10,10,2000,2000,200,10,10,10,10,10,10,10,2000,5000,5000)

    override def receive: Receive = {
      case "START" =>
        tasks.foreach { time =>
          println(s"sending $time")
          workers ! time
          workers ! GetRoutees
          TimeUnit.MILLISECONDS.sleep(time)  //sequencial message processing in logs is a consequence of blocking operation
        }
      case routes:Routees =>
        println(s"current number of routes" + routes.routees.length)
    }
  }

  class DynamicWorker extends Actor{
    override def receive: Receive = {
      case _ => TimeUnit.MILLISECONDS.sleep(500)
    }
  }

} 

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

import akka.actor.Actor
import backend._
import akka.actor.Props
import backend.RegionManager.UpdateUserPosition
import akka.routing.FromConfig
import models.backend.UserPosition

object RegionManagerClient {
  def props(): Props = Props(new RegionManagerClient)
}


class RegionManagerClient extends Actor {

  val regionManagerRouter = context.actorOf(Props.empty.withRouter(FromConfig), "router")

  val settings = Settings(context.system)

  def receive = {
    case p: UserPosition =>
      // Calculate the regionId for the users position
      val regionId = settings.GeoFunctions.regionForPoint(p.position)
      // And send the update to the that region
      regionManagerRouter ! UpdateUserPosition(regionId, p)
  }
} 

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

import akka.actor.Actor
import akka.actor.Props
import akka.routing.ConsistentHashingRouter.ConsistentHashable
import akka.routing.FromConfig
import models.backend.{ RegionId, RegionPoints, UserPosition }
import akka.actor.ActorLogging

object RegionManager {

  def props(): Props = Props[RegionManager]

  
class RegionManager extends Actor with ActorLogging {
  import RegionManager._

  val regionManagerRouter = context.actorOf(Props.empty.withRouter(FromConfig), "router")
  val settings = Settings(context.system)

  def receive = {
    case UpdateUserPosition(regionId, userPosition) =>
      val region = context.child(regionId.name).getOrElse {
        log.debug("Creating region: {}", regionId.name)
        context.actorOf(Region.props(regionId), regionId.name)
      }
      region ! userPosition

    case UpdateRegionPoints(regionId, regionPoints) =>
      val summaryRegion = context.child(regionId.name).getOrElse {
        log.debug("Creating summary region: {}", regionId.name)
        context.actorOf(SummaryRegion.props(regionId), regionId.name)
      }
      summaryRegion ! regionPoints

    case p @ RegionPoints(regionId, _) =>

      // count reported by child region, propagate it to summary region on responsible node
      settings.GeoFunctions.summaryRegionForRegion(regionId).foreach { summaryRegionId =>
        regionManagerRouter ! UpdateRegionPoints(summaryRegionId, p)
      }
  }

} 

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

import java.io.File

import akka.actor._
import akka.routing.FromConfig
import com.typesafe.config.ConfigFactory



class LoggerActorWorker extends Actor {
  def receive = {
    case x:Any => println(s"Message from ${context.self.path}")
  }
}

object Router {
  def main(args: Array[String]) {
    val config = ConfigFactory.parseFile(new File("src/main/resources/router/application.conf"))
    
    val system = ActorSystem("RouterExample", config)
    val router1: ActorRef = system.actorOf(FromConfig.props(Props[LoggerActorWorker]), "router1")

    val onesArray = Array.fill(20)(1)
    onesArray.toArray.par.foreach(value => {
      router1 ! (value)
      Thread.sleep(100)
    })

  }
}