//设置package包名称以及导入依赖的类
package ru.fediq.scrapingkit

import akka.actor.{ActorSystem, Props}
import akka.http.scaladsl.model.Uri
import akka.routing.RoundRobinPool
import com.typesafe.scalalogging.StrictLogging
import net.ceedubs.ficus.Ficus._
import net.ceedubs.ficus.readers.ArbitraryTypeReader._
import ru.fediq.scrapingkit.backend.{FeedExporter, LinksHistory, LinksQueue, PageCache}
import ru.fediq.scrapingkit.model.PageRef
import ru.fediq.scrapingkit.platform._
import ru.fediq.scrapingkit.scraper.Scraper

import scala.util.Try

class ScrapingKitReactor(
  linksQueue: LinksQueue,
  linksHistory: LinksHistory,
  pageCache: PageCache,
  exporter: FeedExporter,
  scrapers: Map[String, Scraper],
  redirectFilter: Option[(PageRef, Uri) => Boolean] = None
)(implicit val system: ActorSystem)
  extends AnyRef with AutoCloseable with StrictLogging {

  val config = system.settings.config.as[ScrapingKitConfig]("scrapingkit")

  val queueingActor = system
    .actorOf(
      Props(new QueueingActor(linksQueue, linksHistory, config))
        .withDispatcher("pinnedDispatcher"),
      "queueing"
    )

  val downloadingActor = system
    .actorOf(
      Props(new DownloadingActor(pageCache, config, redirectFilter)),
      "downloading"
    )

  val scrapingActor = system
    .actorOf(
      RoundRobinPool(config.scrapingThreads, routerDispatcher = "pinnedDispatcher")
        .props(Props(new ScrapingActor(scrapers, exporter, config))),
      "scraping"
    )

  system.registerOnTermination(close())

  override def close() = {
    logger.info("Stopping ScarpingKit Reactor")
    Try(linksQueue.close())
    Try(linksHistory.close())
    Try(pageCache.close())
    Try(exporter.close())
    logger.info("Stopped")
  }
} 

//设置package包名称以及导入依赖的类
package com.ubirch.transformer.actor

import akka.actor.{ActorLogging, ActorRef, Props}
import akka.camel.{CamelMessage, Consumer}
import akka.routing.RoundRobinPool
import com.ubirch.avatar.config.Config
import com.ubirch.avatar.core.device.DeviceManager
import com.ubirch.avatar.model.rest.device.DeviceDataRaw
import com.ubirch.avatar.util.actor.ActorNames
import com.ubirch.util.json.{Json4sUtil, MyJsonProtocol}

import scala.concurrent.ExecutionContextExecutor


class TransformerConsumerActor extends Consumer with ActorLogging with MyJsonProtocol {

  val accessKey: String = Config.awsAccessKey
  val secretKey: String = Config.awsSecretAccessKey

  override def endpointUri = s"aws-sqs://${Config.awsSqsQueueTransformer}?accessKey=$accessKey&secretKey=$secretKey&concurrentConsumers=2&maxMessagesPerPoll=10"

  override def autoAck: Boolean = true

  implicit val executionContext: ExecutionContextExecutor = context.dispatcher

  val transformerActor: ActorRef = context.actorOf(new RoundRobinPool(Config.akkaNumberOfWorkers).props(Props[TransformerPreprocessorActor]), ActorNames.TRANSFORMER_PRE)

  //TODO fix error handling, in case of error the message should be resend later?
  override def receive: Receive = {
    case msg: CamelMessage =>
      log.debug(s"received ${msg.bodyAs[String]}")
      msg.body match {
        case drdStr: String =>
          Json4sUtil.string2JValue(drdStr) match {
            case Some(drdJson) =>
              drdJson.extractOpt[DeviceDataRaw] match {
                case Some(drd) =>
                  DeviceManager.infoByHashedHwId(drd.a).map {
                    case Some(device) =>
                      transformerActor ! (device, drd)
                    case None =>
                      log.error(s"no device found for hashedHwdeviceId: ${drd.a}")
                  }
                case None =>
                  log.error(s"invalid json message from device: $drdStr")
              }
            case None =>
              log.error(s"invalid message from device: $drdStr")
          }

        case _ =>
          log.error(s"received invalid message body: ${msg.body}")
      }

    case _ =>
      log.error("received unknown message")
  }
} 

