add lifecycle Monitoring

Author: wangkexue

reactive-programming.org

@@ -1754,3 +1754,108 @@ class Listener(source: ActorRef) extends Actor {
 Actor-local state cannot be kept across restarts, only external state can be managed like this.
 
 Child actors not stopped during restart will be restarted recursively.
+** Lifecycle Monitoring and the Error Kernel
+*** Lifecycle Monitoring
+The only observable transition occurs when stopping an actor:
+ - having an =ActorRef= implies liveness (at some earlier point)
+ - restarts are not externally visible
+ - after stop there will be no more responses
+No replies could also be due to communication failure, therefore Akka supports Lifecycle Monitoring a.k.a. DeathWatch.
+ - an =Actor= registers its interest using =context.watch(target)=
+ - it will receive a =Terminated(target)= message when target stops
+ - it will not receive any direct messages from target thereafter
+*** The DeathWatch API
+#+begin_src scala
+trait ActorContext {
+ def watch(target: ActorRef): ActorRef
+ def unwatch(target: ActorRef): ActorRef
+ ...
+}
+
+case class Terminated private[akka] (actor: ActorRef)
+ (val existenceConfired: Boolean, val addressTerminated: Boolean)
+ extends AutoReceiveMessage with PossiblyHarmful
+#+end_src
+*** The Children List
+Each =actor= maintains a list of the =actors= it created:
+ - the =child= has been entered when =context.actorOf= returns
+ - the =child= has been removed when =Terminated= is received
+ - an =actor= name is available IFF there is no such =child=
+#+begin_src scala
+trait ActorContext {
+ def children: Iterable[ActorRef]
+ def child(name: String): Option[ActorRef]
+ ...
+}
+#+end_src
+*** Applying DeathWatch to Contoller & Getter
+#+begin_src scala
+class Controller extends Actor with ActorLogging {
+ override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 5) {
+ case _: Exception => SupervisorStrategy.Restart
+ }
+ def receive = {
+ case Check(url, depth) =>
+ if (!cache(url) && depth > 0)
+ context.watch(context.actorOf(getterProps(url, depth - 1)))
+ cache += url
+ case Terminated(_) =>
+ if (context.children.isEmpty) context.parent ! Result(cache)
+ case ReceiveTimeout => context.children foreach context.stop
+ }
+ ...
+}
+#+end_src
+*** The Error Kernel
+Keep important data near the root, delegate risk to the leaves.
+ - restarts are recursive (supervised actors are part of the state)
+ - restarts are more frequent near the leaves
+ - avoid restarting Actors with important state
+*** Application to Receptionist
+ - Always stop Contoller if it has a problem.
+ - React to =Terminated= to catch cases where =no Result= was sent.
+ - Discard =Terminated= after =Result= was sent.
+#+begin_src scala
+class Receptionist extends Actor {
+ override def supervisorStrategy = SupervisorStrategy.stoppingStrategy
+ ...
+}
+#+end_src
+*** Interjection: the EventStream
+Actors can direct messages only at known address.
+
+The EventStream allows publication of messages to an unknown audience.
+
+Every actor can optionally subscribe to (parts of) the EventStream.
+#+begin_src scala
+trait EventStream {
+ def subscribe(subscriber: ActorRef, topic: Class[_]): Boolean
+ def unsubscribe(subscriber: ActorRef, topic: Class[_]): Boolean
+ def unsubscribe(subscriber: ActorRef): Unit
+ def publish(event: AnyRef): Unit
+}
+
+class Listener extends Actor {
+ context.system.eventStream.subscribe(self, classOf[LogEvent])
+ def receive = {
+ case e: LogEvent => ...
+ }
+ override def postStop(): Unit = {
+ context.system.eventStream.unsubscribe(self)
+ }
+}
+#+end_src
+*** Where do Unhandled Messages Go?
+=Actor.Receive= is a partial function, the behavior may not apply.
+
+=Unhandled= messages are passed into the =unhandled= method:
+#+begin_src scala
+trait Actor {
+ ...
+ def unhandled(message: Any): Unit = message match {
+ case Terminated(target) => throw new DeathPactException(target)
+ case msg =>
+ context.system.eventStream.publish(UnhandledMessage(msg, sender, self))
+ }
+}
+#+end_src