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

import language.existentials

import akka.actor.Actor.Receive
import akka.actor.ActorContext
import akka.actor.ActorCell
import akka.event.Logging.Debug

object LoggingReceive {

  
class LoggingReceive(source: Option[AnyRef], r: Receive, label: Option[String])(implicit context: ActorContext) extends Receive {
  def this(source: Option[AnyRef], r: Receive)(implicit context: ActorContext) = this(source, r, None)
  def isDefinedAt(o: Any): Boolean = {
    val handled = r.isDefinedAt(o)
    val (str, clazz) = LogSource.fromAnyRef(source getOrElse context.asInstanceOf[ActorCell].actor)
    context.system.eventStream.publish(Debug(str, clazz, "received " + (if (handled) "handled" else "unhandled") + " message " + o
      + (label match {
        case Some(l) ? " in state " + l
        case _       ? ""
      })))
    handled
  }
  def apply(o: Any): Unit = r(o)
} 

//设置package包名称以及导入依赖的类
package org.dele.misc.learnAkka

import akka.actor.Actor.Receive
import akka.actor.{Actor, ActorSystem, Props}


object Hello extends App {
  case class Greeting(greeting:String)
  case class Greet(name:String)
  class HelloActor extends Actor {
    var _greeting = ""
    override def receive: Receive = {
      case Greeting(greeting) => _greeting = greeting
      case name: String => println(s"${_greeting} $name")
    }
  }
  val system = ActorSystem("actor-demo-1")

  val hello = system.actorOf(Props[HelloActor])

  hello ! "bob"
  hello ! Greeting("Hello")
  hello ! "bob"
  hello ! "alice"
  hello ! Greeting("Halo")
  hello ! "bob"
  hello ! "alice"

  Thread.sleep(1000)

  system.terminate()
} 

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

import java.net.InetSocketAddress

import akka.actor.Actor.Receive
import akka.actor.ActorRef
import akka.io.Tcp.Received
import akka.testkit.{TestActorRef, TestActor}
import akka.util.ByteString
import com.init6.{ViLeNetActor, ViLeNetTestComponent}
import org.scalatest.FlatSpec

import scala.collection.mutable.ArrayBuffer




class TelnetMessageHandlerSpec extends FlatSpec {

  "telnet message handler" should "handle split packets" in {

    readTest(ByteString("hello\r\nhow are you"))
    readTest(ByteString("\r\nhello\r\n"))
  }

  var buffer = ArrayBuffer[Byte]()