//设置package包名称以及导入依赖的类
package com.ubirch.avatar.backend.route

import akka.actor.{ActorSystem, Props}
import akka.http.scaladsl.HttpExt
import akka.http.scaladsl.server.{Directives, Route}
import akka.routing.RoundRobinPool
import akka.stream.Materializer
import akka.util.Timeout
import com.typesafe.scalalogging.slf4j.StrictLogging
import com.ubirch.avatar.config.Config
import com.ubirch.avatar.core.actor.MessageValidatorActor
import com.ubirch.avatar.model.rest.device.DeviceDataRaw
import com.ubirch.avatar.util.actor.ActorNames
import com.ubirch.avatar.util.server.RouteConstants._
import com.ubirch.util.http.response.ResponseUtil
import com.ubirch.util.model.JsonResponse
import com.ubirch.util.mongo.connection.MongoUtil

import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration._
import scala.language.postfixOps
import de.heikoseeberger.akkahttpjson4s.Json4sSupport._


class DeviceUpdateBulkRoute(implicit mongo: MongoUtil, httpClient: HttpExt, materializer: Materializer)
  extends ResponseUtil
    with Directives
    with StrictLogging {

  implicit val system = ActorSystem()
  implicit val executionContext: ExecutionContextExecutor = system.dispatcher
  implicit val timeout = Timeout(Config.actorTimeout seconds)

  private val validatorActor = system.actorOf(new RoundRobinPool(Config.akkaNumberOfWorkers).props(Props(new MessageValidatorActor())), ActorNames.MSG_VALIDATOR)

  val route: Route = {

    path(update / bulk) {

      pathEnd {

        post {

          post {
            entity(as[DeviceDataRaw]) { sdm =>

              validatorActor ! sdm

              complete(JsonResponse(message = "processing started"))
            }
          }
        }
      }
    }
  }
} 

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

import akka.actor.SupervisorStrategy.{Escalate, Restart, Resume, Stop}
import akka.actor.{Actor, OneForOneStrategy, Props, Terminated}
import akka.routing.RoundRobinPool
import com.common.Notificacao
import com.common.TipoNotificacao._
import com.exemplo5.ColetorNotificacoesActor.{ContarNotificacoes, EnviarNotificacao}
import com.exemplo5.NotificacaoPagamentoActor.NotificarPagamento

import scala.collection.mutable.ListBuffer
import scala.concurrent.duration._


class ColetorNotificacoesActor extends Actor {

  override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
    case _: ArithmeticException => Resume
    case _: NullPointerException => Restart
    case _: IllegalArgumentException => Restart
    case _: Exception => Escalate
  }

  var notificacoes = new ListBuffer[Notificacao]()

  val notificacaoPagamentoActor = context.actorOf(RoundRobinPool(10).props(Props[NotificacaoPagamentoActor]), "NotificacaoPagamentoActor")

  context.watch(notificacaoPagamentoActor)

  def receive = {
    case EnviarNotificacao(notificacao) => {
      notificacoes += notificacao

      notificacao.tipo match {
        case PAGAMENTO => notificacaoPagamentoActor ! NotificarPagamento(notificacao)
        case SAQUE => notificacaoPagamentoActor ! NotificarPagamento(notificacao)
        case TRANSFERENCIA => notificacaoPagamentoActor ! NotificarPagamento(notificacao)
      }
    }
    case ContarNotificacoes => sender ! notificacoes.size
    case Terminated(ator) => println(s"############### Ator ${ator.path} finalizado ##############")
  }

}

object ColetorNotificacoesActor {

  case object ContarNotificacoes
  case class EnviarNotificacao(notificacao: Notificacao)

} 

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

import akka.actor.{ActorRef, ActorSystem, Props}
import akka.routing.RoundRobinPool

import scala.concurrent.duration._

