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

import akka.actor.{ActorLogging, PoisonPill, Actor, ReceiveTimeout}
import akka.cluster.sharding.ShardRegion.Passivate

trait Passivation extends ActorLogging {
  this: Actor =>

  protected def passivate(receive: Receive): Receive = receive.orElse {
    // tell parent actor to send us a poisinpill
    case ReceiveTimeout =>
      log.info(s" $self ReceiveTimeout: passivating. ")
      context.parent ! Passivate(stopMessage = PoisonPill)

    // stop
    case PoisonPill =>
      log.info(s" $self PoisonPill")
      context.stop(self)
  }
} 

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

import scala.concurrent.duration._
import akka.actor.Actor
import akka.actor.ActorIdentity
import akka.actor.ActorRef
import akka.actor.Identify
import akka.actor.ReceiveTimeout
import akka.actor.Terminated

class LookupActor(path: String) extends Actor {

  sendIdentifyRequest()

  def sendIdentifyRequest(): Unit = {
    context.actorSelection(path) ! Identify(path)
    import context.dispatcher
    context.system.scheduler.scheduleOnce(3.seconds, self, ReceiveTimeout)
  }

  def receive = identifying

  def identifying: Actor.Receive = {
    case ActorIdentity(`path`, Some(actor)) =>
      context.watch(actor)
      context.become(active(actor))
    case ActorIdentity(`path`, None) => println(s"Remote actor not available: $path")
    case ReceiveTimeout              => sendIdentifyRequest()
    case _                           => println("Not ready yet")
  }

  def active(actor: ActorRef): Actor.Receive = {
    case op: MathOp => actor ! op
    case result: MathResult => result match {
      case AddResult(n1, n2, r) =>
        printf("Add result: %d + %d = %d\n", n1, n2, r)
      case SubtractResult(n1, n2, r) =>
        printf("Sub result: %d - %d = %d\n", n1, n2, r)
    }
    case Terminated(`actor`) =>
      println("Calculator terminated")
      sendIdentifyRequest()
      context.become(identifying)
    case ReceiveTimeout =>
    // ignore

  }
} 

//设置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 org.remote.app.controller

import akka.actor.{Actor, ActorIdentity, ActorRef, Identify, Props, ReceiveTimeout, Terminated}
import org.remote.app.messaging.Action
import org.remote.app.sender.ManagerActor

import scala.concurrent.duration._

object ControllerActor {
  def props(remoteHost: String): Props = {
    Props(classOf[ControllerActor], remoteHost)
  }

  def actorName(host: String) = s"${host}controller"
}
class ControllerActor(remoteHost: String) extends Actor {

  val remoteActorPath = s"akka.tcp://${context.system.name}@$remoteHost/user/${ManagerActor.actorName}"

  def resolve(): Unit = {
    context.actorSelection(remoteActorPath) ! Identify(remoteActorPath)
    import context.dispatcher
    context.system.scheduler.scheduleOnce(10.seconds, self, ReceiveTimeout)
  }

  resolve()

  override def receive: Receive = identifying

  def identifying: Receive = {
    case ActorIdentity(`remoteActorPath`, Some(actor)) =>
      context.watch(actor)
      println(s"ActorIdentity was received from ${actor.path}")
      context become active(actor)
    case ActorIdentity(_, None) => println(s"Remote actor not available $remoteActorPath")
    case ReceiveTimeout         => resolve()
    case a                      => println(s"Not ready yet for $a")
  }

  def active(actor: ActorRef): Receive = {
    case ac: Action =>
      println(ac)
      actor ! ac
    case Terminated(`actor`) =>
      println(s"${actor.path} was terminated")
      resolve()
      context.become(identifying)
    case ReceiveTimeout => //Ignore
  }
} 

//设置package包名称以及导入依赖的类
package hmda.persistence.model

import java.util.concurrent.TimeUnit

import akka.actor.ReceiveTimeout
import akka.persistence.PersistentActor
import akka.stream.ActorMaterializer
import hmda.persistence.messages.CommonMessages.{ Event, Shutdown }
import hmda.persistence.PersistenceConfig._
import scala.concurrent.duration.Duration

abstract class HmdaPersistentActor extends PersistentActor with HmdaActor {

