class MyActor extends Actor {
    def receive = {
    
    }
}

In Scala, the receive block is actually a partial function, which allows the usage of pattern matching syntax.

Creating actors

Actor with default constructor

使用actorOf创建actor，两个参数Props中是actor class对象，name是actor name

val system = ActorSystem("MyActorSystem")
val myActor = system.actorOf(Props[MyActor], name = "myActor")

To create an Actor, we created the ActorSystem class and invoked the actorOf() method on the same. 
The actorOf() method accepted two arguments—first was the Props object and second was the actor's name passed as a String object. 
The Props object accepted the Actor class object, which needed to instantiated and started. 
The actor is started after the object has been instantiated.

The instantiated actor is held using ActorRef. ActorRef provides an immutable and serial able handle to the underlying actor.

Actor with non-default constructor

class MyActor(initialise:Int) extends Actor {
    def receive = {
    }
}
val system = ActorSystem("MyActorSystem")
val myActor = system.actorOf(Props(new MyActor(10)), name = "myActor") //call-by-name block

Creating an actor within an actor hierarchy

class SupervisorActor extends Actor {
    val myWorkerActor = context.actorOf(Props[MyWorkerActor],"myWorkerActor")
}

In this case, within the Actor object, we created the child using the parent context.

Messaging model

Fire and forget messages – tell()

单向发送

actor ! msg
//or
actor.tell(msg)
//or
actor tell msg

两个参数，一个是msg, 一个是sender actor's reference 
By default, the caller's actor reference is passed implicitly if nothing is specified, 第二个参数是implicitly包含sender的, 用于receiver的reply-back

DeadLetterActor用于接受所有发送给dead或不存在的actor的message 
But if the caller is not an actor or if the actor that invoked no longer exists, then the reply is sent to the dead letter actor. 
In Akka, DeadLetterActorRef is the default implementation of the dead letters' service, where all messages are rerouted whose callers are shut down or nonexistent.

当然你也可以传入一个其他的actor

//explicit passing of another actor reference
actor.tell(msg, anotherActorRef)

Send and receive messages – ask()

the message is sent asynchronously and future is returned, which represents a potential reply. 
In the case of send and receive message mode, actors use the ask() method for sending a message and wait on future for the reply.

val result: Future[Order]　= ask(orderActor, userId).mapTo[Order] //用userId去orderActor获取这个user的order

Replying to messages

One of the key aspects of communicating via messages is the ability to reply back to the actors. 
When an actor receives a message and a reply is expected back, the actor can make use of the sender available within the actor to reply back.

def receive = {
    case message:String =>
        sender ! (message + "world") //从message中获取sender
}

In case, no sender is specified or available, the reply is sent back to the DeadLetterActorRef.

Forwarding messages

Forward的不同在于会保留original sender reference 
When writing an actor that provides functionality such as routing or load balancing or replication, the incoming messages will get forwarded to the target actors. In this case, it is very important that the original sender reference is maintained and passed on to the target actors. This makes sure that the replies go to the original sender and not the mediator actor.

actor.forward(message)

Stopping actors

Actor termination involves multiple steps. Once the STOP signal is received by the actor, the following actions take place: 
1. Actor stops processing the mailbox messages. 
2. Actor sends the STOP signal to all the children. 
3. Actor waits for termination message from all its children. 
4. Next, Actor starts the self-termination process that involves the following: 
    -Invoking the postStop() method 
    -Dumping the attached mailbox 
    -Publishing the terminated message on DeathWatch 
    -Informing the supervisor about self-termination

用以下方法去stop actor：

//first option of shutting down the actors by shutting down
//the ActorSystem
system.shutdown()
//second option of shutting down the actor by sending a
//poisonPill message
actor ! PoisonPill
//third option of shutting down itself
context.stop(self)
//or stop the child actors
context.stop(childActorRef)

Killing actors 
An actor can be killed when a kill() message is sent to it. Unlike PoisonPill, which is an asynchronous way to shut down the actor, kill() is a synchronous 
way. The killed actor sends ActorKilledException to its parent.