object ScalingOutWorker extends App {
  val actorSystem = ActorSystem("WorkerActorSystem")
  implicit val dispatcher = actorSystem.dispatcher

  val selection = actorSystem.actorSelection("akka.tcp://[email protected]:2552/user/masterActor")
  selection.resolveOne(3 seconds).onSuccess {
    case masterActor : ActorRef =>
      println("We got the ActorRef for the master actor")
      val pool = RoundRobinPool(10)
      val workerPool = actorSystem.actorOf(Props[WorkerActor].withRouter(pool), "workerActor")
      masterActor ! RegisterWorker(workerPool)
  }
}


object ScalingOutMaster extends App {
  val actorSystem = ActorSystem("MasterActorSystem")
  val masterActor = actorSystem.actorOf(Props[MasterActor], "masterActor")

  (1 to 100).foreach(i => {
    masterActor ! Work(s"$i")
    Thread.sleep(5000) //Simulates sending work to the master actor every 5 seconds
  })

} 

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

import akka.actor.{ActorSystem, Props}
import akka.routing.RoundRobinPool

object CountDownLatchApp extends App {
  implicit val actorSystem = ActorSystem()
  import actorSystem._
  val routeesToSetUp = 2
  val countDownLatch = CountDownLatch(routeesToSetUp)

  actorSystem.actorOf(Props(classOf[CountDownLatchWorker], countDownLatch)
    .withRouter(RoundRobinPool(routeesToSetUp)), "workers")

  //Future based solution
  countDownLatch.result.onSuccess { case _ => log.info("Future completed successfully") }

  //Await based solution
  countDownLatch.await()
  actorSystem.terminate()
} 

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

import java.util.UUID

import akka.actor.{Actor, ActorSystem, Props}
import akka.routing.RoundRobinPool
import fresco.filesystem.Repository
import fresco.logging.Logging



case object Create

class RepositoryCreation extends Actor with Logging {
  override def receive: Receive = {
    case Create => {
      logger.debug(s"created")
      val repoLoc = "/Users/suhail/tmp/fresco"
      Repository(repoLoc)
      Thread.sleep(2000)
    }
    case _ => {

    }
  }
}

object Launcher extends App with Logging {
//  val system = ActorSystem("system")
//  val creator = system.actorOf(RoundRobinPool(100).props(Props[RepositoryCreation]))
//  creator ! Create
//  system.terminate
//  logger.debug("application shutdown")
  (1 to 10).foreach(i => {
    val s: String = UUID.randomUUID().toString();
    println(s"id: $s")
  })

} 

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

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.persistence.cassandra.query.scaladsl.CassandraReadJournal
import akka.persistence.query.PersistenceQuery
import akka.routing.RoundRobinPool
import com.todos.repository.{TodoRepositoryProcessor, TodoRepositoryView}

class Guardian() extends Actor with ActorLogging {
  log.info("TodoService up and running...")

  val todoRepositoryProcessor: ActorRef = context.actorOf(
    TodoRepositoryProcessor.props(),
    name = TodoRepositoryProcessor.name
  )

  val readJournal: CassandraReadJournal = PersistenceQuery(context.system)
    .readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)

  val todoRepositoryView: ActorRef = context.actorOf(
    TodoRepositoryView.props(readJournal).withRouter(RoundRobinPool(5)),
    name = TodoRepositoryView.name
  )

  context.actorOf(
    Api.props(
      todoRepositoryViewPath = todoRepositoryView.path,
      todoRepositoryProcessorPath = todoRepositoryProcessor.path
    ),
    name = Api.name
  )

  def receive: Receive = Actor.emptyBehavior
}

object Guardian {
  val name: String = "guardian"

  def props(): Props = {
    Props(
      classOf[Guardian]
    )
  }

} 

//设置package包名称以及导入依赖的类
import akka.actor._
import akka.routing.RoundRobinPool
import scala.concurrent.duration._