  implicit val system = context.system
  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()

  def updateState(event: Event): Unit

  override def preStart(): Unit = {
    super.preStart()
    val timeout = configuration.getInt("hmda.persistent-actor-timeout")
    context.setReceiveTimeout(Duration.create(timeout, TimeUnit.SECONDS))
  }

  override def receiveCommand: Receive = {
    case ReceiveTimeout =>
      self ! Shutdown

    case Shutdown =>
      context stop self
  }

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

} 

//设置package包名称以及导入依赖的类
package org.tanukkii.reactive.snowflake

import akka.actor.{Props, ReceiveTimeout}
import akka.cluster.sharding.ShardRegion
import akka.cluster.sharding.ShardRegion.Passivate

object ShardedIdRouterProtocol {
  case object StopSending
}

class ShardedIdRouter extends IdRouter {
  import ShardedIdRouterProtocol._

  override def unhandled(msg: Any) = msg match {
    case ReceiveTimeout =>
      context.parent ! Passivate(stopMessage = StopSending)
    case StopSending => context.stop(self)
    case other => super.unhandled(other)
  }
}

object ShardedIdRouter {
  import IdRouterProtocol._

  def props: Props = Props(new ShardedIdRouter)

  def name: String = "IdRouter"

  val shardName: String = "IdRouter"

  val idExtractor: ShardRegion.ExtractEntityId = {
    case [email protected](datacenterId, workerId) => (s"$datacenterId-$workerId", cmd)
  }

  val shardResolver: ShardRegion.ExtractShardId = {
    case [email protected](datacenterId, workerId) => s"${datacenterId % 32}-${workerId % 32}"
  }
} 

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

import akka.actor.{ActorLogging, Props, ReceiveTimeout}
import akka.cluster.sharding.ShardRegion
import akka.cluster.sharding.ShardRegion.Passivate
import akka.persistence.{PersistentActor, RecoveryCompleted}


class Borrower extends PersistentActor with ActorLogging {

  import Borrower._

  override def persistenceId: String = "borrower-" + self.path.name

  override def receiveRecover = {
    case e: BorrowerEvent => updateState(e)
    case RecoveryCompleted =>  }

  override def receiveCommand = {
    case CreateBorrower(catalogId, borrowerId, name) => persist(BorrowerCreated(catalogId, borrowerId, name))(e => {
      updateState(e)
      sender() ! e
    })
    case ReceiveTimeout ? context.parent ! Passivate(stopMessage = Stop)
    case Stop ? context.stop(self)
  }

  def updateState(e: BorrowerEvent) = e match {
    case i: BorrowerCreated =>
  }
}

object Borrower {
  sealed trait BorrowerCommand {
    val catalogId: Int
    val borrowerId: Int
  }

  final case class CreateBorrower(catalogId: Int, borrowerId: Int, name: String) extends BorrowerCommand

  sealed trait BorrowerEvent {
    val borrowerId: Int
  }

  final case class BorrowerCreated(catalogId: Int, borrowerId: Int, name: String) extends BorrowerEvent

  val extractEntityId: ShardRegion.ExtractEntityId = {
    case i: BorrowerCommand => (s"${i.catalogId}_${i.borrowerId}", i)
  }

  val numberOfShards = 128

  val extractShardId: ShardRegion.ExtractShardId = {
    case i: BorrowerCommand => math.abs(s"${i.catalogId}_${i.borrowerId}".hashCode % numberOfShards).toString
  }

  def props() = Props(new Borrower())
} 

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

import akka.actor.{Actor, ActorLogging, ActorRef, ReceiveTimeout}
import com.packt.chapter10.SocialMediaAggregator.{Report, StartFetching, StopFetching}

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

class SocialMediaStalker(aggregator: ActorRef, userId: String) extends Actor with ActorLogging {
  import context.dispatcher

  context.setReceiveTimeout(10 seconds)

  val counts = mutable.Map.empty[String, Int].withDefaultValue(0)

  override def preStart() = {
    log.info("Politely asking to aggregate")
    aggregator ! StartFetching(userId, 1 second)
    context.system.scheduler.scheduleOnce(5 second, aggregator, StopFetching(userId))
  }