actor ! Kill

HotSwap

其实就是可以动态的切换，或暂时改变，消息处理逻辑

Another key functionality of Akka is the ability to HotSwap an actor's message loop functionality at runtime. 
The functionality is provided via the getContext().become() and getContext().unbecome() methods. 
The HotSwapped code is kept in a stack, which can be pushed and popped when the become() or unbecome() methods are invoked.

case class PING
case class PONG
class PingPongActor extends Actor {
    import context._
    var count = 0
    def receive: Receive = {
            case PING =>
                println("PING")
                count = count + 1
                Thread.sleep(100)
                self ! PONG
                    become {  //切换到pong的逻辑，ping的逻辑会暂时放在stack里面
                        case PONG =>
                        println("PONG")
                        count = count + 1
                        Thread.sleep(100)
                        self ! PING
                        unbecome() //恢复ping的逻辑
                    }
        if(count > 10) context.stop(self)
    }
}

虽然我觉得这个例子没必要一定要这样实现，但反应出message loop functionality可以runtime的在ping和pong直接切换

//定义actor CalculatorInt的接口
//有4个method, 对于actor可以理解为, 4种message
trait CalculatorInt {
    def add(first: Int, second: Int): Future[Int] //异步方式, 类似ask
    def subtract(first: Int, second: Int): Future[Int]
    def incrementCount(): Unit  //类似tell,不需要返回
    def incrementAndReturn(): Option[Int] //同步方式,会blocking等待
}

//实现actor CalculatorInt 

class Calculator extends CalculatorInt with PreStart with PostStop {
    var counter: Int = 0
    import TypedActor.dispatcher
    def add(first: Int, second: Int): Future[Int] =
        Promise successful first + second
    def subtract(first: Int, second: Int): Future[Int] =
        Promise successful first - second
    def incrementCount(): Unit = counter += 1
    def incrementAndReturn(): Option[Int] = {
        counter += 1
        Some(counter)
    }

    def preStart(): Unit = { //初始化
        log.info ("Actor Started")
    }
    def postStop(): Unit = {
        log.info ("Actor Stopped")
    }

}

//创建TypedActor
val _system = ActorSystem("TypedActorsExample")
val calculator: CalculatorInt = TypedActor(_system).typedActorOf(TypedProps[Calculator]())

//Fire and forget, 类似tell
calculator.incrementCount()
//Send and receive
val future = calculator.add(14,6);
val result = Await.result(future, 5 second);
//Method invocation in a blocking way
val result = calculator.incrementAndReturn()

//To shut down the typed actor
TypedActor(system).stop(calculator)
//Other way to stop the actor is invoke the Poisonpill method
TypedActor(system).poisonPill(calculator)

val router = system.actorOf(Props[MyActor].withRouter(RoundRobinRouter (nrOfInstances = 5)) , name = "myRouterActor")

//创建remote actor的列表
val addresses = Seq( Address("akka", "remotesys", "host1", 1234),Address("akka", "remotesys", "host2", 1234))
val routerRemote = system.actorOf(Props[MyEchoActor].withRouter(RemoteRouterConfig(RoundRobinRouter(5), addresses)))

//Work Actor的定义,提高对state的更改和查询
case class Result
class WorkerActor extends Actor with ActorLogging {
    var state: Int = 0 //actor需要维护的状态
            
    override def preStart() {
        log.info("Starting WorkerActor instance hashcode # {}", this.hashCode())
    }
    override def postStop() {
        log.info("Stopping WorkerActor instance hashcode # {}", this.hashCode())
    }
    
    //对于正常的msg外,会抛出各种异常 
    def receive: Receive = {
        case value: Int =>
            if (value <= 0) //小于等于0的时候,抛出ArithmeticException
                throw new ArithmeticException("Number equal or less than zero")
            else
                state = value
        case result: Result => //对于结果查询,返回结果
            sender ! state
        case ex: NullPointerException =>
            throw new NullPointerException("Null Value Passed")
        case _ => //其他的输入抛出IllegalArgumentException
            throw new IllegalArgumentException("Wrong Argument")
    }
}