class Master(nrOfWorkers: Int, nrOfMessages: Int, nrOfElements: Int, listener: ActorRef)
  extends Actor {
 
  var pi: Double = _
  var nrOfResults: Int = _
  val start: Long = System.currentTimeMillis
 
  val workerRouter = context.actorOf(
    Props[Worker].withRouter(RoundRobinPool(nrOfWorkers)), name = "workerRouter")

  def receive = {
	  case Calculate ?
	    for (i ? 0 until nrOfMessages) workerRouter ! Work(i * nrOfElements, nrOfElements)
	  case Result(value) ?
	    pi += value
	    nrOfResults += 1
	    if (nrOfResults == nrOfMessages) {
	      // Send the result to the listener
	      listener ! PiApproximation(pi, duration = (System.currentTimeMillis - start).millis)
	      // Stops this actor and all its supervised children
	      context.stop(self)
	    }
	}
} 

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

import akka.actor.{ActorSystem, Props}
import akka.routing.RoundRobinPool

object RedPacketMain extends App {
    val system = ActorSystem("RedPacket")

    //default router supervision strategy is "Escalate", here we use "Resume" instead since we want generator not impacted by error.
    val generatorRouter = system.actorOf(RoundRobinPool(100, supervisorStrategy = ResumeSupervisor()).props(Props[RedPacketGenerator]), "generatorRouter")
    val clientRouter = system.actorOf(RoundRobinPool(100).props(Props(classOf[RedPacketClient], generatorRouter)), "clientRouter")
    val monitor = system.actorOf(Props[RedPacketMonitor], "redPacketMonitor")
    1 to 1000 foreach {
      _ => clientRouter ! RedPacketClient.Shake    
    } 
    //TODO we may need to refer akka's shutdown pattern(http://doc.akka.io/docs/akka/snapshot/scala/howto.html#Shutdown_Patterns_in_Akka_2) to stop the system gracefully.
    //system.shutdown()
} 

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

import java.util.concurrent.TimeUnit

import akka.actor._
import akka.pattern._
import akka.routing.{RoundRobinPool, RoundRobinRoutingLogic}
import akka.stream.ActorMaterializer
import akka.stream.javadsl.Framing
import akka.stream.scaladsl.{Flow, Sink, Tcp}
import akka.util.{ByteString, Timeout}
import kld.actors.{CalcDistance, DocumentHandler, GetStatus, KLDWorker}

import scala.util.{Failure, Success}

object Main extends App {

  import scala.concurrent.ExecutionContext.Implicits.global

  implicit val system = ActorSystem("KLD")

  implicit val materializer = ActorMaterializer()
  implicit val timeout = Timeout(20, TimeUnit.MINUTES)

  val worker = system.actorOf(Props(classOf[KLDWorker]).withRouter(RoundRobinPool(nrOfInstances = 10)), name = "Worker")

  val documentHandler = system.actorOf(Props(classOf[DocumentHandler], worker), name = "DocumentHandler")

  val connections = Tcp().bind("0.0.0.0", 8888)

  val handler = Sink.foreach[Tcp.IncomingConnection] { conn =>
    println("Client connected from: " + conn.remoteAddress)
    conn.handleWith(Flow[ByteString].via(Framing.delimiter(
      ByteString("""\d"""), Int.MaxValue
    )).map(_.utf8String)
      .mapAsyncUnordered(5000) { document: String =>
        val msg = if (document.length == 4 && document.equals("calc")) CalcDistance()
                  else if (document.length == 3 && document.equals("get")) GetStatus() else document
        (documentHandler ? msg).map(_.asInstanceOf[String]).map(res =>
          ByteString(res + "\n")
        )
    })
  }

  val binding = connections.to(handler).run()

  binding.onComplete {
    case Success(b) =>
      println("Please wait while data is getting ready.")
      Utils.writeCleaned()
      println("Server started, listening on: " + b.localAddress)


    case Failure(e) =>
      println(s"Could not bind server: " + e.getMessage)
      system.terminate()
  }

} 

//设置package包名称以及导入依赖的类
package aia.performance.throughput

