package service
package user
-import gr.grnet.aquarium.actor._
-
-import gr.grnet.aquarium.actor.message.event.{ProcessResourceEvent, ProcessIMEvent}
-import gr.grnet.aquarium.actor.message.config.{InitializeUserActorState, AquariumPropertiesLoaded}
-import gr.grnet.aquarium.util.date.TimeHelpers
+import gr.grnet.aquarium.{Real, 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
+import gr.grnet.aquarium.message.avro.gen.{UserAgreementHistoryMsg, IMEventMsg, ResourceEventMsg}
+import gr.grnet.aquarium.message.avro.{ModelFactory, MessageFactory, MessageHelpers}
import gr.grnet.aquarium.service.event.BalanceEvent
-import gr.grnet.aquarium.event.model.im.IMEventModel
-import message._
-import config.AquariumPropertiesLoaded
-import config.InitializeUserActorState
-import event.ProcessIMEvent
-import event.ProcessResourceEvent
+import gr.grnet.aquarium.util.date.TimeHelpers
import gr.grnet.aquarium.util.{LogHelpers, shortClassNameOf}
-import gr.grnet.aquarium.{Aquarium, AquariumInternalError}
-import gr.grnet.aquarium.computation.BillingMonthInfo
-import gr.grnet.aquarium.charging.state.UserStateBootstrap
-import gr.grnet.aquarium.charging.state.{WorkingAgreementHistory, WorkingUserState, UserStateModel}
-import gr.grnet.aquarium.charging.reason.{InitialUserActorSetup, RealtimeChargingReason}
-import gr.grnet.aquarium.policy.{PolicyDefinedFullPriceTableRef, StdUserAgreement}
-import gr.grnet.aquarium.event.model.resource.{StdResourceEvent, ResourceEventModel}
-import message.GetUserBalanceRequest
-import message.GetUserBalanceResponse
-import message.GetUserBalanceResponseData
-import message.GetUserStateRequest
-import message.GetUserStateResponse
-import message.GetUserWalletRequest
-import message.GetUserWalletResponse
-import message.GetUserWalletResponseData
-import scala.Left
-import gr.grnet.aquarium.charging.state.WorkingAgreementHistory
-import scala.Some
-import scala.Right
-import gr.grnet.aquarium.policy.StdUserAgreement
-import gr.grnet.aquarium.charging.state.UserStateBootstrap
-import gr.grnet.aquarium.charging.bill.{AbstractBillEntry, BillEntry}
+import gr.grnet.aquarium.policy.{ResourceType, PolicyModel}
+import gr.grnet.aquarium.charging.bill.BillEntryMsg
/**
*
*/
class UserActor extends ReflectiveRoleableActor {
- private[this] var _userID: String = "<?>"
- private[this] var _workingUserState: WorkingUserState = _
- private[this] var _userCreationIMEvent: IMEventModel = _
- private[this] val _workingAgreementHistory: WorkingAgreementHistory = new WorkingAgreementHistory
- private[this] var _latestIMEventID: String = ""
- private[this] var _latestResourceEventID: String = ""
- private[this] var _userStateBootstrap: UserStateBootstrap = _
-
- def unsafeUserID = {
- if(!haveUserID) {
+ private[this] var _imMsgCount = 0
+ private[this] var _userStateModel: UserStateModel = _
+
+ def userID = {
+ if(!haveUserState) {
throw new AquariumInternalError("%s not initialized")
}
- this._userID
+ this._userStateModel.userID
}
override def postStop() {
}
private[this] def shutmedown(): Unit = {
- if(haveUserID) {
+ if(haveUserState) {
aquarium.akkaService.invalidateUserActor(this)
}
}
private[this] def chargingService = aquarium.chargingService
- private[this] def stdUserStateStoreFunc = (userState: UserStateModel) ⇒ {
- aquarium.userStateStore.insertUserState(userState)
+ def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
}
- @inline private[this] def haveUserID = {
- this._userID ne null
+ private[this] def unsafeUserCreationIMEventMsg = {
+ this._userStateModel.unsafeUserCreationIMEvent
}
- @inline private[this] def haveUserCreationIMEvent = {
- this._userCreationIMEvent ne null
+ private[this] def haveAgreements = {
+ (this._userStateModel ne null)
}
- def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
+ private[this] def haveUserCreationEvent = {
+ haveAgreements &&
+ this._userStateModel.hasUserCreationEvent
}
- @inline private[this] def haveAgreements = {
- this._workingAgreementHistory.size > 0
+ private[this] def haveUserState = {
+ (this._userStateModel ne null)
}
- @inline private[this] def haveWorkingUserState = {
- this._workingUserState ne null
- }
+ private[this] def isInitial = this._userStateModel.isInitial
- @inline private[this] def haveUserStateBootstrap = {
- this._userStateBootstrap ne null
- }
+ /**
+ * Creates the agreement history from all the stored IMEvents.
+ *
+ * @return (`true` iff there was a user CREATE event, the number of events processed)
+ */
+ private[this] def createUserAgreementHistoryFromIMEvents(userID: String): (Boolean, Int) = {
+ DEBUG("createUserAgreementHistoryFromStoredIMEvents()")
+ assert(haveUserState, "haveUserState")
- private[this] def updateAgreementHistoryFrom(imEvent: IMEventModel): Unit = {
- if(imEvent.isCreateUser) {
- if(haveUserCreationIMEvent) {
- throw new AquariumInternalError(
- "Got user creation event (id=%s) but I already have one (id=%s)",
- this._userCreationIMEvent.id,
- imEvent.id
- )
+
+ var _imcounter = 0
+
+ val hadCreateEvent = aquarium.imEventStore.foreachIMEventInOccurrenceOrder(userID) { imEvent ⇒
+ _imcounter += 1
+ DEBUG("Replaying [%s/%s] %s", shortClassNameOf(imEvent), _imcounter, imEvent)
+
+ if(_imcounter == 1 && !MessageHelpers.isUserCreationIMEvent(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 {
+ val effectiveFromMillis = imEvent.getOccurredMillis
+ val role = imEvent.getRole
+ // calling unsafe just for the side-effect
+ assert(
+ aquarium.unsafeFullPriceTableForRoleAt(role, effectiveFromMillis) ne null,
+ "aquarium.unsafeFullPriceTableForRoleAt(%s, %s) ne null".format(role, effectiveFromMillis)
+ )
- this._userCreationIMEvent = imEvent
+ this._userStateModel.insertUserAgreementMsgFromIMEvent(imEvent)
+ true
+ }
}
- val effectiveFromMillis = imEvent.occurredMillis
- val role = imEvent.role
- // calling unsafe just for the side-effect
- assert(null ne aquarium.unsafePriceTableForRoleAt(role, effectiveFromMillis))
-
- val newAgreement = StdUserAgreement(
- imEvent.id,
- Some(imEvent.id),
- effectiveFromMillis,
- Long.MaxValue,
- role,
- PolicyDefinedFullPriceTableRef
- )
+ this._imMsgCount = _imcounter
- this._workingAgreementHistory += newAgreement
+ DEBUG("Agreements: %s", this._userStateModel.userAgreementHistoryMsg)
+ (hadCreateEvent, _imcounter)
}
- private[this] def updateLatestIMEventIDFrom(imEvent: IMEventModel): Unit = {
- this._latestIMEventID = imEvent.id
+ private[this] def saveFirstUserState(userID: String) {
+ this._userStateModel.userStateMsg.setIsFirst(true)
+ this._userStateModel.updateUserStateMsg(
+ aquarium.userStateStore.insertUserState(this._userStateModel.userStateMsg)
+ )
}
- private[this] def updateLatestResourceEventIDFrom(rcEvent: ResourceEventModel): Unit = {
- this._latestResourceEventID = rcEvent.id
+ private[this] def saveSubsequentUserState() {
+ this._userStateModel.userStateMsg.setIsFirst(false)
+ this._userStateModel.updateUserStateMsg(
+ aquarium.userStateStore.insertUserState(this._userStateModel.userStateMsg)
+ )
}
- /**
- * Creates the initial state that is related to IMEvents.
- */
- private[this] def initializeStateOfIMEvents(): Unit = {
- // NOTE: this._userID is already set up by onInitializeUserActorState()
- aquarium.imEventStore.foreachIMEventInOccurrenceOrder(this._userID) { imEvent ⇒
- DEBUG("Replaying %s", imEvent)
+ private[this] def loadLastKnownUserStateAndUpdateAgreements() {
+ val userID = this._userStateModel.userID
+ aquarium.userStateStore.findLatestUserState(userID) match {
+ case None ⇒
+ // First user state ever
+ saveFirstUserState(userID)
- updateAgreementHistoryFrom(imEvent)
- updateLatestIMEventIDFrom(imEvent)
+ case Some(latestUserState) ⇒
+ this._userStateModel.updateUserStateMsg(latestUserState)
}
+ }
- if(haveAgreements) {
- DEBUG("Initial agreement history %s", this._workingAgreementHistory.toJsonString)
- logSeparator()
- }
+ private[this] def processResourceEventsAfterLastKnownUserState() {
+ // Update the user state snapshot with fresh (ie not previously processed) events.
+ aquarium.resourceEventStore.foreachResourceEventOccurredInPeriod(
+ this._userStateModel.userID,
+ this._userStateModel.latestResourceEventOccurredMillis,
+ TimeHelpers.nowMillis()
+ )
}
- /**
- * Resource events are processed only if the user has been created and has agreements.
- * Otherwise nothing can be computed.
- */
- private[this] def shouldProcessResourceEvents: Boolean = {
- haveUserCreationIMEvent && haveAgreements && haveUserStateBootstrap
+ private[this] def makeUserStateMsgUpToDate() {
+ loadLastKnownUserStateAndUpdateAgreements()
+ processResourceEventsAfterLastKnownUserState()
}
- private[this] def loadWorkingUserStateAndUpdateAgreementHistory(): Unit = {
- assert(this.haveAgreements, "this.haveAgreements")
- assert(this.haveUserCreationIMEvent, "this.haveUserCreationIMEvent")
-
- val userCreationMillis = this._userCreationIMEvent.occurredMillis
- val userCreationRole = this._userCreationIMEvent.role // initial role
- val userCreationIMEventID = this._userCreationIMEvent.id
-
- if(!haveUserStateBootstrap) {
- this._userStateBootstrap = UserStateBootstrap(
- this._userID,
- userCreationMillis,
- aquarium.initialUserAgreement(userCreationRole, userCreationMillis, Some(userCreationIMEventID)),
- aquarium.initialUserBalance(userCreationRole, userCreationMillis)
- )
+ private[this] def checkInitial(nextThing: () ⇒ Any = () ⇒ {}): Boolean = {
+ if(!isInitial) {
+ return false
}
- val now = TimeHelpers.nowMillis()
- this._workingUserState = chargingService.replayMonthChargingUpTo(
- BillingMonthInfo.fromMillis(now),
- now,
- this._userStateBootstrap,
- aquarium.currentResourceTypesMap,
- InitialUserActorSetup(),
- aquarium.userStateStore.insertUserState
- )
+ val (userCreated, imEventsCount) = createUserAgreementHistoryFromIMEvents(userID)
- // 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(haveWorkingUserState) {
- this._workingUserState.workingAgreementHistory.setFrom(this._workingAgreementHistory)
- DEBUG("Computed working user state %s", this._workingUserState.toJsonString)
+ if(userCreated) {
+ makeUserStateMsgUpToDate()
}
+
+ nextThing()
+
+ true
}
- private[this] def initializeStateOfResourceEvents(event: InitializeUserActorState): Unit = {
- if(!this.haveAgreements) {
- DEBUG("Cannot initializeResourceEventsState() from %s. There are no agreements", event)
+ /**
+ * Processes [[gr.grnet.aquarium.message.avro.gen.IMEventMsg]]s that come directly from the
+ * messaging hub (rabbitmq).
+ */
+ def onIMEventMsg(imEvent: IMEventMsg) {
+ if(checkInitial()) {
return
}
- if(!this.haveUserCreationIMEvent) {
- DEBUG("Cannot initializeResourceEventsState() from %s. I never got a CREATE IMEvent", event)
+ // Check for out of sync (regarding IMEvents)
+ val isOutOfSyncIM = imEvent.getOccurredMillis < this._userStateModel.latestIMEventOccurredMillis
+ if(isOutOfSyncIM) {
+ // clear all resource state
+ // FIXME implement
+
return
}
- // We will also need this functionality when receiving IMEvents, so we place it in a method
- loadWorkingUserStateAndUpdateAgreementHistory()
+ // Check out of sync (regarding ResourceEvents)
+ val isOutOfSyncRC = false // FIXME implement
+ if(isOutOfSyncRC) {
+ // TODO
- if(haveWorkingUserState) {
- DEBUG("Initial working user state %s", this._workingUserState.toJsonString)
- logSeparator()
+ return
}
- }
- def onInitializeUserActorState(event: InitializeUserActorState): Unit = {
- this._userID = event.userID
- DEBUG("Got %s", event)
+ // Make new agreement
+ this._userStateModel.insertUserAgreementMsgFromIMEvent(imEvent)
+ this._imMsgCount += 1
- initializeStateOfIMEvents()
- initializeStateOfResourceEvents(event)
- }
-
- /**
- * 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 hadUserCreationIMEvent = haveUserCreationIMEvent
-
- if(!haveAgreements) {
- // This IMEvent has arrived after any ResourceEvents
- INFO("Arrived after any ResourceEvent: %s", imEvent.toDebugString)
- initializeStateOfIMEvents()
- }
- else {
- if(this._latestIMEventID == 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", imEvent.toDebugString)
- logSeparator()
-
- //this._latestIMEventID = imEvent.id
- return
- }
- if(imEvent.isAddCredits) {
- if(!hadUserCreationIMEvent && haveUserCreationIMEvent)
- loadWorkingUserStateAndUpdateAgreementHistory()
- onHandleAddCreditsEvent(imEvent)
-
- } else {
- updateAgreementHistoryFrom(imEvent)
- updateLatestIMEventIDFrom(imEvent)
- }
- }
-
- // Must also update user state if we know when in history the life of a user begins
- if(!hadUserCreationIMEvent && haveUserCreationIMEvent) {
- INFO("Processing user state, since we had a CREATE IMEvent")
- loadWorkingUserStateAndUpdateAgreementHistory()
+ if(MessageHelpers.isUserCreationIMEvent(imEvent)) {
+ makeUserStateMsgUpToDate()
}
- logSeparator()
+ DEBUG("Agreements: %s", this._userStateModel.userAgreementHistoryMsg)
}
- /* Convert astakos message for adding credits
- to a regular RESOURCE message */
- def onHandleAddCreditsEvent(imEvent : IMEventModel) = {
- val credits = imEvent.details(IMEventModel.DetailsNames.credits).toInt.toDouble
- val event = new StdResourceEvent(
- imEvent.id,
- imEvent.occurredMillis,
- imEvent.receivedMillis,
- imEvent.userID,
- imEvent.clientID,
- imEvent.eventType,
- imEvent.eventType,
- credits,
- imEvent.eventVersion,
- imEvent.details
- )
- //Console.err.println("Event: " + event)
- //Console.err.println("Total credits before: " + _workingUserState.totalCredits)
- onProcessResourceEvent(new ProcessResourceEvent(event))
- //Console.err.println("Total credits after: " + _workingUserState.totalCredits)
- //Console.err.println("OK.")
- }
-
- def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
- val rcEvent = event.rcEvent
-
- if(!shouldProcessResourceEvents) {
- // This means the user has not been created (at least, as far as Aquarium is concerned).
- // So, we do not process any resource event
- DEBUG("Not processing %s", rcEvent.toJsonString)
- logSeparator()
-
+ def onResourceEventMsg(rcEvent: ResourceEventMsg) {
+ if(checkInitial()) {
return
}
- // Since the latest resource event per resource is recorded in the user state,
- // we do not need to query the store. Just query the in-memory state.
- // Note: This is a similar situation with the first IMEvent received right after the user
- // actor is created.
- if(this._latestResourceEventID == rcEvent.id) {
- INFO("Ignoring first %s", rcEvent.toDebugString)
- logSeparator()
+ if(!haveUserCreationEvent) {
+ DEBUG("No agreements. Ignoring %s", rcEvent)
return
}
- val now = TimeHelpers.nowMillis()
- val currentResourcesMap = aquarium.currentResourceTypesMap
- val chargingReason = RealtimeChargingReason(None, now)
-
- val nowBillingMonthInfo = BillingMonthInfo.fromMillis(now)
- val nowYear = nowBillingMonthInfo.year
- val nowMonth = nowBillingMonthInfo.month
-
- val eventOccurredMillis = rcEvent.occurredMillis
- val eventBillingMonthInfo = BillingMonthInfo.fromMillis(eventOccurredMillis)
- val eventYear = eventBillingMonthInfo.year
- val eventMonth = eventBillingMonthInfo.month
-
- def computeBatch(): Unit = {
- DEBUG("Going for out of sync charging")
- this._workingUserState = 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,
- chargingReason,
- stdUserStateStoreFunc
- )
-
- updateLatestResourceEventIDFrom(rcEvent)
- }
+ assert(haveUserState, "haveUserState")
- def computeRealtime(): Unit = {
- DEBUG("Going for in sync charging")
- chargingService.processResourceEvent(
- rcEvent,
- this._workingUserState,
- chargingReason,
- nowBillingMonthInfo,
- true
- )
+ val oldTotalCredits = this._userStateModel.totalCreditsAsReal
- updateLatestResourceEventIDFrom(rcEvent)
- }
+ chargingService.processResourceEvent(
+ rcEvent.getReceivedMillis,
+ rcEvent,
+ this._userStateModel,
+ aquarium.currentResourceMapping,
+ true
+ )
- val oldTotalCredits = this._workingUserState.totalCredits
- // FIXME check these
- if(nowYear != eventYear || nowMonth != eventMonth) {
- DEBUG(
- "nowYear(%s) != eventYear(%s) || nowMonth(%s) != eventMonth(%s)",
- nowYear, eventYear,
- nowMonth, eventMonth
- )
- computeBatch()
- }
- else if(this._workingUserState.latestResourceEventOccurredMillis < rcEvent.occurredMillis) {
- DEBUG("this._workingUserState.latestResourceEventOccurredMillis < rcEvent.occurredMillis")
- DEBUG(
- "%s < %s",
- TimeHelpers.toYYYYMMDDHHMMSSSSS(this._workingUserState.latestResourceEventOccurredMillis),
- TimeHelpers.toYYYYMMDDHHMMSSSSS(rcEvent.occurredMillis)
- )
- computeRealtime()
- }
- else {
- computeBatch()
+ val newTotalCredits = this._userStateModel.totalCreditsAsReal
+
+ if(oldTotalCredits.signum * newTotalCredits.signum < 0) {
+ aquarium.eventBus ! new BalanceEvent(userID, newTotalCredits >= 0)
}
- val newTotalCredits = this._workingUserState.totalCredits
- if(oldTotalCredits * newTotalCredits < 0)
- aquarium.eventBus ! new BalanceEvent(this._workingUserState.userID,
- newTotalCredits>=0)
- DEBUG("Updated %s", this._workingUserState)
- logSeparator()
+
+ DEBUG("Updated %s", this._userStateModel)
}
def onGetUserBillRequest(event: GetUserBillRequest): Unit = {
+ checkInitial()
+
try{
val timeslot = event.timeslot
- val state= if(haveWorkingUserState) Some(this._workingUserState) else None
- val billEntry = AbstractBillEntry.fromWorkingUserState(timeslot,this._userID,state)
- val billData = GetUserBillResponseData(this._userID,billEntry)
+ val resourceTypes: Map[String, ResourceType] = 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._userStateModel.userStateMsg) 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)
+ e.printStackTrace()
+ sender ! GetUserBillResponse(Left("Internal Server Error [AQU-BILL-0001]"), 500)
}
}
def onGetUserBalanceRequest(event: GetUserBalanceRequest): Unit = {
+ checkInitial()
+
val userID = event.userID
- (haveUserCreationIMEvent, haveWorkingUserState) match {
+ (haveAgreements, haveUserState) match {
case (true, true) ⇒
// (User CREATEd, with balance state)
- sender ! GetUserBalanceResponse(Right(GetUserBalanceResponseData(this._userID, this._workingUserState.totalCredits)))
+ val realtimeMillis = TimeHelpers.nowMillis()
+ chargingService.calculateRealtimeUserState(
+ this._userStateModel,
+ aquarium.currentResourceMapping,
+ realtimeMillis
+ )
+
+ sender ! GetUserBalanceResponse(Right(GetUserBalanceResponseData(this.userID, this._userStateModel.totalCredits)))
case (true, false) ⇒
// (User CREATEd, no balance state)
sender ! GetUserBalanceResponse(
Right(
GetUserBalanceResponseData(
- this._userID,
- aquarium.initialUserBalance(this._userCreationIMEvent.role, this._userCreationIMEvent.occurredMillis)
- )))
+ this.userID,
+ aquarium.initialUserBalance(this.unsafeUserCreationIMEventMsg.getRole, this.unsafeUserCreationIMEventMsg.getOccurredMillis).toString()
+ )))
case (false, true) ⇒
// (Not CREATEd, with balance state)
}
def onGetUserStateRequest(event: GetUserStateRequest): Unit = {
- haveWorkingUserState match {
+ checkInitial()
+
+ haveUserState match {
case true ⇒
- sender ! GetUserStateResponse(Right(this._workingUserState))
+ val realtimeMillis = TimeHelpers.nowMillis()
+ chargingService.calculateRealtimeUserState(
+ this._userStateModel,
+ aquarium.currentResourceMapping,
+ realtimeMillis
+ )
+
+ sender ! GetUserStateResponse(Right(this._userStateModel.userStateMsg))
case false ⇒
sender ! GetUserStateResponse(Left("No state for user %s [AQU-STA-0006]".format(event.userID)), 404)
}
def onGetUserWalletRequest(event: GetUserWalletRequest): Unit = {
- haveWorkingUserState match {
+ checkInitial()
+
+ haveUserState match {
case true ⇒
DEBUG("haveWorkingUserState: %s", event)
+ val realtimeMillis = TimeHelpers.nowMillis()
+ chargingService.calculateRealtimeUserState(
+ this._userStateModel,
+ aquarium.currentResourceMapping,
+ realtimeMillis
+ )
+
sender ! GetUserWalletResponse(
Right(
GetUserWalletResponseData(
- this._userID,
- this._workingUserState.totalCredits,
- this._workingUserState.walletEntries.toList
- )))
+ this.userID,
+ this._userStateModel.totalCredits,
+ MessageFactory.newWalletEntriesMsg(this._userStateModel.userStateMsg.getWalletEntries)
+ )))
case false ⇒
DEBUG("!haveWorkingUserState: %s", event)
- haveUserCreationIMEvent match {
+ haveAgreements match {
case true ⇒
- DEBUG("haveUserCreationIMEvent: %s", event)
+ DEBUG("haveAgreements: %s", event)
sender ! GetUserWalletResponse(
Right(
GetUserWalletResponseData(
- this._userID,
- aquarium.initialUserBalance(this._userCreationIMEvent.role, this._userCreationIMEvent.occurredMillis),
- Nil
- )))
+ this.userID,
+ Real.toMsgField(aquarium.initialUserBalance(this.unsafeUserCreationIMEventMsg.getRole, this.unsafeUserCreationIMEventMsg.getOccurredMillis)),
+ MessageFactory.newWalletEntriesMsg()
+ )))
case false ⇒
DEBUG("!haveUserCreationIMEvent: %s", event)
}
}
+ /**
+ * Initializes the actor's internal state.
+ *
+ * @param userID
+ */
+ def onSetUserActorUserID(userID: String) {
+ // Create the full agreement history from the original sources (IMEvents)
+ this._userStateModel = ModelFactory.newInitialUserStateModel(
+ userID,
+ Real(0),
+ TimeHelpers.nowMillis()
+ )
+
+ require(this._userStateModel.isInitial, "this._userStateModel.isInitial")
+ }
+
private[this] def D_userID = {
- this._userID
+ if(haveUserState) userID else "???"
}
private[this] def DEBUG(fmt: String, args: Any*) =
projects / aquarium / blobdiff