  override def postStop() = {
    log.info(s"Stopping. Overall counts for $userId: $counts")
  }

  def receive = {
    case Report(list) =>
      val stats = list.groupBy(_.socialNetwork).mapValues(_.map(_.posts.size).sum)
      log.info(s"New report: $stats")
      stats.foreach(kv => counts += kv._1 -> (counts(kv._1) + kv._2))
    case ReceiveTimeout =>
      context.stop(self)
  }
} 

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

import scala.collection.immutable
import scala.concurrent.duration._
import akka.actor.ActorLogging
import akka.actor.Props
import akka.actor.PoisonPill
import akka.actor.ReceiveTimeout
import akka.cluster.sharding.ShardRegion
import akka.cluster.sharding.ShardRegion.Passivate
import akka.persistence.PersistentActor

object AuthorListing {

  def props(): Props = Props(new AuthorListing)

  case class PostSummary(author: String, postId: String, title: String)
  case class GetPosts(author: String)
  case class Posts(list: immutable.IndexedSeq[PostSummary])

  val idExtractor: ShardRegion.ExtractEntityId = {
    case s: PostSummary => (s.author, s)
    case m: GetPosts    => (m.author, m)
  }

  val shardResolver: ShardRegion.ExtractShardId = msg => msg match {
    case s: PostSummary   => (math.abs(s.author.hashCode) % 100).toString
    case GetPosts(author) => (math.abs(author.hashCode) % 100).toString
  }

  val shardName: String = "AuthorListing"
}

class AuthorListing extends PersistentActor with ActorLogging {
  import AuthorListing._

  override def persistenceId: String = self.path.parent.name + "-" + self.path.name

  // passivate the entity when no activity
  context.setReceiveTimeout(2.minutes)

  var posts = Vector.empty[PostSummary]

  def receiveCommand = {
    case s: PostSummary =>
      persist(s) { evt =>
        posts :+= evt
        log.info("Post added to {}'s list: {}", s.author, s.title)
      }
    case GetPosts(_) =>
      sender() ! Posts(posts)
    case ReceiveTimeout => context.parent ! Passivate(stopMessage = PoisonPill)
  }

  override def receiveRecover: Receive = {
    case evt: PostSummary => posts :+= evt

  }

} 

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

import akka.actor.{Actor, ActorLogging, PoisonPill, ReceiveTimeout}
import akka.cluster.sharding.ShardRegion.Passivate
import akka.persistence.{SaveSnapshotFailure, SaveSnapshotSuccess}

trait ActorHelpers extends ActorLogging {
  this: Actor =>

  protected def utils(receive: Receive): Receive = receive.orElse {
    case evt: SaveSnapshotSuccess =>
      log.debug("Snapshot success!")
    case evt: SaveSnapshotFailure =>
      log.error("Snapshot failure!")
    case ReceiveTimeout =>
      log.debug("{} received timeout, passivating...", self.path.name)
      context.parent ! Passivate(stopMessage = PoisonPill)
    case Passivate(stopMessage) =>
      sender ! stopMessage
  }

  protected def unknownMessage: Receive = {
    case msg: Any =>
      log.warning("{} received unknown message {}", self.path.name, msg)
  }
} 

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

import akka.actor.{Actor, ReceiveTimeout}
import com.dvisagie.vote.injector.Provider

import scala.concurrent.duration._

class LoginActor(implicit provider: Provider) extends Actor {
  import LoginActor._
  context.setReceiveTimeout(5.seconds)

  def receive: Receive = {
    case request: LoginRequest =>
      sender() ! LoginResponse
      context stop self
    case ReceiveTimeout =>
      context stop self
  }

}
object LoginActor {
    case class LoginRequest(username: String, password: String)
    case class LoginResponse(token: String)
} 

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

import akka.actor.{Actor, ReceiveTimeout}
import com.dvisagie.vote.injector.Provider

import scala.concurrent.duration._

class RegistrationActor(implicit provider: Provider) extends Actor {
  import RegistrationActor._
  context.setReceiveTimeout(5.seconds)


  def receive: Receive = {
    case request: UserRegistrationRequest =>
      sender() ! Unit
      context stop self
    case ReceiveTimeout =>
      context stop self
  }
}
object RegistrationActor {
  case class UserRegistrationRequest(username: String, email: String, password: String)
} 

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

