package service
package user
-import gr.grnet.aquarium.actor._
-
-import akka.config.Supervision.Temporary
-import gr.grnet.aquarium.actor.message.event.{ProcessResourceEvent, ProcessIMEvent}
-import gr.grnet.aquarium.computation.data.IMStateSnapshot
-import gr.grnet.aquarium.actor.message.config.{InitializeUserState, ActorProviderConfigured, AquariumPropertiesLoaded}
-import gr.grnet.aquarium.computation.{BillingMonthInfo, UserStateBootstrappingData, UserState}
+import gr.grnet.aquarium.AquariumInternalError
+import gr.grnet.aquarium.actor.message.GetUserBalanceRequest
+import gr.grnet.aquarium.actor.message.GetUserBalanceResponse
+import gr.grnet.aquarium.actor.message.GetUserBalanceResponseData
+import gr.grnet.aquarium.actor.message.GetUserBillRequest
+import gr.grnet.aquarium.actor.message.GetUserBillResponse
+import gr.grnet.aquarium.actor.message.GetUserBillResponseData
+import gr.grnet.aquarium.actor.message.GetUserStateRequest
+import gr.grnet.aquarium.actor.message.GetUserStateResponse
+import gr.grnet.aquarium.actor.message.GetUserWalletRequest
+import gr.grnet.aquarium.actor.message.GetUserWalletResponse
+import gr.grnet.aquarium.actor.message.GetUserWalletResponseData
+import gr.grnet.aquarium.actor.message.config.AquariumPropertiesLoaded
+import gr.grnet.aquarium.charging.state.{UserStateModel, UserAgreementHistoryModel, UserStateBootstrap}
+import gr.grnet.aquarium.computation.BillingMonthInfo
+import gr.grnet.aquarium.message.avro.gen.{IMEventMsg, ResourceEventMsg, UserStateMsg}
+import gr.grnet.aquarium.message.avro.{ModelFactory, MessageFactory, MessageHelpers, AvroHelpers}
+import gr.grnet.aquarium.service.event.BalanceEvent
import gr.grnet.aquarium.util.date.TimeHelpers
-import gr.grnet.aquarium.event.model.im.IMEventModel
-import gr.grnet.aquarium.{AquariumException, Aquarium}
-import gr.grnet.aquarium.actor.message.{GetUserStateResponse, GetUserBalanceResponseData, GetUserBalanceResponse, GetUserStateRequest, GetUserBalanceRequest}
-import gr.grnet.aquarium.util.{LogHelpers, shortClassNameOf, shortNameOfClass, shortNameOfType}
-import gr.grnet.aquarium.computation.reason.{NoSpecificChangeReason, InitialUserActorSetup, UserStateChangeReason, IMEventArrival, InitialUserStateSetup}
+import gr.grnet.aquarium.util.{LogHelpers, shortClassNameOf}
+import gr.grnet.aquarium.policy.{ResourceType, PolicyModel}
+import gr.grnet.aquarium.charging.bill.BillEntryMsg
/**
*
*/
class UserActor extends ReflectiveRoleableActor {
+ private[this] var _rcMsgCount = 0
+ private[this] var _imMsgCount = 0
private[this] var _userID: String = "<?>"
- private[this] var _imState: IMStateSnapshot = _
- private[this] var _userState: UserState = _
+ private[this] var _userState: UserStateModel = _
+ private[this] var _userCreationIMEvent: IMEventMsg = _
+ private[this] var _userAgreementHistoryModel: UserAgreementHistoryModel = _
+ private[this] var _latestIMEventOriginalID: String = ""
+ private[this] var _latestIMEventOccurredMillis: Long = -1L
+ private[this] var _latestResourceEventOriginalID: String = ""
+ private[this] var _userStateBootstrap: UserStateBootstrap = _
+
+ def unsafeUserID = {
+ if(!haveUserID) {
+ throw new AquariumInternalError("%s not initialized")
+ }
- self.lifeCycle = Temporary
+ this._userID
+ }
- private[this] def _shutmedown(): Unit = {
- if(haveUserState) {
- UserActorCache.invalidate(_userID)
- }
+ override def postStop() {
+ DEBUG("I am finally stopped (in postStop())")
+ aquarium.akkaService.notifyUserActorPostStop(this)
+ }
- self.stop()
+ private[this] def shutmedown(): Unit = {
+ if(haveUserID) {
+ aquarium.akkaService.invalidateUserActor(this)
+ }
}
override protected def onThrowable(t: Throwable, message: AnyRef) = {
LogHelpers.logChainOfCauses(logger, t)
ERROR(t, "Terminating due to: %s(%s)", shortClassNameOf(t), t.getMessage)
- _shutmedown()
+ shutmedown()
}
def role = UserActorRole
- private[this] def aquarium: Aquarium = Aquarium.Instance
- private[this] def userStateComputations = aquarium.userStateComputations
-
- private[this] def _timestampTheshold = {
- aquarium.props.getLong(Aquarium.Keys.user_state_timestamp_threshold).getOr(10000)
- }
+ private[this] def chargingService = aquarium.chargingService
- private[this] def haveUserState = {
- this._userState ne null
+ private[this] def stdUserStateStoreFunc = (userState: UserStateMsg) ⇒ {
+ aquarium.userStateStore.insertUserState(userState)
}
- private[this] def haveIMState = {
- this._imState ne null
+ @inline private[this] def haveUserID = {
+ this._userID ne null
}
def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
}
- def onActorProviderConfigured(event: ActorProviderConfigured): Unit = {
+ @inline private[this] def haveAgreements = {
+ (this._userAgreementHistoryModel ne null) && this._userAgreementHistoryModel.size > 0
}
- private[this] def _updateIMStateRoleHistory(imEvent: IMEventModel, roleCheck: Option[String]) = {
- if(haveIMState) {
- val (newState,
- creationTimeChanged,
- activationTimeChanged,
- roleChanged) = this._imState.updatedWithEvent(imEvent, roleCheck)
-
- this._imState = newState
- (creationTimeChanged, activationTimeChanged, roleChanged)
- } else {
- this._imState = IMStateSnapshot.initial(imEvent)
- (
- imEvent.isCreateUser,
- true, // first activation status is a change by default??
- true // first role is a change by default??
- )
- }
+ @inline private[this] def haveUserState = {
+ this._userState ne null
}
- /**
- * Creates the IMStateSnapshot and returns the number of updates it made to it.
- */
- private[this] def createInitialIMState(): Unit = {
- val store = aquarium.imEventStore
+ @inline private[this] def haveUserStateBootstrap = {
+ this._userStateBootstrap ne null
+ }
- var _roleCheck = None: Option[String]
+ private[this] def createUserAgreementHistoryModel(imEvent: IMEventMsg) {
+ assert(MessageHelpers.isIMEventCreate(imEvent))
+ assert(this._userAgreementHistoryModel eq null)
+ assert(this._userCreationIMEvent eq null)
- // this._userID is already set up
- store.replayIMEventsInOccurrenceOrder(this._userID) { imEvent ⇒
- DEBUG("Replaying %s", imEvent)
+ this._userCreationIMEvent = imEvent
+ this._userAgreementHistoryModel = ModelFactory.newUserAgreementHistoryModelFromIMEvent(
+ imEvent,
+ imEvent.getOriginalID
+ )
+ }
- val (creationTimeChanged, activationTimeChanged, roleChanged) = _updateIMStateRoleHistory(imEvent, _roleCheck)
- _roleCheck = this._imState.roleHistory.lastRoleName
+ private[this] def updateAgreementHistoryFrom(imEvent: IMEventMsg): Unit = {
+ val isCreateUser = MessageHelpers.isIMEventCreate(imEvent)
+ if(isCreateUser) {
+ if(haveAgreements) {
+ throw new AquariumInternalError(
+ "Got user creation event (id=%s) but I already have one (id=%s)",
+ this._userCreationIMEvent.getOriginalID,
+ imEvent.getOriginalID
+ )
+ }
- DEBUG(
- "(creationTimeChanged, activationTimeChanged, roleChanged)=(%s, %s, %s) using %s",
- creationTimeChanged, activationTimeChanged, roleChanged,
- imEvent
- )
+ createUserAgreementHistoryModel(imEvent) // now we have an agreement history
+ createUserStateBootstrap(imEvent)
}
- DEBUG("New %s = %s", shortNameOfType[IMStateSnapshot], this._imState)
+ val effectiveFromMillis = imEvent.getOccurredMillis
+ val role = imEvent.getRole
+ // calling unsafe just for the side-effect
+ assert(null ne aquarium.unsafeFullPriceTableForRoleAt(role, effectiveFromMillis))
+
+ // add to model (will update the underlying messages as well)
+ val newUserAgreementModel = ModelFactory.newUserAgreementModelFromIMEvent(imEvent, imEvent.getOriginalID)
+ this._userAgreementHistoryModel += newUserAgreementModel
+
+ // We assume that we always call this method with in-sync events
+ assert(imEvent.getOccurredMillis >= this._latestIMEventOccurredMillis)
+ updateLatestIMEventStateFrom(imEvent)
}
- /**
- * Resource events are processed only if the user has been activated.
- */
- private[this] def shouldProcessResourceEvents: Boolean = {
- haveIMState && this._imState.hasBeenCreated
+// private[this] def updateLatestIMEventIDFrom(imEvent: IMEventMsg): Unit = {
+// this._latestIMEventOriginalID = imEvent.getOriginalID
+// }
+
+ private[this] def updateLatestIMEventStateFrom(imEvent: IMEventMsg) {
+ this._latestIMEventOriginalID = imEvent.getOriginalID
+ this._latestIMEventOccurredMillis = imEvent.getOccurredMillis
+ this._imMsgCount += 1
}
- private[this] def loadUserStateAndUpdateRoleHistory(): Unit = {
- val userCreationMillis = this._imState.userCreationMillis.get
- val initialRole = this._imState.roleHistory.firstRole.get.name
+ private[this] def updateLatestResourceEventIDFrom(rcEvent: ResourceEventMsg): Unit = {
+ this._latestResourceEventOriginalID = rcEvent.getOriginalID
+ }
- val userStateBootstrap = UserStateBootstrappingData(
- this._userID,
- userCreationMillis,
- initialRole,
- aquarium.initialAgreementForRole(initialRole, userCreationMillis),
- aquarium.initialBalanceForRole(initialRole, userCreationMillis)
- )
+ /**
+ * Creates the initial state that is related to IMEvents.
+ *
+ * @return `true` if there was a user CREATE event
+ */
+ private[this] def initializeStateOfIMEvents(): Boolean = {
+ DEBUG("initializeStateOfIMEvents()")
- val userState = userStateComputations.doFullMonthlyBilling(
- userStateBootstrap,
- BillingMonthInfo.fromMillis(TimeHelpers.nowMillis()),
- aquarium.currentResourcesMap,
- InitialUserStateSetup(),
- None
- )
+ // NOTE: this._userID is already set up our caller
+ var _imcounter = 0
- this._userState = userState
+ aquarium.imEventStore.foreachIMEventInOccurrenceOrder(this._userID) { imEvent ⇒
+ _imcounter += 1
+ DEBUG("Replaying [%s] %s", _imcounter, imEvent)
- // Final touch: Update role history
- if(haveIMState && haveUserState) {
- if(this._userState.roleHistory != this._imState.roleHistory) {
- this._userState = newUserStateWithUpdatedRoleHistory(InitialUserActorSetup())
+ if(_imcounter == 1 && !MessageHelpers.isIMEventCreate(imEvent)) {
+ // The very first event must be a CREATE event. Otherwise we abort initialization.
+ // This will normally happen during devops :)
+ INFO("Ignoring first %s since it is not CREATE", shortClassNameOf(imEvent))
+ false
+ }
+ else {
+ updateAgreementHistoryFrom(imEvent)
+ true
}
}
}
- private[this] def createInitialUserState(event: InitializeUserState): Unit = {
- 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
- }
+ private[this] def loadUserStateAndUpdateAgreementHistory(): Unit = {
+ assert(this.haveAgreements, "this.haveAgreements")
- if(!this._imState.hasBeenCreated) {
- // Cannot set the initial state!
- DEBUG("Cannot create %s from %s, since user has not been created", shortNameOfType[UserState], event)
- return
+ if(!haveUserStateBootstrap) {
+ this._userStateBootstrap = aquarium.getUserStateBootstrap(this._userCreationIMEvent)
}
+ logger.debug("#### this._userStateBootStrap %s".format(this._userStateBootstrap.toString))
+ val now = TimeHelpers.nowMillis()
+ this._userState = chargingService.replayMonthChargingUpTo(
+ BillingMonthInfo.fromMillis(now),
+ now,
+ this._userStateBootstrap,
+ aquarium.currentResourceTypesMap,
+ aquarium.userStateStore.insertUserState
+ )
- // We will also need this functionality when receiving IMEvents,
- // so we place it in a method
- loadUserStateAndUpdateRoleHistory()
-
+ // Final touch: Update agreement history in the working user state.
+ // The assumption is that all agreement changes go via IMEvents, so the
+ // state this._workingAgreementHistory is always the authoritative source.
if(haveUserState) {
- DEBUG("%s = %s", shortNameOfType[UserState], this._userState)
+ this._userState.userAgreementHistoryModel = this._userAgreementHistoryModel
+ DEBUG("Computed working user state %s", AvroHelpers.jsonStringOfSpecificRecord(this._userState.msg))
}
}
- def onInitializeUserState(event: InitializeUserState): Unit = {
- this._userID = event.userID
- DEBUG("Got %s", event)
+ private[this] def initializeStateOfResourceEvents(): Unit = {
+ DEBUG("initializeStateOfResourceEvents()")
+ assert(haveAgreements)
- createInitialIMState()
- createInitialUserState(event)
+ // We will also need this functionality when receiving IMEvents, so we place it in a method
+ loadUserStateAndUpdateAgreementHistory()
+
+ if(haveUserState) {
+ DEBUG("Initial working user state %s", AvroHelpers.jsonStringOfSpecificRecord(this._userState.msg))
+ logSeparator()
+ }
}
/**
- * Creates a new user state, taking into account the latest role history in IM state snapshot.
- * Having an IM state snapshot is a prerequisite, otherwise you get an exception; so check before you
- * call this.
+ * Initializes the actor state from DB.
*/
- private[this] def newUserStateWithUpdatedRoleHistory(stateChangeReason: UserStateChangeReason): UserState = {
- this._userState.copy(
- roleHistory = this._imState.roleHistory,
- // FIXME: Also update agreement
- stateChangeCounter = this._userState.stateChangeCounter + 1,
- lastChangeReason = stateChangeReason
- )
+ def initializeUserActorState(userID: String): Boolean = {
+ this._userID = userID
+
+ if(initializeStateOfIMEvents()) {
+ initializeStateOfResourceEvents()
+ // Even if we have no resource events, the user is at least CREATEd
+ true
+ }
+ else {
+ false
+ }
+ }
+
+ def createUserStateBootstrap(imEvent: IMEventMsg) {
+ assert(MessageHelpers.isIMEventCreate(imEvent), "MessageHelpers.isIMEventCreate(imEvent)")
+ assert(this._userCreationIMEvent == imEvent, "this._userCreationIMEvent == imEvent")
+
+ this._userStateBootstrap = aquarium.getUserStateBootstrap(this._userCreationIMEvent)
}
/**
- * 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.
+ * Processes [[gr.grnet.aquarium.message.avro.gen.IMEventMsg]]s that come directly from the
+ * messaging hub (rabbitmq).
*/
- def onProcessIMEvent(processEvent: ProcessIMEvent): Unit = {
- val imEvent = processEvent.imEvent
+ def onIMEventMsg(imEvent: IMEventMsg) {
+ if(!haveAgreements) {
+ // If we have no agreements so far, then it does not matter what kind of event
+ // this is. So we replay the log (ehm.. store)
+ initializeUserActorState(imEvent.getUserID)
+
+ return
+ }
- if(!haveIMState) {
- // This is an error. Should have been initialized from somewhere ...
- throw new AquariumException("Got %s while being uninitialized".format(processEvent))
+ // Check for out of sync (regarding IMEvents)
+ val isOutOfSyncIM = imEvent.getOccurredMillis < this._latestIMEventOccurredMillis
+ if(isOutOfSyncIM) {
+ // clear all resource state
+ // FIXME implement
+
+ return
}
- 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 just after %s birth", imEvent.toDebugString, shortClassNameOf(this))
- logSeparator()
+ // Check out of sync (regarding ResourceEvents)
+ val isOutOfSyncRC = false // FIXME implement
+ if(isOutOfSyncRC) {
+ // TODO
+
return
}
- val (creationTimeChanged,
- activationTimeChanged,
- roleChanged) = _updateIMStateRoleHistory(imEvent, this._imState.roleHistory.lastRoleName)
+ // OK, seems good
+ assert(!MessageHelpers.isIMEventCreate(imEvent), "!MessageHelpers.isIMEventCreate(imEvent)")
+ updateAgreementHistoryFrom(imEvent)
+ }
- DEBUG(
- "(creationTimeChanged, activationTimeChanged, roleChanged)=(%s, %s, %s) using %s",
- creationTimeChanged, activationTimeChanged, roleChanged,
- imEvent
- )
+ def onResourceEventMsg(rcEvent: ResourceEventMsg) {
+ if(!haveAgreements) {
+ DEBUG("No agreement. Ignoring %s", rcEvent)
- // Must also update user state if we know when in history the life of a user begins
- if(creationTimeChanged) {
- if(!haveUserState) {
- loadUserStateAndUpdateRoleHistory()
- INFO("Loaded %s due to %s", shortNameOfType[UserState], imEvent)
- } else {
- // Just update role history
- this._userState = newUserStateWithUpdatedRoleHistory(IMEventArrival(imEvent))
- INFO("Updated %s due to %s", shortNameOfType[UserState], imEvent)
- }
+ return
+ }
+
+ val now = TimeHelpers.nowMillis()
+ // TODO: Review this and its usage in user state.
+ // TODO: The assumption is that the resource set increases all the time,
+ // TODO: so the current map contains everything ever known (assuming we do not run backwards in time).
+ val currentResourcesMap = aquarium.currentResourceTypesMap
+
+ val nowBillingMonthInfo = BillingMonthInfo.fromMillis(now)
+ val nowYear = nowBillingMonthInfo.year
+ val nowMonth = nowBillingMonthInfo.month
+
+ val eventOccurredMillis = rcEvent.getOccurredMillis
+ val eventBillingMonthInfo = BillingMonthInfo.fromMillis(eventOccurredMillis)
+ val eventYear = eventBillingMonthInfo.year
+ val eventMonth = eventBillingMonthInfo.month
+
+ def computeBatch(): Unit = {
+ DEBUG("Going for out of sync charging for %s", rcEvent.getOriginalID)
+ this._userState = chargingService.replayMonthChargingUpTo(
+ nowBillingMonthInfo,
+ // Take into account that the event may be out-of-sync.
+ // TODO: Should we use this._latestResourceEventOccurredMillis instead of now?
+ now max eventOccurredMillis,
+ this._userStateBootstrap,
+ currentResourcesMap,
+ stdUserStateStoreFunc
+ )
+
+ updateLatestResourceEventIDFrom(rcEvent)
}
- DEBUG("New %s = %s", shortNameOfType[IMStateSnapshot], this._imState)
+ def computeRealtime(): Unit = {
+ DEBUG("Going for in sync charging for %s", rcEvent.getOriginalID)
+ chargingService.processResourceEvent(
+ rcEvent,
+ this._userState,
+ nowBillingMonthInfo,
+ true
+ )
+
+ updateLatestResourceEventIDFrom(rcEvent)
+ }
+ val oldTotalCredits =
+ if(this._userState!=null)
+ this._userState.totalCredits
+ else
+ 0.0D
+ // FIXME check these
+ if(this._userState eq null) {
+ computeBatch()
+ }
+ else if(nowYear != eventYear || nowMonth != eventMonth) {
+ DEBUG(
+ "nowYear(%s) != eventYear(%s) || nowMonth(%s) != eventMonth(%s)",
+ nowYear, eventYear,
+ nowMonth, eventMonth
+ )
+ computeBatch()
+ }
+ else if(this._userState.latestResourceEventOccurredMillis < rcEvent.getOccurredMillis) {
+ DEBUG("this._workingUserState.latestResourceEventOccurredMillis < rcEvent.occurredMillis")
+ DEBUG(
+ "%s < %s",
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(this._userState.latestResourceEventOccurredMillis),
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(rcEvent.getOccurredMillis)
+ )
+ computeRealtime()
+ }
+ else {
+ DEBUG("OUT OF ORDER! this._workingUserState.latestResourceEventOccurredMillis=%s and rcEvent.occurredMillis=%s",
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(this._userState.latestResourceEventOccurredMillis),
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(rcEvent.getOccurredMillis))
+
+ computeBatch()
+ }
+ val newTotalCredits = this._userState.totalCredits
+ if(oldTotalCredits * newTotalCredits < 0)
+ aquarium.eventBus ! new BalanceEvent(this._userState.userID,
+ newTotalCredits>=0)
+ DEBUG("Updated %s", this._userState)
logSeparator()
}
- 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)
- logSeparator()
- return
+ def onGetUserBillRequest(event: GetUserBillRequest): Unit = {
+ try{
+ val timeslot = event.timeslot
+ val resourceTypes = aquarium.policyStore.
+ loadSortedPolicyModelsWithin(timeslot.from.getTime,
+ timeslot.to.getTime).
+ values.headOption match {
+ case None => Map[String,ResourceType]()
+ case Some(policy:PolicyModel) => policy.resourceTypesMap
+ }
+ val state= if(haveUserState) Some(this._userState.msg) else None
+ val billEntryMsg = BillEntryMsg.fromWorkingUserState(timeslot,this._userID,state,resourceTypes)
+ //val billEntryMsg = MessageFactory.createBillEntryMsg(billEntry)
+ //logger.debug("BILL ENTRY MSG: " + billEntryMsg.toString)
+ val billData = GetUserBillResponseData(this._userID,billEntryMsg)
+ sender ! GetUserBillResponse(Right(billData))
+ } catch {
+ case e:Exception =>
+ e.printStackTrace()
+ sender ! GetUserBillResponse(Left("Internal Server Error [AQU-BILL-0001]"), 500)
}
}
def onGetUserBalanceRequest(event: GetUserBalanceRequest): Unit = {
val userID = event.userID
- (haveIMState, haveUserState) match {
+ (haveAgreements, haveUserState) match {
case (true, true) ⇒
- // (have IMState, have UserState)
- this._imState.hasBeenActivated match {
- case true ⇒
- // (have IMState, activated, have UserState)
- self reply GetUserBalanceResponse(Right(GetUserBalanceResponseData(userID, this._userState.totalCredits)))
+ // (User CREATEd, with balance state)
+ val realtimeMillis = TimeHelpers.nowMillis()
+ chargingService.calculateRealtimeUserState(
+ this._userState,
+ BillingMonthInfo.fromMillis(realtimeMillis),
+ realtimeMillis
+ )
- case false ⇒
- // (have IMState, not activated, have UserState)
- // Since we have user state, we should have been activated
- self reply GetUserBalanceResponse(Left("Internal Server Error [AQU-BAL-0001]"), 500)
- }
+ sender ! GetUserBalanceResponse(Right(GetUserBalanceResponseData(this._userID, this._userState.totalCredits)))
case (true, false) ⇒
- // (have IMState, no UserState)
- this._imState.hasBeenActivated match {
- case true ⇒
- // (have IMState, activated, no UserState)
- // Since we are activated, we should have some state.
- self reply GetUserBalanceResponse(Left("Internal Server Error [AQU-BAL-0002]"), 500)
- case false ⇒
- // (have IMState, not activated, no UserState)
- // The user is virtually unknown
- self reply GetUserBalanceResponse(Left("User %s not activated [AQU-BAL-0003]".format(userID)), 404 /*Not found*/)
- }
+ // (User CREATEd, no balance state)
+ // Return the default initial balance
+ sender ! GetUserBalanceResponse(
+ Right(
+ GetUserBalanceResponseData(
+ this._userID,
+ aquarium.initialUserBalance(this._userCreationIMEvent.getRole, this._userCreationIMEvent.getOccurredMillis)
+ )))
case (false, true) ⇒
- // (no IMState, have UserState
- // A bit ridiculous situation
- self reply GetUserBalanceResponse(Left("Unknown user %s [AQU-BAL-0004]".format(userID)), 404/*Not found*/)
+ // (Not CREATEd, with balance state)
+ // Clearly this is internal error
+ sender ! GetUserBalanceResponse(Left("Internal Server Error [AQU-BAL-0001]"), 500)
case (false, false) ⇒
- // (no IMState, no UserState)
- self reply GetUserBalanceResponse(Left("Unknown user %s [AQU-BAL-0005]".format(userID)), 404/*Not found*/)
+ // (Not CREATEd, no balance state)
+ // The user is completely unknown
+ sender ! GetUserBalanceResponse(Left("Unknown user %s [AQU-BAL-0004]".format(userID)), 404/*Not found*/)
}
}
def onGetUserStateRequest(event: GetUserStateRequest): Unit = {
haveUserState match {
case true ⇒
- self reply GetUserStateResponse(Right(this._userState))
+ val realtimeMillis = TimeHelpers.nowMillis()
+ chargingService.calculateRealtimeUserState(
+ this._userState,
+ BillingMonthInfo.fromMillis(realtimeMillis),
+ realtimeMillis
+ )
+
+ sender ! GetUserStateResponse(Right(this._userState.msg))
case false ⇒
- self reply GetUserStateResponse(Left("No state for user %s [AQU-STA-0006]".format(event.userID)))
+ sender ! GetUserStateResponse(Left("No state for user %s [AQU-STA-0006]".format(event.userID)), 404)
+ }
+ }
+
+ def onGetUserWalletRequest(event: GetUserWalletRequest): Unit = {
+ haveUserState match {
+ case true ⇒
+ DEBUG("haveWorkingUserState: %s", event)
+ val realtimeMillis = TimeHelpers.nowMillis()
+ chargingService.calculateRealtimeUserState(
+ this._userState,
+ BillingMonthInfo.fromMillis(realtimeMillis),
+ realtimeMillis
+ )
+
+ sender ! GetUserWalletResponse(
+ Right(
+ GetUserWalletResponseData(
+ this._userID,
+ this._userState.totalCredits,
+ MessageFactory.newWalletEntriesMsg(this._userState.msg.getWalletEntries)
+ )))
+
+ case false ⇒
+ DEBUG("!haveWorkingUserState: %s", event)
+ haveAgreements match {
+ case true ⇒
+ DEBUG("haveAgreements: %s", event)
+ sender ! GetUserWalletResponse(
+ Right(
+ GetUserWalletResponseData(
+ this._userID,
+ aquarium.initialUserBalance(this._userCreationIMEvent.getRole, this._userCreationIMEvent.getOccurredMillis),
+ MessageFactory.newWalletEntriesMsg()
+ )))
+
+ case false ⇒
+ DEBUG("!haveUserCreationIMEvent: %s", event)
+ sender ! GetUserWalletResponse(Left("No wallet for user %s [AQU-WAL-00 8]".format(event.userID)), 404)
+ }
}
}
projects / aquarium / blobdiff