import akka.testkit.TestProbe
import akka.actor.{Props, ActorSystem}
import org.scalatest.{WordSpecLike, BeforeAndAfterAll, MustMatchers}
import akka.routing.RoundRobinPool
import com.typesafe.config.ConfigFactory
import aia.performance.{ProcessCPURequest, SystemMessage, ProcessRequest}
import concurrent.duration._


class ThroughputCPUTest extends WordSpecLike
  with BeforeAndAfterAll
  with MustMatchers {

  val configuration = ConfigFactory.load("performance/through")
  implicit val system = ActorSystem("ThroughputTest", configuration)

  "System" must {
    "fails to with cpu" in {
      val nrWorkers = 40
      val nrMessages = nrWorkers * 40

      val end = TestProbe()
      val workers = system.actorOf(
        RoundRobinPool(nrWorkers).props(
          Props(new ProcessCPURequest(250 millis, end.ref)).withDispatcher("my-dispatcher")),
        "Workers-cpu")

      val startTime = System.currentTimeMillis()
      for (i <- 0 until nrMessages) {
        workers ! new SystemMessage(startTime, 0, "")
      }
      val msg = end.receiveN(n = nrMessages, max = 9000 seconds).asInstanceOf[Seq[SystemMessage]]
      val endTime = System.currentTimeMillis()
      val total = endTime - startTime
      println("total process time %d Average=%d".format(total, total / nrMessages))
      val grouped = msg.groupBy(_.id)
      grouped.map {
        case (key, listMsg) => (key, listMsg.foldLeft(0L) { (m, x) => math.max(m, x.duration) })
      }.foreach(println(_))

      Thread.sleep(1000)

      system.stop(workers)
      
    }
  }
} 

//设置package包名称以及导入依赖的类
package aia.performance.throughput

import akka.testkit.TestProbe
import akka.actor.{Props, ActorSystem}
import org.scalatest.{WordSpecLike, BeforeAndAfterAll, MustMatchers}
import akka.routing.RoundRobinPool
import com.typesafe.config.ConfigFactory
import aia.performance.{ProcessCPURequest, SystemMessage, ProcessRequest}
import concurrent.duration._

class ThroughputTest extends WordSpecLike
  with BeforeAndAfterAll
  with MustMatchers {

  val configuration = ConfigFactory.load("performance/through")
  implicit val system = ActorSystem("ThroughputTest", configuration)

  "System" must {
    "fails to perform" in {
      val nrMessages = 99
      val nrWorkers = 3
      val statDuration = 2000 millis //((nrMessages * 10)+1000)/4 millis

      val end = TestProbe()
      val workers = system.actorOf(
        RoundRobinPool(nrWorkers).props(Props(new ProcessRequest(1 second, end.ref)).withDispatcher("my-dispatcher")),
        "Workers")

      val startTime = System.currentTimeMillis()
      for (i <- 0 until nrMessages) {
        workers ! new SystemMessage(startTime, 0, "")
      }
      val msg = end.receiveN(n = nrMessages, max = 9000 seconds).asInstanceOf[Seq[SystemMessage]]
      val endTime = System.currentTimeMillis()
      val total = endTime - startTime
      println("total process time %d Average=%d".format(total, total / nrMessages))
      val grouped = msg.groupBy(_.id)
      grouped.map {
        case (key, listMsg) => (key, listMsg.foldLeft(0L) { (m, x) => math.max(m, x.duration) })
      }.foreach(println(_))

      Thread.sleep(1000)

      system.stop(workers)
      
    }
  }
} 

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

import akka.actor.Props
import akka.routing.RoundRobinPool
import org.apache.commons.mail.{DefaultAuthenticator, Email}
import play.api.Play
import play.api.libs.concurrent.Akka
import play.api.Play.current