import java.util.UUID

import akka.actor.{Actor, ReceiveTimeout}
import com.dvisagie.vote.injector.Provider
import com.dvisagie.vote.repositories.UserRepository

import scala.concurrent.duration._

class UserControllerActor(implicit provider: Provider) extends Actor {
  import UserControllerActor._
  context.setReceiveTimeout(5.seconds)

  val userRepository: UserRepository = provider.userRepository

  def receive: Receive = {
    case createUserRequest: CreateUserRequest =>
      sender() ! CreationRequestResponse("message received", createUserRequest.username)
      context stop self
    case id: UUID =>
      sender() ! userRepository.getUserForId(id)
      context stop self
    case username: String =>
      sender() ! userRepository.getUserForUsername(username)
      context stop self
    case ReceiveTimeout =>
      context stop self
  }
}

object UserControllerActor {

  final case class CreateUserRequest(
    username: String,
    firstNames: String,
    lastName: String,
    email: String)

  final case class CreationRequestResponse(
    message: String,
    username: String)

  final case class UserResponse(
    username: String,
    firstNames: String,
    lastNames: String)
} 

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

import akka.actor.{Actor, ActorRef, Props, ReceiveTimeout}
import akka.remote.WireFormats.FiniteDuration

import scala.collection.mutable
import communication.{Ping, Request}

import scala.concurrent.Await
import scala.concurrent.duration._
import akka.pattern.ask

import scala.util.{Failure, Try}


private class NodeController(nodeMap: mutable.HashMap[String, ActorRef]) extends Actor {

  implicit val timeout = akka.util.Timeout(30 seconds)

  context.setReceiveTimeout(timeout.duration)

  override def receive: Receive = {
    case Request(nodeName: String) => {
      println(s"pinged by $nodeName")
      if (!nodeMap.keys.exists(_ == nodeName)) {
        nodeMap.put(nodeName, sender)
        println(s"$nodeName registered")
      } else {
        println(s"$nodeName already exists!")
      }
    }
    case ReceiveTimeout => {
      nodeMap.par.foreach { case (name, nodeRef) =>
        Try(Await.result(nodeRef.ask(Ping()), 1 second)) match {
          case Failure(_) => nodeMap.remove(name)
          case _ => ()
        }
      }
      context.setReceiveTimeout(timeout.duration)
    }
    case _ => println("Sth else has happenned")
  }
}


object NodeController {
  def props(nodeMap: mutable.HashMap[String, ActorRef]) = Props(new NodeController(nodeMap))
} 

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

import akka.actor.{PoisonPill, Actor, ReceiveTimeout}
import com.boldradius.cqrs.AuctionCommandQueryProtocol.InvalidAuctionAck
import com.boldradius.util.Logging
import akka.contrib.pattern.ShardRegion.Passivate

trait Passivation extends Logging {
  this: Actor =>

  protected def passivate(receive: Receive): Receive = receive.orElse{
    // tell parent actor to send us a poisinpill
    case ReceiveTimeout =>
      self.logDebug("ReceiveTimeout: passivating. " + _.toString)
      context.parent ! Passivate(stopMessage = PoisonPill)

    // stop
    case PoisonPill => context.stop(self.logDebug("PoisonPill" + _.toString))
  }
} 

//设置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 utils.advert

import akka.actor.{ReceiveTimeout, Actor}
import anorm._
import play.api.Play.current
import play.api.db.DB
import utils._

import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.{Failure, Try}

class AccountManager extends Actor {
  override def receive = {
    case "Start" =>
      context.setReceiveTimeout(30 minutes)
    case ReceiveTimeout =>
      AccessLogger.debug("Begin to manage advertisers accounts periodically")

      Try(DB.withTransaction { implicit c =>
        SQL("select advert.manage_accounts(NULL)").execute()
      }) match {
        case Failure(ex) => AccessLogger.error(s"Fail to manage advertisers accounts periodically: ${ex.getMessage}")
        case _ => AccessLogger.debug("Managing advertisers accounts completes once")
      }
    case "Stop" =>
      context.setReceiveTimeout(Duration.Undefined)
  }
} 

