code.grnet.gr Git - aquarium/blob - src/main/scala/gr/grnet/aquarium/actor/service/router/RouterActor.scala

   1 /*
   2  * Copyright 2011-2012 GRNET S.A. All rights reserved.
   3  *
   4  * Redistribution and use in source and binary forms, with or
   5  * without modification, are permitted provided that the following
   6  * conditions are met:
   7  *
   8  *   1. Redistributions of source code must retain the above
   9  *      copyright notice, this list of conditions and the following
  10  *      disclaimer.
  11  *
  12  *   2. Redistributions in binary form must reproduce the above
  13  *      copyright notice, this list of conditions and the following
  14  *      disclaimer in the documentation and/or other materials
  15  *      provided with the distribution.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  18  * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  19  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  20  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  21  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  24  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  25  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  27  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  28  * POSSIBILITY OF SUCH DAMAGE.
  29  *
  30  * The views and conclusions contained in the software and
  31  * documentation are those of the authors and should not be
  32  * interpreted as representing official policies, either expressed
  33  * or implied, of GRNET S.A.
  34  */
  35
  36 package gr.grnet.aquarium.actor
  37 package service
  38 package router
  39
  40 import gr.grnet.aquarium.util.shortClassNameOf
  41 import gr.grnet.aquarium.service.RoleableActorProviderService
  42 import message.service.router._
  43 import akka.actor.ActorRef
  44 import user.{UserActorCache, UserActorSupervisor}
  45 import message.config.{AquariumPropertiesLoaded, ActorProviderConfigured}
  46
  47 /**
  48  * Business logic router. Incoming messages are routed to appropriate destinations. Replies are routed back
  49  * appropriately.
  50  *
  51  * @author Christos KK Loverdos <loverdos@gmail.com>.
  52  */
  53 class RouterActor extends ReflectiveRoleableActor {
  54   private[this] var _actorProvider: RoleableActorProviderService = _
  55
  56   def role = RouterRole
  57
  58   private[this] def _launchUserActor(userID: String): ActorRef = {
  59     // create a fresh instance
  60     val userActor = _actorProvider.actorForRole(UserActorRole)
  61     UserActorCache.put(userID, userActor)
  62     UserActorSupervisor.supervisor.link(userActor)
  63
  64     userActor
  65   }
  66
  67   private[this] def _findOrCreateUserActor(userID: String): ActorRef = {
  68     UserActorCache.get(userID) match {
  69       case Some(userActorRef) ⇒
  70         userActorRef
  71       case None ⇒
  72         _launchUserActor(userID)
  73     }
  74   }
  75
  76   private[this] def _forwardToUserActor(userID: String, m: RouterMessage): Unit = {
  77     try {
  78       _findOrCreateUserActor(userID) forward m
  79
  80     } catch { case t: Throwable ⇒
  81       logger.error("While forwarding to user actor for userID = %s".format(userID), t)
  82       // FIXME: We have a message that never gets to the user actor.
  83       // FIXME: We should probably shut the user actor down.
  84     }
  85   }
  86
  87   def onAquariumPropertiesLoaded(m: AquariumPropertiesLoaded): Unit = {
  88     logger.info("Configured with {}", shortClassNameOf(m))
  89   }
  90
  91   def onActorProviderConfigured(m: ActorProviderConfigured): Unit = {
  92     this._actorProvider = m.actorProvider
  93     logger.info("Configured with {}", shortClassNameOf(m))
  94   }
  95
  96   def onProcessIMEvent(m: ProcessIMEvent): Unit = {
  97      _forwardToUserActor(m.imEvent.userID, m)
  98   }
  99
 100   def onRequestUserBalance(m: RequestUserBalance): Unit = {
 101     _forwardToUserActor(m.userID, m)
 102   }
 103
 104   def onUserRequestGetState(m: UserRequestGetState): Unit = {
 105     _forwardToUserActor(m.userID, m)
 106   }
 107
 108   def onProcessResourceEvent(m: ProcessResourceEvent): Unit = {
 109     _forwardToUserActor(m.rcEvent.userID, m)
 110   }
 111
 112   def onAdminRequestPingAll(m: AdminRequestPingAll): Unit = {
 113
 114   }
 115
 116   override def postStop = {
 117     UserActorCache.stop
 118   }
 119 }