package user
import gr.grnet.aquarium.actor._
-import gr.grnet.aquarium.user._
-
-import gr.grnet.aquarium.util.shortClassNameOf
-import gr.grnet.aquarium.util.chainOfCauses
-import gr.grnet.aquarium.util.date.TimeHelpers
-import gr.grnet.aquarium.actor.message.service.router._
-import message.config.{ActorProviderConfigured, AquariumPropertiesLoaded}
-import gr.grnet.aquarium.event.im.IMEventModel
-import akka.config.Supervision.Temporary
-import akka.actor.PoisonPill
-import gr.grnet.aquarium.{AquariumException, Configurator}
+import akka.config.Supervision.Temporary
+import gr.grnet.aquarium.Aquarium
+import gr.grnet.aquarium.actor.message.event.{ProcessResourceEvent, ProcessIMEvent}
+import gr.grnet.aquarium.actor.message.{GetUserStateRequest, GetUserBalanceRequest}
+import gr.grnet.aquarium.computation.data.IMStateSnapshot
+import gr.grnet.aquarium.event.model.im.IMEventModel
+import gr.grnet.aquarium.actor.message.config.{InitializeUserState, ActorProviderConfigured, AquariumPropertiesLoaded}
/**
*
*/
class UserActor extends ReflectiveRoleableActor {
- private[this] var _userID: String = _
- private[this] var _userState: UserState = _
+ private[this] var _imState: IMStateSnapshot = _
+// private[this] var _userState: UserState = _
+// private[this] var _newUserState: NewUserState = _
self.lifeCycle = Temporary
+// private[this] def _userID = this._newUserState.userID
private[this] def _shutmedown(): Unit = {
- if(_haveFullState) {
- UserActorCache.invalidate(this._userID)
- }
+// if(_haveUserState) {
+// UserActorCache.invalidate(_userID)
+// }
- self ! PoisonPill
+ self.stop()
}
override protected def onThrowable(t: Throwable, message: AnyRef) = {
- ERROR("Oops!\n", chainOfCauses(t).map("!! " + _) mkString "\n")
- ERROR(t, "Terminating due to: %s(%s)", shortClassNameOf(t), t.getMessage)
+ logChainOfCauses(t)
+// ERROR(t, "Terminating due to: %s(%s)", shortClassNameOf(t), t.getMessage)
_shutmedown()
}
def role = UserActorRole
- private[this] def _configurator: Configurator = Configurator.MasterConfigurator
-// private[this] def _userId = _userState.userId
+ private[this] def aquarium: Aquarium = Aquarium.Instance
private[this] def _timestampTheshold =
- _configurator.props.getLong(Configurator.Keys.user_state_timestamp_threshold).getOr(10000)
+ aquarium.props.getLong(Aquarium.Keys.user_state_timestamp_threshold).getOr(10000)
- private[this] def _haveFullState = {
- (this._userID ne null) && (this._userState ne null)
+// private[this] def _haveUserState = {
+// this._newUserState ne null
+// }
+
+ private[this] def _haveIMState = {
+ this._imState ne null
}
def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
def onActorProviderConfigured(event: ActorProviderConfigured): Unit = {
}
- private[this] def _computeAgreementForNewUser(imEvent: IMEventModel): String = {
- // FIXME: Implement based on the role
- "default"
- }
+ private[this] def reloadIMState(userID: String): Unit = {
+ val store = aquarium.imEventStore
+ store.replayIMEventsInOccurrenceOrder(userID) { imEvent ⇒
+ logger.debug("Replaying %s".format(imEvent))
- private[this] def processCreateUser(imEvent: IMEventModel): Unit = {
- val userID = imEvent.userID
- this._userID = userID
+ val newState = this._imState match {
+ case null ⇒
+ IMStateSnapshot.initial(imEvent)
- val store = _configurator.storeProvider.userStateStore
- // try find user state. normally should ot exist
- val latestUserStateOpt = store.findLatestUserStateByUserID(userID)
- if(latestUserStateOpt.isDefined) {
- logger.error("Got %s(%s, %s) but user already exists. Ingoring".format(
- userID,
- shortClassNameOf(imEvent),
- imEvent.eventType))
+ case currentState ⇒
+ currentState.copyWithEvent(imEvent)
+ }
- return
+ this._imState = newState
}
- val initialAgreementName = _computeAgreementForNewUser(imEvent)
- val newUserState = DefaultUserStateComputations.createInitialUserState(
- userID,
- imEvent.occurredMillis,
- imEvent.isActive,
- 0.0,
- List(imEvent.role),
- initialAgreementName)
-
- this._userState = newUserState
-
- // FIXME: If this fails, then the actor must be shut down.
- store.insertUserState(newUserState)
+ logger.debug("Recomputed %s".format(this._imState))
}
- private[this] def processModifyUser(imEvent: IMEventModel): Unit = {
- val now = TimeHelpers.nowMillis()
-
- if(!_haveFullState) {
- ERROR("Got %s(%s) but have no state. Shutting down", shortClassNameOf(imEvent), imEvent.eventType)
- _shutmedown()
- return
- }
-
- this._userState = this._userState.modifyFromIMEvent(imEvent, now)
+ def onInitializeUserState(event: InitializeUserState): Unit = {
+ logger.debug("Got %s".format(event))
+ reloadIMState(event.userID)
}
- def onProcessSetUserID(event: ProcessSetUserID): Unit = {
- this._userID = event.userID
+ private[this] def _getAgreementNameForNewUser(imEvent: IMEventModel): String = {
+ // FIXME: Implement based on the role
+ "default"
}
- def onProcessIMEvent(event: ProcessIMEvent): Unit = {
- val imEvent = event.imEvent
- if(imEvent.isCreateUser) {
- processCreateUser(imEvent)
- } else if(imEvent.isModifyUser) {
- processModifyUser(imEvent)
+ /**
+ * Process [[gr.grnet.aquarium.event.model.im.IMEventModel]]s.
+ * When this method is called, we assume that all proper checks have been made and it
+ * is OK to proceed with the event processing.
+ */
+ def onProcessIMEvent(processEvent: ProcessIMEvent): Unit = {
+ val imEvent = processEvent.imEvent
+ val hadIMState = _haveIMState
+
+ if(hadIMState) {
+ if(this._imState.latestIMEvent.id == imEvent.id) {
+ // This happens when the actor is brought to life, then immediately initialized, and then
+ // sent the first IM event. But from the initialization procedure, this IM event will have
+ // already been loaded from DB!
+ logger.debug("Ignoring first %s after birth".format(imEvent.toDebugString))
+ return
+ }
+
+ this._imState = this._imState.copyWithEvent(imEvent)
} else {
- throw new AquariumException("Cannot interpret %s".format(imEvent))
+ this._imState = IMStateSnapshot.initial(imEvent)
}
+
+// DEBUG("%s %s = %s", if(hadIMState) "Update" else "Set", shortClassNameOf(this._imState), this._imState)
}
- def onRequestUserBalance(event: RequestUserBalance): Unit = {
+ def onGetUserBalanceRequest(event: GetUserBalanceRequest): Unit = {
val userId = event.userID
- // FIXME: Implement threshold
- self reply UserResponseGetBalance(userId, _userState.creditsSnapshot.creditAmount)
+ // FIXME: Implement
+// self reply GetUserBalanceResponse(userId, Right(_userState.creditsSnapshot.creditAmount))
}
- def onUserRequestGetState(event: UserRequestGetState): Unit = {
+ def onGetUserStateRequest(event: GetUserStateRequest): Unit = {
val userId = event.userID
- // FIXME: implement
- self reply UserResponseGetState(userId, this._userState)
+ // FIXME: Implement
+// self reply GetUserStateResponse(userId, Right(this._userState))
}
def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
- }
-
+ val rcEvent = event.rcEvent
- private[this] def D_userID = {
- if(this._userID eq null)
- "<NOT INITIALIZED>" // We always get a userID first
- else
- if(this._userState eq null)
- "%s, NO STATE".format(this._userID)
- else
- "%s".format(this._userID)
+ logger.info("Got\n{}", rcEvent.toJsonString)
}
- private[this] def DEBUG(fmt: String, args: Any*) =
- logger.debug("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
-
- private[this] def INFO(fmt: String, args: Any*) =
- logger.info("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
-
- private[this] def WARN(fmt: String, args: Any*) =
- logger.warn("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
-
- private[this] def ERROR(fmt: String, args: Any*) =
- logger.error("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
- private[this] def ERROR(t: Throwable, fmt: String, args: Any*) =
- logger.error("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)), t)
+// private[this] def D_userID = {
+// if(this._newUserState eq null)
+// if(this._imState eq null)
+// "<NOT INITIALIZED>"
+// else
+// this._imState.latestIMEvent.userID
+// else
+// this._newUserState.userID
+// }
+//
+// private[this] def DEBUG(fmt: String, args: Any*) =
+// logger.debug("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
+//
+// private[this] def INFO(fmt: String, args: Any*) =
+// logger.info("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
+//
+// private[this] def WARN(fmt: String, args: Any*) =
+// logger.warn("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
+//
+// private[this] def ERROR(fmt: String, args: Any*) =
+// logger.error("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
+//
+// private[this] def ERROR(t: Throwable, fmt: String, args: Any*) =
+// logger.error("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)), t)
}
projects / aquarium / blobdiff