//Supervisor Actor的定义, 负责创建Work Actor和supervisorStrategy
class SupervisorActor extends Actor with ActorLogging {
    val childActor = context.actorOf(Props[WorkerActor],name = "workerActor") //创建子Actor
    
    //创建supervisorStrategy是Supervisor的关键
    //前两个参数, maxNrOfRetries:重试的次数
      //the number of times an actor is allowed to be restarted before it is assumed to be dead. 
      //A negative number implies no limits. 
    //withinTimeRange:重试的间隔
    //关键是第三个参数, 定义了异常处理的策略
      //下面针对worker抛出的不同的异常, 匹配了不同的处理方法
    override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 10 seconds) {
        case _: ArithmeticException => Resume 
        case _: NullPointerException => Restart
        case _: IllegalArgumentException => Stop
        case _: Exception => Escalate
    }
    
    //可以接收两种msg
    def receive = {
        case result: Result =>
            childActor.tell(result, sender)
        case msg: Object =>
            childActor ! msg
    }
}

val system = ActorSystem("faultTolerance")
val log = system.log
val originalValue: Int = 0
val supervisor = system.actorOf(Props[SupervisorActor], name = "supervisor")
log.info("Sending value 8, no exceptions should be thrown! ")
var mesg: Int = 8
supervisor ! mesg
implicit val timeout = Timeout(5 seconds)
var future = (supervisor ? new Result).mapTo[Int]
var result = Await.result(future, timeout.duration)
log.info("Value Received-> {}", result)

mesg = -8
supervisor ! mesg

supervisor ! new NullPointerException

supervisor ? "Do Something"

case class Result
case class DeadWorker
case class RegisterWorker(val worker: ActorRef, val supervisor: ActorRef)

class WorkerActor extends Actor with ActorLogging {
    import org.akka.essentials.supervisor.example1.Result
    var state: Int = 0
    override def preStart() {
        log.info("Starting WorkerActor instance hashcode # {}", this.hashCode())
    }
    override def postStop() {
        log.info("Stopping WorkerActor instance hashcode # {}", this.hashCode());
    }
    def receive: Receive = {
        case value: Int =>
            state = value
        case result: Result =>
            sender ! state
        case _ =>
            context.stop(self) //worker会terminate
    }
}

//定义MonitorActor
class MonitorActor extends Actor with ActorLogging {
    var monitoredActors = new HashMap[ActorRef, ActorRef]
    def receive: Receive = {
        case t: Terminated => //收到actor发来的terminated msg
            if (monitoredActors.contains(t.actor)) {
                log.info("Received Worker Actor Termination Message -> " + t.actor.path)
                log.info("Sending message to Supervisor")
                val value: Option[ActorRef] = monitoredActors.get(t.actor)
                value.get ! new DeadWorker() //通知注册的supervisor
            }
        case msg: RegisterWorker => //注册
            context.watch(msg.worker) //关键一步,调用context.watch来监控这个worker actor, 这样worker会在terminated时, 发送msg给monitor
            monitoredActors += msg.worker -> msg.supervisor //将actor和想监控他的supervisor放入map
    }
}

//首先定义成Ref,支持STM的首要条件, STM就是通过Ref的原子切换来实现transaction的
val (x, y) = (Ref(10), Ref(0)) 

def swap = atomic { implicit txn => //简写, 看transfer更清晰
    x = x + y
    y = x - y
    x = x - y
}

def transfer(n: Int) {
    atomic { implicit txn => //加上atomic
        x -= n
        y += n
    }
}

