package user
import gr.grnet.aquarium.actor._
-import gr.grnet.aquarium.user._
-import gr.grnet.aquarium.util.shortClassNameOf
-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 gr.grnet.aquarium.{AquariumInternalError, AquariumException, Configurator}
-
+import gr.grnet.aquarium.Aquarium
+import gr.grnet.aquarium.util.{shortClassNameOf, shortNameOfClass}
+import gr.grnet.aquarium.actor.message.event.{ProcessResourceEvent, ProcessIMEvent}
+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}
+import gr.grnet.aquarium.actor.message.{GetUserBalanceResponseData, GetUserBalanceResponse, GetUserStateRequest, GetUserBalanceRequest}
+import gr.grnet.aquarium.computation.{BillingMonthInfo, UserStateBootstrappingData, UserState}
+import gr.grnet.aquarium.util.date.TimeHelpers
+import gr.grnet.aquarium.logic.accounting.Policy
+import gr.grnet.aquarium.computation.reason.InitialUserStateSetup
/**
*
*/
class UserActor extends ReflectiveRoleableActor {
- private[this] var _userID: String = _
+ private[this] var _userID: String = "<?>"
+ private[this] var _imState: IMStateSnapshot = _
private[this] var _userState: UserState = _
self.lifeCycle = Temporary
private[this] def _shutmedown(): Unit = {
- if(_haveFullState) {
- UserActorCache.invalidate(this._userID)
+ if(_haveUserState) {
+ UserActorCache.invalidate(_userID)
}
self.stop()
def role = UserActorRole
- private[this] def _configurator: Configurator = Configurator.MasterConfigurator
-// private[this] def _userId = _userState.userId
-
- private[this] def _timestampTheshold =
- _configurator.props.getLong(Configurator.Keys.user_state_timestamp_threshold).getOr(10000)
-
+ private[this] def aquarium: Aquarium = Aquarium.Instance
+ private[this] def userStateComputations = aquarium.userStateComputations
- private[this] def _haveFullState = {
- (this._userID ne null) && (this._userState ne null)
+ private[this] def _timestampTheshold = {
+ aquarium.props.getLong(Aquarium.Keys.user_state_timestamp_threshold).getOr(10000)
}
- private[this] def _havePartialState = {
- (this._userID ne null) && (this._userState eq null)
+ private[this] def _haveUserState = {
+ this._userState ne null
}
+ private[this] def _haveIMState = {
+ this._imState ne null
+ }
def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
}
def onActorProviderConfigured(event: ActorProviderConfigured): Unit = {
}
- private[this] def _getAgreementNameForNewUser(imEvent: IMEventModel): String = {
- // FIXME: Implement based on the role
- "default"
- }
+ private[this] def createIMState(event: InitializeUserState): Unit = {
+ val userID = event.userID
+ val store = aquarium.imEventStore
+ // TODO: Optimization: Since IMState only records roles, we should incrementally
+ // TODO: built it only for those IMEvents that changed the role.
+ store.replayIMEventsInOccurrenceOrder(userID) { imEvent ⇒
+ logger.debug("Replaying %s".format(imEvent))
- private[this] def processCreateUser(imEvent: IMEventModel): Unit = {
- this._userID = imEvent.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(this._userID)
- if(latestUserStateOpt.isDefined) {
- logger.error("Got %s(%s, %s) but user already exists. Ingoring".format(
- this._userID,
- shortClassNameOf(imEvent),
- imEvent.eventType))
+ case currentState ⇒
+ currentState.updateHistoryWithEvent(imEvent)
+ }
- return
+ this._imState = newState
}
- val initialAgreementName = _getAgreementNameForNewUser(imEvent)
- val newUserState = DefaultUserStateComputations.createInitialUserState(
- this._userID,
- imEvent.occurredMillis,
- imEvent.isActive,
- 0.0,
- List(imEvent.role),
- initialAgreementName)
-
- this._userState = newUserState
+ DEBUG("Recomputed %s = %s", shortNameOfClass(classOf[IMStateSnapshot]), this._imState)
+ }
- // FIXME: If this fails, then the actor must be shut down.
- store.insertUserState(newUserState)
+ /**
+ * Resource events are processed only if the user has been activated.
+ */
+ private[this] def shouldProcessResourceEvents: Boolean = {
+ _haveIMState && this._imState.hasBeenActivated
}
- private[this] def processModifyUser(imEvent: IMEventModel): Unit = {
- val now = TimeHelpers.nowMillis()
+ private[this] def createUserState(event: InitializeUserState): Unit = {
+ val userID = event.userID
+ val referenceTime = event.referenceTimeMillis
- if(!_haveFullState) {
- ERROR("Got %s(%s) but have no state. Shutting down", shortClassNameOf(imEvent), imEvent.eventType)
- _shutmedown()
+ if(!_haveIMState) {
+ // Should have been created from `createIMState()`
+ DEBUG("Cannot create user state from %s, since %s = %s", event, shortNameOfClass(classOf[IMStateSnapshot]), this._imState)
return
}
- this._userState = this._userState.modifyFromIMEvent(imEvent, now)
- }
+ if(!this._imState.hasBeenActivated) {
+ // Cannot set the initial state!
+ DEBUG("Cannot create user state from %s, since user is inactive", event)
+ return
+ }
- def onProcessSetUserID(event: ProcessSetUserID): Unit = {
- this._userID = event.userID
+ val userActivationMillis = this._imState.userActivationMillis.get
+ val initialRole = this._imState.roleHistory.firstRole.get.name
+
+ val userStateBootstrap = UserStateBootstrappingData(
+ this._userID,
+ userActivationMillis,
+ initialRole,
+ aquarium.initialAgreementForRole(initialRole, userActivationMillis),
+ aquarium.initialBalanceForRole(initialRole, userActivationMillis)
+ )
+
+ userStateComputations.doFullMonthlyBilling(
+ userStateBootstrap,
+ BillingMonthInfo.fromMillis(TimeHelpers.nowMillis()),
+ aquarium.currentResourcesMap,
+ InitialUserStateSetup,
+ None
+ )
}
- def onProcessIMEvent(event: ProcessIMEvent): Unit = {
- val now = TimeHelpers.nowMillis()
+ def onInitializeUserState(event: InitializeUserState): Unit = {
+ val userID = event.userID
+ this._userID = userID
+ DEBUG("Got %s", event)
- val imEvent = event.imEvent
- // If we already have a userID but it does not match the incoming userID, then this is an internal error
- if(_havePartialState && (this._userID != imEvent.userID)) {
- throw new AquariumInternalError(
- "Got userID = %s but already have userID = %s".format(imEvent.userID, this._userID))
- }
+ createIMState(event)
+ createUserState(event)
+ }
- // If we get an IMEvent without having a user state, then we query for the latest user state.
- if(!_haveFullState) {
- val userStateOpt = _configurator.userStateStore.findLatestUserStateByUserID(this._userID)
- this._userState = userStateOpt match {
- case Some(userState) ⇒
- userState
-
- case None ⇒
- val initialAgreementName = _getAgreementNameForNewUser(imEvent)
- val initialUserState = DefaultUserStateComputations.createInitialUserState(
- this._userID,
- imEvent.occurredMillis,
- imEvent.isActive,
- 0.0,
- List(imEvent.role),
- initialAgreementName)
-
- DEBUG("Got initial state")
- initialUserState
- }
+ /**
+ * 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
+
+ if(!_haveIMState) {
+ // This is an error. Should have been initialized from somewhere ...
+ throw new Exception("Got %s while being uninitialized".format(processEvent))
}
- if(imEvent.isModifyUser && this._userState.isInitial) {
- INFO("Got a '%s' but have not received '%s' yet", imEvent.eventType, IMEventModel.EventTypeNames.create)
+ 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!
+ INFO("Ignoring first %s after birth", imEvent.toDebugString)
return
}
- if(imEvent.isCreateUser && !this._userState.isInitial) {
- INFO("Got a '%s' but my state is not initial", imEvent.eventType)
+ this._imState = this._imState.updateHistoryWithEvent(imEvent)
+
+ INFO("Update %s = %s", shortClassNameOf(this._imState), this._imState)
+ }
+
+ def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
+ val rcEvent = event.rcEvent
+
+ if(!shouldProcessResourceEvents) {
+ // This means the user has not been activated. So, we do not process any resource event
+ DEBUG("Not processing %s", rcEvent.toJsonString)
return
}
+ }
- this._userState = this._userState.modifyFromIMEvent(imEvent, now)
- if(imEvent.isCreateUser) {
- processCreateUser(imEvent)
- } else if(imEvent.isModifyUser) {
- processModifyUser(imEvent)
- } else {
- throw new AquariumException("Cannot interpret %s".format(imEvent))
+ def onGetUserBalanceRequest(event: GetUserBalanceRequest): Unit = {
+ val userID = event.userID
+
+ if(!_haveIMState) {
+ // No IMEvent has arrived, so this user is virtually unknown
+ self reply GetUserBalanceResponse(Left("User not found"), 404/*Not found*/)
+ }
+ else if(!_haveUserState) {
+ // The user is known but we have no state.
+ // Ridiculous. Should have been created at least during initialization.
}
- }
- def onRequestUserBalance(event: RequestUserBalance): Unit = {
- val userId = event.userID
- // FIXME: Implement threshold
- self reply UserResponseGetBalance(userId, _userState.creditsSnapshot.creditAmount)
+ if(!_haveUserState) {
+ self reply GetUserBalanceResponse(Left("Not found"), 404/*Not found*/)
+ } else {
+ self reply GetUserBalanceResponse(Right(GetUserBalanceResponseData(userID, this._userState.totalCredits)))
+ }
}
- 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 = {
- }
-
-
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)
+ this._userID
}
private[this] def DEBUG(fmt: String, args: Any*) =
- logger.debug("UserActor[%s]: %s".format(D_userID, fmt.format(args: _*)))
+ logger.debug("User[%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: _*)))
+ logger.info("User[%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: _*)))
+ logger.warn("User[%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: _*)))
+ logger.error("User[%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)
+ logger.error("User[%s]: %s".format(D_userID, fmt.format(args: _*)), t)
}
projects / aquarium / blobdiff