package object support {
  object EmailUtil {
    val smtpHost = Play.application.configuration.getString("smtp.host").getOrElse("mail.m8chat.com")
    val smtpPort = Play.application.configuration.getInt("smtp.port").getOrElse(587)
    val smtpUser = Play.application.configuration.getString("smtp.user").getOrElse("[email protected]")
    val smtpPassword = Play.application.configuration.getString("smtp.password").getOrElse("64PscSUvTHRV8wB3")
    val smtpStartTls = Play.application.configuration.getBoolean("smtp.startTls").getOrElse(true)
    val smtpDebug = Play.application.configuration.getBoolean("smtp.debug").getOrElse(false)

    val noReplySender = "[email protected]"

    def send(mail: Email) = {
      mail.setHostName(EmailUtil.smtpHost)
      mail.setSmtpPort(EmailUtil.smtpPort)
      mail.setAuthenticator(new DefaultAuthenticator(EmailUtil.smtpUser, EmailUtil.smtpPassword))
      mail.setSSLCheckServerIdentity(false)
      mail.setStartTLSRequired(EmailUtil.smtpStartTls)
      mail.setDebug(EmailUtil.smtpDebug)
      mail.send()
    }
  }

  val ContactUsEmailRouter = Akka.system.actorOf(RoundRobinPool(3).props(Props[ContactUsEmailSender]), "support.ContactUsEmailRouter")
  val AdminEmailSender = Akka.system.actorOf(Props[AdminEmailSender], "support.AdminEmailSender")
} 

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

import akka.actor.{Props, ReceiveTimeout, Actor}
import akka.routing.RoundRobinPool
import models.advert.BillEmail
import org.joda.time.DateTime
import play.api.Logger
import play.api.db.DB
import utils._
import anorm._
import play.api.Play.current
import scala.util.{Failure, Try}

import scala.concurrent.duration._
import scala.language.postfixOps

class BillEmailScheduler extends Actor {
  val billEmailRouter = context.actorOf(RoundRobinPool(5).props(Props[BillEmailSender]), "advert.billEmailRouter")

  override def receive = {
    case "Start" =>
      context.setReceiveTimeout(10 minutes)
    case ReceiveTimeout =>
      Logger.debug("Begin to schedule sending all billing emails")

      Try(DB.withTransaction { implicit c =>
        SQL("select * from advert.get_bills_to_email()")
          .apply().map(row => billEmailRouter ! BillEmail(row[String]("_name"), row[String]("_email"),
          row[DateTime]("_issued_at"), row[Option[DateTime]]("_paid_at"),
          row[Option[DateTime]]("_expiring_at"), row[Option[DateTime]]("_canceled_at"),
          row[BigDecimal]("_amount"), row[String]("_status")))
      }) match {
        case Failure(ex) => AccessLogger.error(s"Fail to schedule sending all billing emails: ${ex.getMessage}")
        case _ => AccessLogger.debug("Schedule sending all billing emails completes once")
      }
    case "Stop" =>
      context.setReceiveTimeout(Duration.Undefined)
  }
} 

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

import akka.actor.{Props, Actor, ActorLogging}
import akka.routing.RoundRobinPool

class Ch6PingActor extends Actor with ActorLogging {
  val childrenPool = context.actorOf(RoundRobinPool(5).props(Props[Ch6Ping1Actor]), "Ping1Actor")

  def receive = {
    case Start() =>
      
      1 to 13 foreach {
        i => childrenPool ! Hash(i)
      }

    case message: Any =>
      unhandled(message)
  }
} 

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

import akka.actor.{ActorRef, ActorSystem, Props}
import akka.routing.RoundRobinPool
import com.knoldus.configuration.Configuration
import com.typesafe.config.ConfigFactory

// $COVERAGE-OFF$Disabling highlighting by default until a workaround for https://issues.scala-lang.org/browse/SI-8596 is found
trait LocalActorRefFactory {

  val actors: Map[String, ActorRef]
  val system: ActorSystem
  def getReceiver(name: String): ActorRef = {
    actors.get(name) match {
      case None => throw new IllegalArgumentException("No Actor could be looked up for the specified name " + name)
      case Some(actorRef) => actorRef
    }
  }
}