//Account Actor
class AccountActor(accountNumber: String, inBalance: Float) extends Actor {
    val balance = Ref(inBalance) //定义Ref
    def receive = {
        case value: AccountBalance =>
            sender ! new AccountBalance(accountNumber, balance.single.get)
            
        case coordinated @ Coordinated(message: AccountDebit) => //@变量绑定, 从TransferActor得到的coordinated对象
             //coordinated.atomic

           //通过atomic将下面的操作加到transaction

           //从TransferActor得到的coordinated对象, 一个coordinated对象代表一个transaction, 所以必须将所有transaction中的操作加到同一个coordinated对象中
            coordinated atomic { implicit t => 
                //check for funds availability
                if (balance.get(t) > message.amount)
                    balance.transform(_ - message.amount)
                else
                    throw new IllegalStateException("Insufficient Balance")
            }
        case coordinated @ Coordinated(message: AccountCredit) =>
            // coordinated atomic ...
            coordinated atomic { implicit t =>
                balance.transform(_ + message.amount)
            }
    }
}

//TransferActor
class TransferActor extends Actor {
    //创建子account actor
    val fromAccount = "XYZ";
    val toAccount = "ABC";
    val from = context.actorOf(Props(new AccountActor(fromAccount, 5000)), name = fromAccount)
    val to = context.actorOf(Props(new AccountActor(toAccount, 1000)), name = toAccount)
    
    //定义supervisorStrategy进行fault tolerant
    implicit val timeout = Timeout(5 seconds)
    override val supervisorStrategy =
        AllForOneStrategy(maxNrOfRetries = 10,
            withinTimeRange = 10 seconds) {
        case _: CoordinatedTransactionException => Resume
        case _: IllegalStateException => Resume
        case _: IllegalArgumentException => Stop
        case _: Exception => Escalate
    }
    
    def receive: Receive = {
        case message: TransferMsg =>
            val coordinated = Coordinated() //创建Coordinated, 开始一个transaction
            coordinated atomic { implicit t => //使用atomic将下面两个msg加入transaction
                to ! coordinated(new AccountCredit(message.amtToBeTransferred)) //将coordinated对象发送给子actor
                from ! coordinated(new AccountDebit(message.amtToBeTransferred))
            }
        //正常的读取操作, 不需要transaction
        case message: AccountBalance =>
            if (message.accountNumber.equalsIgnoreCase(fromAccount)) {
                from.tell(message, sender)
            } else if (message.accountNumber.equalsIgnoreCase(toAccount)) {
                to.tell(message, sender)
            }
    }
}

//BankActor
class BankActor extends Actor with ActorLogging {
    //创建子transferActor
    val transferActor = context.actorOf(Props[TransferActor], name = "TransferActor")
    
    implicit val timeout = Timeout(5 seconds)
    def receive = {
        case transfer: TransferMsg =>
            transferActor ! transfer
        case balance: AccountBalance =>
            val future = ask(transferActor, balance).mapTo[AccountBalance]
            val account = Await.result(future, timeout.duration)
            log.info("Account #{} , Balance {}", account.accountNumber, account.accountBalance)
    }
    
    override val supervisorStrategy = AllForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 10 seconds) {
        case _: CoordinatedTransactionException => Resume
        case _: IllegalStateException => Stop
        case _: IllegalArgumentException => Stop
        case _: Exception => Escalate
    }
}

class AccountActor(accountNumber: String, inBalance: Float) extends Transactor {
    val balance = Ref(inBalance)
    //atomatically, 原子操作, 可以同时接收普通msg和coordinated msg
    //对于coordinated msg对象, 会把自己加入到该transaction中去
    //对于普通msg对象, 直接起单独的transaction运行
    def atomically = implicit txn => {
        case message: AccountDebit =>
            if (balance.single.get < message.amount)
                throw new IllegalStateException("Insufficient Balance")
            else
                balance transform (_ - message.amount)
        case message: AccountCredit =>
            balance transform (_ + message.amount)
    }
    
    //普通操作, 无论收到什么msg, 都完全bypass coordinated transactions
    override def normally: Receive = {
        case value: AccountBalance =>
            sender ! new AccountBalance(accountNumber, balance.single.get)
    }
}