//设置package包名称以及导入依赖的类
package com.github.j5ik2o.spetstore.adaptor.aggregate

import akka.actor.{ ReceiveTimeout, Actor }
import akka.cluster.sharding.ShardRegion.Passivate

object PassivationSupportProtocol {
  case object StopWriting
}

trait PassivationSupport extends Actor {
  import PassivationSupportProtocol._

  override def unhandled(msg: Any): Unit = msg match {
    case ReceiveTimeout =>
      context.parent ! Passivate(stopMessage = StopWriting)
    case StopWriting => context.stop(self)
    case other       => super.unhandled(other)
  }
} 

//设置package包名称以及导入依赖的类
package org.alcaudon.runtime

import java.io.File
import java.net.URI

import akka.actor.{Actor, ActorLogging, ReceiveTimeout}
import com.amazonaws.auth.BasicAWSCredentials
import org.alcaudon.core.ActorConfig
import org.alcaudon.runtime.BlobLocation.AWSInformation

import scala.util.{Failure, Success}

object BlobDownloader {
  case class DownloadBlob(uri: URI, file: File)
  case class DownloadFinished(uuid: String, file: File)
  case class DownloadFailed(uuid: String, reason: Throwable)
  case class DownloadTimeout(msg: String) extends Throwable
}

class BlobDownloader(uuid: String)
    extends Actor
    with ActorLogging
    with ActorConfig {
  import BlobDownloader._

  val downloadTimeout = config.blob.downloadTimeout
  val awsCredentials =
    new BasicAWSCredentials(config.blob.s3.accessKey, config.blob.s3.secretKey)
  implicit val awsInfo =
    AWSInformation(config.blob.s3.region, awsCredentials)

  context.setReceiveTimeout(downloadTimeout)

  def receive = {
    case DownloadBlob(uri: URI, file: File) =>
      log.debug("Download {} to {}", uri, file)
      BlobLocation(uri).download(file) match {
        case Success(path) =>
          sender() ! DownloadFinished(uuid, file)
        case Failure(reason) =>
          sender() ! DownloadFailed(uuid, reason)
      }
      context.stop(self)
    case ReceiveTimeout =>
      log.warning("Timeout waiting for download of {}", uuid)
      sender() ! DownloadFailed(
        uuid,
        DownloadTimeout(s"Timeout downloading $uuid, after $downloadTimeout"))
      context.stop(self)
  }
} 

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

import akka.actor.{Actor, ActorLogging, ActorRef, Props, ReceiveTimeout}

import scala.concurrent.duration._

class SessionActor(sessionId: Long, statsActor: ActorRef, chatActorManager: ActorRef) extends Actor with ActorLogging {

  var requestsHistory = Seq.empty[Request]

  override def receive: Receive = receiveStateHelp(onHelp = false)

  def receiveStateHelp(onHelp: Boolean): Receive = {
    case [email protected](_, _, path, _, _) =>
      requestsHistory = r +: requestsHistory

      // Handle Help chat session
      path match {
        case "/help" if !onHelp =>
          context.become(receiveStateHelp(onHelp = true))
          context.setReceiveTimeout(2 seconds)
        case _ =>
          resetState()
      }
    case ReceiveTimeout if onHelp =>
      chatActorManager ! StartChat(sessionId)
      resetState()
    case ReceiveTimeout =>
      //log.debug("Receive timeout : End of session -> sending stats")
      sendStats()
      //context.stop(self)
    case EOS =>
      log.info("EOS : End of session -> sending stats")
      sendStats()
      //context.stop(self)
  }

  override def postStop(): Unit = {
    log.info(s"SessionActor has been stopped for {}", sessionId)
  }

  private def sendStats() =
    statsActor ! SessionStats(sessionId, requestsHistory.reverse)

  private def resetState() = {
    context.become(receiveStateHelp(onHelp = false))
    context.setReceiveTimeout(5 seconds)
  }

}

object SessionActor {

  def props(sessionId: Long, statsActor: ActorRef, chatActorManager: ActorRef): Props = Props(classOf[SessionActor], sessionId, statsActor, chatActorManager)

}

final case class SessionStats(sessionId: Long, requestsHistory: Seq[Request])