Akka: comment puis-je attraper l'échec d'un acteur à l'intérieur d'un autre acteur (autre qu'un enfant)?

Question

J'ai deux acteurs:

ProcessManager qui gère certains processus dans le système (par exemple, l'enregistrement des utilisateurs, etc)

notificateur - doit en informer l'utilisateur si une erreur est survenue dans ProcessManager . J'ai besoin d'intercepter l'échec de l'acteur ProcessManager (il a échoué et s'est arrêté pour n'importe quelle raison, par exemple, à cause d'ActorInitializationException ou du temps de redémarrage maximum atteint et l'acteur Process Manager a été arrêté).

class ProcessManager extends Actor { 
     override def receive: Receive = { 
     ... 
     } 
    } 

    class Notifier extends Actor { 
     override def receive: Receive = { 
     PROCESS MANAGER ACTOR FAILED AND STOPPED => 
      // Here I need to catch failure of ProcessManager actor 
      // (it was failed and stopped for what ever 
      // reason, for example, because of ActorInitializationException 
      // or max restart time reached and Process manager actor was stopped). 
      // 
      // Then do some stuff, for example, send message to the client via web socket. 
     } 
    } 


    class MyController @Inject() (cc: ControllerComponents, actorSystem: ActorSystem) 
     (implicit exec: ExecutionContext) extends AbstractController(cc) { 


     // I need to catch failure of processManager in this actor. 
     val notifier = actorSystem.actorOf(Props(classOf[Notifier])) 

     def registerUser = Action.async {   

      // Actor may be stopped because of ActorInitializationException here 
      val processManager = actorSystem.actorOf(Props(classOf[ProcessManager])) 
       ... 

     // OR it may be stopped here for any reason. 
     processManager ! "some message which will fail and stop pm actor" 

     Future.successfull(Ok("Thanks.")) 
     } 
    } 

Comment puis-je prendre fin (en raison de l'échec) de ProcessManager acteur à l'intérieur notificateur acteur? Permettez-moi d'expliquer le contexte de mon problème.

EDIT

Je crée PM acteur dans le contrôleur de jeu et lui envoie le message (Tell) et je renvoie la réponse Ok immédiatement à l'utilisateur. L'acteur PM crée un autre acteur enfant et lors de la création, ActorInitializationException est levé. J'ai besoin d'aviser l'utilisateur (via la prise Web, en utilisant l'acteur Notifier) ​​que quelque chose s'est mal passé.

Answer 1

Vous pouvez utiliser DeathWatch pour enregistrer l'acteur Notifier pour la réception du message Terminated lorsque l'acteur ProcessManager arrête définitivement. Notifier aura besoin d'une référence à l'ProcessManager acteur pour le DeathWatch, et une façon de le faire est d'envoyer la référence à ProcessManager comme un message (c'est sûr car ActorRef est immuable et sérialisable).

class Notifier extends Actor { 
    var processManager: Option[ActorRef] = None 

    def receive: Receive = { 
    case aRef: ActorRef => 
     if (processManager.isEmpty) { 
     processManager = Some(aRef) 
     context.watch(aRef) // register to "watch" the process manager 
     } 
    case Terminated => 
     // process manager was permanently stopped 
    case ... 
    } 
} 

object Demo extends App { 
    val actorSystem = ActorSystem("my-actor-system") 

    val notifier = actorSystem.actorOf(Props(classOf[Notifier])) 
    val processManager = actorSystem.actorOf(Props(classOf[ProcessManager])) 

    notifier ! processManager // send processManager's ref to the notifier 
    ... 
    processManager ! "some message which will fail and stop pm actor" 
    ... 
} 

Une mise en garde: il pourrait ne pas être possible pour l'enregistrement DeathWatch se produise devant un ActorInitializationException est jeté lors d'une tentative de créer le ProcessManager.

---

Si vous avez besoin d'envoyer un message à Notifier quand un enfant de ProcessManager lance une exception, outrepasser la stratégie de superviseur ProcessManager et envoyer ce message dans le cadre de la stratégie. Quelque chose comme:

class ProcessManager extends Actor { 
    import akka.actor.OneForOneStrategy 
    import akka.actor.SupervisorStrategy._ 
    import scala.concurrent.duration._ 

    override val supervisorStrategy = 
    OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) { 
     case _: ActorInitializationException => 
     val notifier = context.actorSelection("/path/to/notifier") 
     notifier ! CustomErrorMessage 
     Stop 
     case _: Exception => Escalate 
    } 

    def receive: Receive = { 
    ... 
    } 
}