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.util.shortClassNameOf
+import gr.grnet.aquarium.util.chainOfCauses
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 gr.grnet.aquarium.actor.message.service.router._
import message.config.{ActorProviderConfigured, AquariumPropertiesLoaded}
-import gr.grnet.aquarium.events.{WalletEntry, IMEvent}
+import gr.grnet.aquarium.event.im.IMEventModel
+import akka.config.Supervision.Temporary
+import gr.grnet.aquarium.{AquariumException, Configurator}
/**
* @author Christos KK Loverdos <loverdos@gmail.com>
*/
-class UserActor extends AquariumActor
-with AkkaAMQP
-with ReflectiveAquariumActor
-with Loggable {
- @volatile
- private[this] var _userId: String = _
- @volatile
+class UserActor extends ReflectiveRoleableActor {
+ private[this] var _userID: String = _
private[this] var _userState: UserState = _
- @volatile
- private[this] var _timestampTheshold: Long = _
- private[this] lazy val messenger = producer("aquarium") // FIXME: Read this from configuration
+ self.lifeCycle = Temporary
+
+ private[this] def _shutmedown(): Unit = {
+ if(_haveFullState) {
+ UserActorCache.invalidate(this._userID)
+ }
+
+ self.stop()
+ }
+
+ override protected def onThrowable(t: Throwable, message: AnyRef) = {
+ 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
- /**
- * 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.userEventStore
- 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))
- }
+ private[this] def _timestampTheshold =
+ _configurator.props.getLong(Configurator.Keys.user_state_timestamp_threshold).getOr(10000)
- /**
- * Create an empty state for a user
- */
- def createBlankState = {
- this._userState = DefaultUserStateComputations.createInitialUserState(this._userId, 0L, true, 0.0)
+
+ private[this] def _haveFullState = {
+ (this._userID ne null) && (this._userState ne null)
}
- /**
- * Replay user events on the provided user state
- */
- def replayIMEvents(initState: UserState, events: List[IMEvent],
- from: Long, to: Long): UserState = {
- initState
+ private[this] def _havePartialState = {
+ (this._userID ne null) && (this._userState eq 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));
- }
+ def onActorProviderConfigured(event: ActorProviderConfigured): Unit = {
}
- 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)
+ private[this] def _getAgreementNameForNewUser(imEvent: IMEventModel): String = {
+ // FIXME: Implement based on the role
+ "default"
}
- def onUserActorInitWithUserId(event: UserActorInitWithUserId): Unit = {
- this._userId = event.userId
- DEBUG("Actor starting, loading state")
- }
+ private[this] def processCreateUser(imEvent: IMEventModel): Unit = {
+ this._userID = imEvent.userID
- 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))
- } else {
- //ensureUserState()
- // calcWalletEntries()
- //processResourceEvent(resourceEvent, true)
+ 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))
+
+ return
}
+
+ val initialAgreementName = _getAgreementNameForNewUser(imEvent)
+ val newUserState = DefaultUserStateComputations.createInitialUserState(
+ this._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)
}
- private[this] def processCreateUser(event: IMEvent): 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 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)
}
- private[this] def processModifyUser(event: IMEvent): Unit = {
- val now = TimeHelpers.nowMillis
- val newActive = ActiveStateSnapshot(event.isStateActive, now)
+ def onProcessSetUserID(event: ProcessSetUserID): Unit = {
+ this._userID = event.userID
+ }
- DEBUG("New active status = %s".format(newActive))
+ def onProcessIMEvent(event: ProcessIMEvent): Unit = {
+ val now = TimeHelpers.nowMillis()
- this._userState = this._userState.copy(activeStateSnapshot = newActive)
- }
+ 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 AquariumException("Got userID = %s but already have userID = %s".format(imEvent.userID, this._userID))
+ }
- def onProcessUserEvent(event: ProcessUserEvent): 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)
+ // 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
}
}
- }
- def onRequestUserBalance(event: RequestUserBalance): Unit = {
- val userId = event.userId
- val timestamp = event.timestamp
+ if(imEvent.isModifyUser && this._userState.isInitial) {
+ INFO("Got a '%s' but have not received '%s' yet", imEvent.eventType, IMEventModel.EventTypeNames.create)
+ return
+ }
- if(TimeHelpers.nowMillis - _userState.newestSnapshotTime > 60 * 1000) {
- // calcWalletEntries()
+ if(imEvent.isCreateUser && !this._userState.isInitial) {
+ INFO("Got a '%s' but my state is not initial", imEvent.eventType)
+ return
}
- self reply UserResponseGetBalance(userId, _userState.creditsSnapshot.creditAmount)
- }
- 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
+ this._userState = this._userState.modifyFromIMEvent(imEvent, now)
+
+ if(imEvent.isCreateUser) {
+ processCreateUser(imEvent)
+ } else if(imEvent.isModifyUser) {
+ processModifyUser(imEvent)
} else {
- // FIXME: implement
- ERROR("FIXME: Should have properly computed the user state")
- // ensureUserState()
- self reply UserResponseGetState(userId, this._userState)
+ throw new AquariumException("Cannot interpret %s".format(imEvent))
}
}
- def onActorProviderConfigured(event: ActorProviderConfigured): Unit = {
+ def onRequestUserBalance(event: RequestUserBalance): Unit = {
+ val userId = event.userID
+ // FIXME: Implement threshold
+ self reply UserResponseGetBalance(userId, _userState.creditsSnapshot.creditAmount)
}
- override def postStop {
- DEBUG("Stopping, saving state")
- saveUserState
+ def onUserRequestGetState(event: UserRequestGetState): Unit = {
+ val userId = event.userID
+ // FIXME: implement
+ self reply UserResponseGetState(userId, this._userState)
}
- override def preRestart(reason: Throwable) {
- DEBUG("Actor failed, restarting")
+ def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
}
- override def postRestart(reason: Throwable) {
- DEBUG("Actor restarted succesfully")
+
+ 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)
}
private[this] def DEBUG(fmt: String, args: Any*) =
- logger.debug("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ 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(_userId, fmt.format(args: _*)))
+ 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(_userId, fmt.format(args: _*)))
+ 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(_userId, fmt.format(args: _*)))
+ 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