object LocalActorRefFactory extends LocalActorRefFactory{
  val system: ActorSystem = ActorSystem.create("KnoldusScheduler", ConfigFactory.load)
  val actors: Map[String, ActorRef] = Map(
    Configuration.NAME_FIRST_ACTOR -> system.actorOf(Props(classOf[FirstActor])
      .withRouter(RoundRobinPool(Runtime.getRuntime.availableProcessors()))),
    Configuration.NAME_SECOND_ACTOR -> system.actorOf(Props(classOf[SecondActor])
      .withRouter(RoundRobinPool(Runtime.getRuntime.availableProcessors())))
  )

}
// $COVERAGE-ON$ 

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

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.routing.RoundRobinPool
import nl.mollie.commands.CreatePayment
import nl.mollie.config.MollieConfig
import nl.mollie.connection.{HttpServer, MollieConnection}
import nl.mollie.queries.{GetPayment, ListPaymentIssuers, ListPaymentMethods}

class MollieClientActor(config: MollieConfig) extends Actor with ActorLogging {
  log.info("Mollie client started")

  val connection: HttpServer = MollieConnection(config, context.system)

  def receive: Receive = {
    case cmd: CreatePayment =>
      commandClient forward cmd
    case qry: GetPayment =>
      queryClient forward qry
    case qry: ListPaymentMethods =>
      queryClient forward qry
    case qry: ListPaymentIssuers =>
      queryClient forward qry
  }

  private[this] def queryClient: ActorRef = {
    context.child(MollieQueryActor.name).getOrElse {
      context.actorOf(
        MollieQueryActor.props(
          connection = connection,
          config = config
        ).withRouter(RoundRobinPool(5)),
        MollieQueryActor.name
      )
    }
  }

  private[this] def commandClient: ActorRef = {
    context.child(MollieCommandActor.name).getOrElse {
      context.actorOf(
        MollieCommandActor.props(
          connection = connection,
          config = config
        ).withRouter(RoundRobinPool(5)),
        MollieCommandActor.name
      )
    }
  }

}

object MollieClientActor {

  final val name: String = "mollie"

  def props(config: MollieConfig): Props = Props(
    classOf[MollieClientActor],
    config
  )

} 

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

import akka.actor.{Props, ActorSystem}
import akka.routing.RoundRobinPool

import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

object ApplicationMain extends App {
  val system = ActorSystem("MyActorSystem")
  val concurrent = 100
  val reqpersec = 1000
  val microsecperreq = (1000 * 1000 / reqpersec).microsecond

  val resultActor = system.actorOf(Props(new ResultActor))
  val actor = system.actorOf(RoundRobinPool(concurrent).props(Props(new ProcessActor(resultActor))))

  system.scheduler.schedule(1.second, 2.second, resultActor, ResultActor.Stat())
  system.scheduler.schedule(1.second, microsecperreq, actor, ProcessActor.Action)

  system.awaitTermination()
} 

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

import akka.actor.{Actor, ActorLogging, Props}
import akka.routing.RoundRobinPool
import com.mongodb.MongoCommandException
import model.b2c.Flat
import play.api.{Configuration, Logger}
import repo.FlatRepo


class PersistActor (flatRepo: FlatRepo,
                    configuration: Configuration) extends Actor with ActorLogging {

  val notificationActor = {
    context.actorOf(RoundRobinPool(configuration.get[Int](PersistActor.notificationParallelActors))
      .props(Props(classOf[NotificationActor],configuration)), name = "notificationActor")
  }

  override def receive: Receive = {
    case flat: Flat => {
      try {
        val persistedFlat = flatRepo.addOrUpdateFlat(flat)
        def matchesFilter(flat: Flat): Boolean = {
          if (flat.price.get < 90000
            && flat.size.get >= 40
            && (flat.district.get == "centre" || flat.district.get == "teika" || flat.district.get == "agenskalns" )
          ) true
          else false
        }
        if (Flat.New == persistedFlat.status && matchesFilter(persistedFlat)) {
          Logger.info(s"New flat has been found $persistedFlat Sending out emails...")
          notificationActor ! persistedFlat
        }
      }
      catch {
        case e: MongoCommandException => {
          val message = e.getErrorMessage
          Logger.error(s"$flat flat insert/update has failed with $message, will retry")
          self ! flat
        }
      }
    }
  }
}

object PersistActor {
  val notificationParallelActors = "actor.system.parallel.actors.notification"
}