  def readTest(data: ByteString): ByteString = {
    val readData = data.takeWhile(b => b != '\r' && b != '\n')
    if (data.length == readData.length) {
      // Split packet
      buffer ++= readData
    } else {
      // End of packet found
      if (buffer.nonEmpty) {
        //println(s"1 ${ByteString(buffer.toArray[Byte] ++ readData.toArray[Byte])}")
        buffer.clear()
      } else if (readData.nonEmpty) {
        //println(s"2 $readData")
      }
    }
    val restOfData = data.drop(readData.length).dropWhile(b => b == '\r' || b == '\n')
    if (restOfData.nonEmpty) {
      readTest(restOfData)
    } else {
      ByteString.empty
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.dazito.scala.akkademy.client

import akka.actor.Actor.Receive
import akka.actor.{Actor, Stash}
import com.dazito.scala.dakkabase.messages.{Disconnected, Connected, GetRequest}


class RemoteActorProxy extends Actor with Stash {

    var isOnline = false;

    override def receive: Receive = {
        case message: GetRequest =>
            if(isOnline) {
                processMessage(message)
            }
            else {
                stash()
            }
        case _: Connected =>
            isOnline = true
            unstashAll()
        case _: Disconnected =>
            isOnline = false;
    }

    def processMessage(message: AnyRef) = {
        // Do something
    }
} 

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

import akka.actor.Actor.Receive
import akka.actor._
import com.typesafe.config.ConfigFactory
import spray.routing.{HttpService, HttpServiceActor}

object boxOffice {

  class BoxOffice extends Actor {
    def receive = {
      case _ =>
    }
  }

  trait BoxOfficeCreator {
    this: Actor =>
    def createBoxOffice: ActorRef = {
      context.actorOf(Props[BoxOffice], "boxOffice")
    }
  }

  trait RestApi extends HttpService with ActorLogging with BoxOfficeCreator {
    actor: Actor =>
    val boxOffice = createBoxOffice
  }

  class RestInterface extends HttpServiceActor with RestApi {
    override def receive: Receive = ???
  }

  val config = ConfigFactory.load()
  val system = ActorSystem("singlenode", config.getConfig("remote-single"))

  val restInterface = system.actorOf(Props[RestInterface], "restInterface")

} 

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

import akka.actor.{Actor, ActorRef, Props}
import akka.actor.Actor.Receive
import com.udemy.akka.router.Worker.Work


class RouterActor extends Actor{
  var routees:List[ActorRef] = _

  override def preStart() = {
    routees=List.fill(5)(
      context.actorOf(Props[Worker])
    )
  }
  override def receive: Receive = {
    case msg:Work=>
      println("I am a router and I received a message .. ")
      routees(util.Random.nextInt(routees.size)) forward msg
  }
}

class RouterGroup(routees:List[String]) extends Actor{
  override def receive: Receive = {
    case msg:Work=>{
      println("I am a router group and I received Work message")
      context.actorSelection(routees(util.Random.nextInt(routees.size))) forward msg
    }
  }
} 

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

import akka.actor.Actor
import akka.actor.Actor.Receive
import com.udemy.akka.actorpath.Counter.{Dec, Inc}


class Counter extends Actor{
  var count=0
  override def receive: Receive = {
    case Inc(x)=>count += x
    case Dec(x)=>count -= x
  }
}

object Counter {
  final case class Inc(num:Int)
  final case class Dec(num:Int)
} 

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

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import akka.actor.Actor.Receive
import com.udemy.akka.MusicController.{Play, Stop}
import com.udemy.akka.MusicPlayer.{StartMusic, StopMusic}



object MusicPlayer {
  sealed trait PlayMsg
  case object StopMusic extends PlayMsg
  case object StartMusic extends PlayMsg

}

class MusicPlayer extends Actor{
  override def receive: Receive = {
    case StopMusic => println("I don't want to stop music ....")
    case StartMusic =>
      val controller= context.actorOf(MusicController.props,"controller")
      controller ! Play
    case _ => println("Unknown")
  }

}

object MusicController {
  sealed trait ControllerMsg
  case object Play extends ControllerMsg
  case object Stop extends ControllerMsg
  def props=Props[MusicController]
}

class MusicController extends Actor{
  override def receive: Receive = {
    case Play=> println("Playing music .... ")
    case Stop => println("Stop music .... ")
  }
}
object ActorCreation extends App{
  val system=ActorSystem("CreationSystem")
  private val player: ActorRef = system.actorOf(Props[MusicPlayer],"player")
  player ! StartMusic
  system.terminate()
} 

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

import akka.actor.{Actor, ActorRef, ActorSystem, Props, Terminated}
import akka.actor.Actor.Receive




class Ares(athena:ActorRef) extends Actor{
  override def preStart(): Unit = {
    context.watch(athena)
  }

  override def postStop(): Unit = {
    println("Ares post stop")
  }
  override def receive: Receive = {
    case Terminated=>{

      context.stop(self)
    }
  }
}
class Athena extends Actor{
  override def receive: Receive = {
    case msg=>
      println(s"Athena recieved ${msg}")
      context.stop(self)
  }
}


object Monitor extends App{
  val system =ActorSystem("MonitorSystem")


  private val athena: ActorRef = system.actorOf(Props[Athena],"athena")
  private val ares: ActorRef = system.actorOf(Props(new Ares(athena)),"ares")
  athena ! "hello"
  system.terminate()
} 

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

import akka.actor.{Actor, Stash}
import akka.actor.Actor.Receive
import com.udemy.akka.become.UserStorage.{Connect, Disconnect, Operation}


case class User(username:String,email:String)

object UserStorage{
  trait DBOperation
  object DBOperation{
    case object Create extends DBOperation
    case object Read extends DBOperation
    case object Update extends DBOperation
    case object Delete extends DBOperation
  }

  case object Connect
  case object Disconnect

  case class Operation(operation:DBOperation,user:Option[User])
}

class UserStorage extends Actor with Stash{
  override def receive: Receive = disconnected

  def connected:Actor.Receive= {
    case Disconnect=>
      println("User storage disconnected from DB")
      unstashAll()
      context.unbecome()
    case Operation(op,user)=>
      println(s"User storage recieved $op to do in user : $user")
  }

  def disconnected:Actor.Receive = {
    case Connect =>
      println("User storage connected to DB")
      unstashAll()
      context.become(connected)
    case _ =>
      stash()
  }


} 

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

import akka.actor.{Actor, Status}
import akka.actor.Actor.Receive
import akka.event.Logging
import org.atuwadb.messages.{GetRequest, InvalidMessageTypeException, KeyNotFoundException, SetRequest}

import scala.collection.mutable


class AtuwaDb extends Actor {
  val map = new mutable.HashMap[String, Object]
  val log = Logging(context.system, this)

  override def receive: Receive = {
    case SetRequest(key, value) => {
      log.info("Received SetRequest - key: {} value: {}", key, value)
      map.put(key, value)
      sender() ! Status.Success
    }
    case GetRequest(key) => {
      log.info("Received GetRequest - key: {}", key)
      val response: Option[Object] = map.get(key)
      response match {
        case Some(x) => sender() ! x
        case None => sender() ! Status.Failure(new KeyNotFoundException(key))
      }
    }
    case o => sender() ! Status.Failure(new InvalidMessageTypeException("Received an invalid message: " + o))
  }
} 

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

import akka.actor.Actor.Receive
import akka.actor.{Actor, ActorRef, Props}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import play.api.libs.json.Json
import services.StreamActor.TopHashtag
import twitter4j.Status


class StreamActor[T] (val out : ActorRef, val stream : DStream[T]) extends Actor {

  stream.foreachRDD { rdd =>
    rdd.take(5).foreach { element =>
      out ! element.toString
    }
  }

  override def receive: Receive = {
    case msg: String => out ! "Ok"
    case TopHashtag(top) => out ! Json.toJson(top)
  }

  @scala.throws[Exception](classOf[Exception])
  override def postStop(): Unit = {

  }
}

object StreamActor {
  def props[T](out: ActorRef, stream : DStream[T]) = Props(new StreamActor(out, stream))

  case class TopHashtag(top : Map[String, Int])
} 

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

import java.util.concurrent.TimeUnit

import akka.actor.{Actor, Props}
import akka.actor.Actor.Receive

case class ProcessNumber(num: Int)

object ProcessCounter {
  val props = Props[ProcessCounter]
}

class ProcessCounter extends Actor{
  override def receive: Receive = {
    case ProcessNumber(num) =>
      println(s"Processing $num")
      TimeUnit.SECONDS.sleep(1)
  }
} 

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

import akka.actor.Actor.Receive
import akka.actor.{Actor, ActorLogging, DeadLetter, Props}


object SceneLightStart {
  def main(args: Array[String]): Unit = {
    println("hello world")
    akka.Main.main(Array(classOf[Master].getName))
  }
}

class Master extends Actor {
  override def preStart(): Unit = {
    val slave = context.actorOf(Props[Slave], "slave")
    slave ! Work
    val noone = context.actorOf(Props[DeadLetterReader], "noone")
    noone ! Unknown
  }
  def receive = {
    case Done => context.stop(self)
  }
}

class Slave extends Actor with ActorLogging {
  override def receive = {
    case Work =>
      log.info("I hate my Master!")
      sender() ! Done
  }
}

case object Work

case object Done

case object Unknown

class DeadLetterReader extends Actor with ActorLogging{
  override def preStart(): Unit = context.system.eventStream.subscribe(self, classOf[DeadLetter])
  override def receive: Receive = {
    case Unknown => log.info("Received unknown message")
  }
} 

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

import akka.actor.Actor.Receive
import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
import akka.cluster.pubsub.DistributedPubSub
import com.google.inject.Inject
import common.BarEvents
import play.api.libs.json.Json
import play.api.mvc._
import akka.pattern._
import akka.util.Timeout

import scala.concurrent.ExecutionContext
import scala.concurrent.duration._

class MainController @Inject()(system: ActorSystem)(implicit ec: ExecutionContext)
    extends Controller {

  val cache            = system.actorOf(Props(new Cache()))
  val subscriber       = system.actorOf(Props(new Subscriber(BarEvents.KEY, cache)))
  implicit val timeout = Timeout(5.seconds)

  def list() = Action.async {
    val currently = (cache ? "get").mapTo[Seq[Int]]
    currently.map(x => Json.toJson(x)).map(x => Ok(x))
  }
}

class Subscriber(topic: String, forwardTo: ActorRef) extends Actor with ActorLogging {
  import akka.cluster.pubsub.DistributedPubSubMediator._
  val mediator = DistributedPubSub(context.system).mediator
  mediator ! Subscribe(BarEvents.KEY, self)

  def receive = {
    case ack: SubscribeAck =>
      log.warning("subscribing {}", ack)
    case x =>
      log.warning("forwarding {}", x)
      forwardTo ! x
  }
}

class Cache extends Actor with ActorLogging {
  var cached = Seq.empty[Int]

  def receive: Receive = {
    case "get" =>
      log.warning("getting cache")
      sender ! cached

    case entities: Seq[Int] =>
      log.warning(s"setting cache to $entities")
      cached = entities
  }
} 

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

import actors.StatsActor.{GetStats, RequestReceived, Ping}
import akka.actor.Actor
import akka.actor.Actor.Receive


class StatsActor extends Actor {
  var counter = 0

  override def receive: Receive = {
    case Ping => ()
    case RequestReceived => counter += 1
    case GetStats => sender() ! counter
  }
}

object StatsActor {
  val name = "statsActor"
  val path = s"/user/$name"

  case object Ping
  case object RequestReceived
  case object GetStats
} 

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

import akka.actor.Actor.Receive
import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import com.example.SyncTriggerActor.SyncTrigger

import scala.concurrent.duration._

class SyncTriggerActor(actor: ActorRef) extends Actor with ActorLogging {


  case class TimerTick()

  override def preStart: Unit = {
    super.preStart
    implicit val ec = context.dispatcher
    context.system.scheduler.schedule(1 second, 1 minute, self, new TimerTick())
  }

  override def receive: Receive = {
    case tick:TimerTick =>
      log.info("received timer tick")
      actor ! new SyncTrigger()
  }
}

object SyncTriggerActor {
  def props(actor: ActorRef) = Props(new SyncTriggerActor(actor))
  case class SyncTrigger()
} 

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

import akka.actor._
import akka.actor.Actor.Receive
import com.sun.xml.internal.ws.policy.privateutil.PolicyUtils.ConfigFile
import com.typesafe.config.ConfigFactory

import scala.util.Random


class Frontend extends Actor {

  var backends = IndexedSeq.empty[ActorRef]

  override def receive: Receive = {
    case Add if backends.isEmpty =>
      println(s"Serviec unavailable, cluster doesn't have backend node.")
    case addOp: Add =>
      println(s"Frontend: I will forward add operation to backend node to handle it")
      backends(Random.nextInt(backends.size)) forward(addOp)
    case BackendRegistration if !(backends.contains(sender())) =>
      backends = backends :+ sender()
      context watch(sender())

    case Terminated(a) =>
      backends = backends.filterNot(_ == a)
  }
}

object Frontend {
  private var _frontend: ActorRef = _

  def initiate() = {
    val config = ConfigFactory.load().getConfig("Frontend")
    val system = ActorSystem("ClusterSystem", config)

    _frontend = system.actorOf(Props[Frontend], name = "frontend")
  }

  def getFrontend = _frontend

} 

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

import akka.actor.{Props, ActorSystem, RootActorPath, Actor}
import akka.actor.Actor.Receive
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.MemberUp
import com.typesafe.config.ConfigFactory


class Backend extends Actor {
  val cluster = Cluster(context.system)

  override def preStart(): Unit = {
    cluster.subscribe(self, classOf[MemberUp])

  }

  override def postStop() = {
    cluster.unsubscribe(self)
  }

  override def receive: Receive = {
    case Add(num1, num2) =>
      println(s"============== I'm a backend with path ${self} and I received add operation ===============")
    case MemberUp(member) =>
      if (member.hasRole("frontend")) {
        context.actorSelection(RootActorPath(member.address) / "user" / "frontend") ! BackendRegistration
      }
  }
}

object Backend {
  def initiate(port: Int) = {
    val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port").
      withFallback(ConfigFactory.load.getConfig("Backend"))

    val system = ActorSystem("ClusterSystem", config)

    val Backend = system.actorOf(Props[Backend], name = "Backend")

  }
}

case class Add(num1: Int, num2: Int)
case object BackendRegistration 

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

import language.existentials
import akka.actor.Actor.Receive
import akka.actor.ActorContext
import akka.actor.ActorCell
import akka.actor.DiagnosticActorLogging
import akka.event.Logging.{ LogEvent, LogLevel }
import akka.actor.AbstractActor
import scala.runtime.BoxedUnit

object LoggingReceive {

  
class LoggingReceive(source: Option[AnyRef], r: Receive, label: Option[String], logLevel: LogLevel)(implicit context: ActorContext) extends Receive {
  def this(source: Option[AnyRef], r: Receive, label: Option[String])(implicit context: ActorContext) = this(source, r, label, Logging.DebugLevel)
  def this(source: Option[AnyRef], r: Receive)(implicit context: ActorContext) = this(source, r, None, Logging.DebugLevel)
  def isDefinedAt(o: Any): Boolean = {
    val handled = r.isDefinedAt(o)
    if (context.system.eventStream.logLevel >= logLevel) {
      val src = source getOrElse context.asInstanceOf[ActorCell].actor
      val (str, clazz) = LogSource.fromAnyRef(src)
      val message = "received " + (if (handled) "handled" else "unhandled") + " message " + o + " from " + context.sender() +
        (label match {
          case Some(l) ? " in state " + l
          case _ ? ""
        })
      val event = src match {
        case a: DiagnosticActorLogging ? LogEvent(logLevel, str, clazz, message, a.log.mdc)
        case _ ? LogEvent(logLevel, str, clazz, message)
      }
      context.system.eventStream.publish(event)
    }
    handled
  }
  def apply(o: Any): Unit = r(o)
}