package service
package user
-import java.util.Date
-import com.ckkloverdos.maybe.{Failed, NoVal, Just}
-
-import gr.grnet.aquarium.actor._
-import gr.grnet.aquarium.Configurator
-import gr.grnet.aquarium.user._
-
-import gr.grnet.aquarium.util.Loggable
+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.actor.message.config.InitializeUserActorState
+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.logic.accounting.RoleAgreements
-import gr.grnet.aquarium.messaging.AkkaAMQP
-import gr.grnet.aquarium.actor.message.config.user.UserActorInitWithUserId
-import gr.grnet.aquarium.actor.message.service.dispatcher._
-import message.config.{ActorProviderConfigured, AquariumPropertiesLoaded}
-import gr.grnet.aquarium.event.im.IMEventModel
-import gr.grnet.aquarium.event.{WalletEntry}
-
+import gr.grnet.aquarium.util.{LogHelpers, shortClassNameOf}
+import gr.grnet.aquarium.policy.{ResourceType, PolicyModel}
+import gr.grnet.aquarium.charging.bill.BillEntryMsg
/**
*
* @author Christos KK Loverdos <loverdos@gmail.com>
*/
-class UserActor extends AquariumActor
-with AkkaAMQP
-with ReflectiveAquariumActor
-with Loggable {
- @volatile
- private[this] var _userId: String = _
- @volatile
- private[this] var _userState: UserState = _
- @volatile
- private[this] var _timestampTheshold: Long = _
+class UserActor extends ReflectiveRoleableActor {
+ private[this] var _userID: String = "<?>"
+ private[this] var _userState: UserStateModel = _
+ private[this] var _userCreationIMEvent: IMEventMsg = _
+ private[this] var _userAgreementHistoryModel: UserAgreementHistoryModel = _
+ private[this] var _latestIMEventOriginalID: String = ""
+ private[this] var _latestResourceEventOriginalID: String = ""
+ private[this] var _userStateBootstrap: UserStateBootstrap = _
+
+ def unsafeUserID = {
+ if(!haveUserID) {
+ throw new AquariumInternalError("%s not initialized")
+ }
- private[this] lazy val messenger = producer("aquarium") // FIXME: Read this from configuration
+ this._userID
+ }
- def role = UserActorRole
+ override def postStop() {
+ DEBUG("I am finally stopped (in postStop())")
+ aquarium.akkaService.notifyUserActorPostStop(this)
+ }
+
+ private[this] def shutmedown(): Unit = {
+ if(haveUserID) {
+ aquarium.akkaService.invalidateUserActor(this)
+ }
+ }
- private[this] def _configurator: Configurator = Configurator.MasterConfigurator
+ override protected def onThrowable(t: Throwable, message: AnyRef) = {
+ LogHelpers.logChainOfCauses(logger, t)
+ ERROR(t, "Terminating due to: %s(%s)", shortClassNameOf(t), t.getMessage)
- /**
- * Replay the event log for all events that affect the user state.
- */
- def rebuildState(from: Long, to: Long): Unit = {
- val start = TimeHelpers.nowMillis()
- if(_userState == null)
- createBlankState
-
- //Rebuild state from user events
- val usersDB = _configurator.storeProvider.imEventStore
- val userEvents = usersDB.findIMEventsByUserId(_userId)
- val numUserEvents = userEvents.size
- _userState = replayIMEvents(_userState, userEvents, from, to)
-
- //Rebuild state from resource events
- val eventsDB = _configurator.storeProvider.resourceEventStore
- val resourceEvents = eventsDB.findResourceEventsByUserIdAfterTimestamp(_userId, from)
- val numResourceEvents = resourceEvents.size
- // _userState = replayResourceEvents(_userState, resourceEvents, from, to)
-
- //Rebuild state from wallet entries
- val wallet = _configurator.storeProvider.walletEntryStore
- val walletEnties = wallet.findWalletEntriesAfter(_userId, new Date(from))
- val numWalletEntries = walletEnties.size
- _userState = replayWalletEntries(_userState, walletEnties, from, to)
-
- INFO(("Rebuilt state from %d events (%d user events, " +
- "%d resource events, %d wallet entries) in %d msec").format(
- numUserEvents + numResourceEvents + numWalletEntries,
- numUserEvents, numResourceEvents, numWalletEntries,
- TimeHelpers.nowMillis() - start))
+ shutmedown()
}
- /**
- * Create an empty state for a user
- */
- def createBlankState = {
- this._userState = DefaultUserStateComputations.createInitialUserState(this._userId, 0L, true, 0.0)
+ def role = UserActorRole
+
+ private[this] def chargingService = aquarium.chargingService
+
+ private[this] def stdUserStateStoreFunc = (userState: UserStateMsg) ⇒ {
+ aquarium.userStateStore.insertUserState(userState)
}
- /**
- * Replay user events on the provided user state
- */
- def replayIMEvents(initState: UserState, events: List[IMEventModel],
- from: Long, to: Long): UserState = {
- initState
+ @inline private[this] def haveUserID = {
+ this._userID ne null
}
+ @inline private[this] def haveUserCreationIMEvent = {
+ this._userCreationIMEvent ne null
+ }
- /**
- * Replay wallet entries on the provided user state
- */
- def replayWalletEntries(initState: UserState, events: List[WalletEntry],
- from: Long, to: Long): UserState = {
- initState
+ def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
}
- /**
- * Persist current user state
- */
- private[this] def saveUserState(): Unit = {
- _configurator.storeProvider.userStateStore.storeUserState(this._userState) match {
- case Just(record) => record
- case NoVal => ERROR("Unknown error saving state")
- case Failed(e) =>
- ERROR("Saving state failed: %s".format(e));
- }
+ @inline private[this] def haveAgreements = {
+ (this._userAgreementHistoryModel ne null) && this._userAgreementHistoryModel.size > 0
}
- def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
- this._timestampTheshold = event.props.getLong(Configurator.Keys.user_state_timestamp_threshold).getOr(10000)
- INFO("Setup my timestampTheshold = %s", this._timestampTheshold)
+ @inline private[this] def haveUserState = {
+ this._userState ne null
}
- def onUserActorInitWithUserId(event: UserActorInitWithUserId): Unit = {
- this._userId = event.userId
- DEBUG("Actor starting, loading state")
+ @inline private[this] def haveUserStateBootstrap = {
+ this._userStateBootstrap ne null
}
- def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
- val resourceEvent = event.rcEvent
- if(resourceEvent.userID != this._userId) {
- ERROR("Received %s but my userId = %s".format(event, this._userId))
+ private[this] def updateAgreementHistoryFrom(imEvent: IMEventMsg): Unit = {
+ val isCreateUser = MessageHelpers.isIMEventCreate(imEvent)
+ if(isCreateUser) {
+ if(haveUserCreationIMEvent) {
+ throw new AquariumInternalError(
+ "Got user creation event (id=%s) but I already have one (id=%s)",
+ this._userCreationIMEvent.getOriginalID,
+ imEvent.getOriginalID
+ )
+ }
+
+ this._userCreationIMEvent = imEvent
+ }
+
+ 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)
+ if(this._userAgreementHistoryModel eq null) {
+ this._userAgreementHistoryModel = ModelFactory.newUserAgreementHistoryModelFromIMEvent(imEvent, imEvent.getOriginalID)
} else {
- //ensureUserState()
- // calcWalletEntries()
- //processResourceEvent(resourceEvent, true)
+ val newUserAgreementModel = ModelFactory.newUserAgreementModelFromIMEvent(imEvent, imEvent.getOriginalID)
+ this._userAgreementHistoryModel += newUserAgreementModel
}
}
- private[this] def processCreateUser(event: IMEventModel): Unit = {
- val userId = event.userID
- DEBUG("Creating user from state %s", event)
- val usersDB = _configurator.storeProvider.userStateStore
- usersDB.findUserStateByUserId(userId) match {
- case Just(userState) ⇒
- WARN("User already created, state = %s".format(userState))
- case failed@Failed(e) ⇒
- ERROR("[%s] %s", e.getClass.getName, e.getMessage)
- case NoVal ⇒
- val agreement = RoleAgreements.agreementForRole(event.role)
- DEBUG("User %s assigned agreement %s".format(userId, agreement.name))
-
- this._userState = DefaultUserStateComputations.createInitialUserState(
- userId,
- event.occurredMillis,
- event.isActive, 0.0, List(event.role), agreement.name)
- saveUserState
- DEBUG("Created and stored %s", this._userState)
+ private[this] def updateLatestIMEventIDFrom(imEvent: IMEventMsg): Unit = {
+ this._latestIMEventOriginalID = imEvent.getOriginalID
+ }
+
+ private[this] def updateLatestResourceEventIDFrom(rcEvent: ResourceEventMsg): Unit = {
+ this._latestResourceEventOriginalID = rcEvent.getOriginalID
+ }
+
+ /**
+ * 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)
+
+ updateAgreementHistoryFrom(imEvent)
+ updateLatestIMEventIDFrom(imEvent)
}
+
+ if(haveAgreements) {
+ DEBUG("Initial agreement history %s", this._userAgreementHistoryModel.toJsonString)
+ logSeparator()
+ }
+ }
+
+ /**
+ * 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 processModifyUser(event: IMEventModel): Unit = {
+ private[this] def loadUserStateAndUpdateAgreementHistory(): Unit = {
+ assert(this.haveAgreements, "this.haveAgreements")
+ assert(this.haveUserCreationIMEvent, "this.haveUserCreationIMEvent")
+
+ if(!haveUserStateBootstrap) {
+ this._userStateBootstrap = aquarium.getUserStateBootstrap(this._userCreationIMEvent)
+ }
+ logger.debug("#### this._userStateBootStrap %s".format(this._userStateBootstrap.toString))
val now = TimeHelpers.nowMillis()
- val newActive = ActiveStateSnapshot(event.isStateActive, now)
+ this._userState = chargingService.replayMonthChargingUpTo(
+ BillingMonthInfo.fromMillis(now),
+ now,
+ this._userStateBootstrap,
+ aquarium.currentResourceTypesMap,
+ aquarium.userStateStore.insertUserState
+ )
+
+ // 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) {
+ this._userState.userAgreementHistoryModel = this._userAgreementHistoryModel
+ DEBUG("Computed working user state %s", AvroHelpers.jsonStringOfSpecificRecord(this._userState.msg))
+ }
+ }
- DEBUG("New active status = %s".format(newActive))
+ private[this] def initializeStateOfResourceEvents(event: InitializeUserActorState): Unit = {
+ if(!this.haveAgreements) {
+ DEBUG("Cannot initializeResourceEventsState() from %s. There are no agreements", event)
+ return
+ }
- this._userState = this._userState.copy(activeStateSnapshot = newActive)
+ if(!this.haveUserCreationIMEvent) {
+ DEBUG("Cannot initializeResourceEventsState() from %s. I never got a CREATE IMEvent", event)
+ return
+ }
+
+ // 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()
+ }
}
- def onProcessUserEvent(event: ProcessIMEvent): Unit = {
- val userEvent = event.imEvent
- if(userEvent.userID != this._userId) {
- ERROR("Received %s but my userId = %s".format(userEvent, this._userId))
- } else {
- if(userEvent.isCreateUser) {
- processCreateUser(userEvent)
- } else if(userEvent.isModifyUser) {
- processModifyUser(userEvent)
+ def onInitializeUserActorState(event: InitializeUserActorState): Unit = {
+ this._userID = event.userID
+ DEBUG("Got %s", event)
+
+ initializeStateOfIMEvents()
+ initializeStateOfResourceEvents(event)
+ }
+
+ /**
+ * Process [[gr.grnet.aquarium.message.avro.gen.IMEventMsg]]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 onIMEventMsg(imEvent: IMEventMsg): Unit = {
+ val hadUserCreationIMEvent = haveUserCreationIMEvent
+
+ if(!haveAgreements) {
+ // This IMEvent has arrived after any ResourceEvents
+ INFO("Arrived after any ResourceEvent: %s", AvroHelpers.jsonStringOfSpecificRecord(imEvent))
+ initializeStateOfIMEvents()
+ }
+ else {
+ if(this._latestIMEventOriginalID == imEvent.getOriginalID) {
+ // 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", AvroHelpers.jsonStringOfSpecificRecord(imEvent))
+ logSeparator()
+
+ //this._latestIMEventID = imEvent.id
+ return
}
+
+ 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")
+ loadUserStateAndUpdateAgreementHistory()
}
+
+ logSeparator()
}
- def onRequestUserBalance(event: RequestUserBalance): Unit = {
- val userId = event.userId
- val timestamp = event.timestamp
+ def onResourceEventMsg(rcEvent: ResourceEventMsg): Unit = {
+ 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", AvroHelpers.jsonStringOfSpecificRecord(rcEvent))
+ logSeparator()
+
+ 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._latestResourceEventOriginalID == rcEvent.getOriginalID) {
+ INFO("Ignoring first %s", AvroHelpers.jsonStringOfSpecificRecord(rcEvent))
+ logSeparator()
+
+ 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)
+ }
+
+ 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))
- if(TimeHelpers.nowMillis() - _userState.newestSnapshotTime > 60 * 1000) {
- // calcWalletEntries()
+ computeBatch()
}
- self reply UserResponseGetBalance(userId, _userState.creditsSnapshot.creditAmount)
+ 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 onUserRequestGetState(event: UserRequestGetState): Unit = {
- val userId = event.userId
- if(this._userId != userId) {
- ERROR("Received %s but my userId = %s".format(event, this._userId))
- // TODO: throw an exception here
- } else {
- // FIXME: implement
- ERROR("FIXME: Should have properly computed the user state")
- // ensureUserState()
- self reply UserResponseGetState(userId, this._userState)
+ 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 onActorProviderConfigured(event: ActorProviderConfigured): Unit = {
+ def onGetUserBalanceRequest(event: GetUserBalanceRequest): Unit = {
+ val userID = event.userID
+
+ (haveUserCreationIMEvent, haveUserState) match {
+ case (true, true) ⇒
+ // (User CREATEd, with balance state)
+ val realtimeMillis = TimeHelpers.nowMillis()
+ chargingService.calculateRealtimeUserState(
+ this._userState,
+ BillingMonthInfo.fromMillis(realtimeMillis),
+ realtimeMillis
+ )
+
+ sender ! GetUserBalanceResponse(Right(GetUserBalanceResponseData(this._userID, this._userState.totalCredits)))
+
+ case (true, false) ⇒
+ // (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) ⇒
+ // (Not CREATEd, with balance state)
+ // Clearly this is internal error
+ sender ! GetUserBalanceResponse(Left("Internal Server Error [AQU-BAL-0001]"), 500)
+
+ case (false, false) ⇒
+ // (Not CREATEd, no balance state)
+ // The user is completely unknown
+ sender ! GetUserBalanceResponse(Left("Unknown user %s [AQU-BAL-0004]".format(userID)), 404/*Not found*/)
+ }
}
- override def postStop {
- DEBUG("Stopping, saving state")
- saveUserState
+ def onGetUserStateRequest(event: GetUserStateRequest): Unit = {
+ haveUserState match {
+ case true ⇒
+ val realtimeMillis = TimeHelpers.nowMillis()
+ chargingService.calculateRealtimeUserState(
+ this._userState,
+ BillingMonthInfo.fromMillis(realtimeMillis),
+ realtimeMillis
+ )
+
+ sender ! GetUserStateResponse(Right(this._userState.msg))
+
+ case false ⇒
+ sender ! GetUserStateResponse(Left("No state for user %s [AQU-STA-0006]".format(event.userID)), 404)
+ }
}
- override def preRestart(reason: Throwable) {
- DEBUG("Actor failed, restarting")
+ 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)
+ haveUserCreationIMEvent match {
+ case true ⇒
+ DEBUG("haveUserCreationIMEvent: %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)
+ }
+ }
}
- override def postRestart(reason: Throwable) {
- DEBUG("Actor restarted succesfully")
+ private[this] def D_userID = {
+ this._userID
}
private[this] def DEBUG(fmt: String, args: Any*) =
- logger.debug("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ logger.debug("[%s] - %s".format(D_userID, fmt.format(args: _*)))
private[this] def INFO(fmt: String, args: Any*) =
- logger.info("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ logger.info("[%s] - %s".format(D_userID, fmt.format(args: _*)))
private[this] def WARN(fmt: String, args: Any*) =
- logger.warn("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ logger.warn("[%s] - %s".format(D_userID, fmt.format(args: _*)))
private[this] def ERROR(fmt: String, args: Any*) =
- logger.error("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ logger.error("[%s] - %s".format(D_userID, fmt.format(args: _*)))
+
+ private[this] def ERROR(t: Throwable, fmt: String, args: Any*) =
+ logger.error("[%s] - %s".format(D_userID, fmt.format(args: _*)), t)
}
projects / aquarium / blobdiff