Event refactoring

[aquarium] / src / main / scala / gr / grnet / aquarium / store / memory / MemStore.scala

diff --git a/src/main/scala/gr/grnet/aquarium/store/memory/MemStore.scala b/src/main/scala/gr/grnet/aquarium/store/memory/MemStore.scala
index a19a5a6..871594e 100644 (file)
--- a/src/main/scala/gr/grnet/aquarium/store/memory/MemStore.scala
+++ b/src/main/scala/gr/grnet/aquarium/store/memory/MemStore.scala
@@ -42,13 +42,13 @@ import scala.collection.JavaConversions._
 import java.util.Date
 import collection.mutable.ConcurrentMap
 import java.util.concurrent.ConcurrentHashMap
-import gr.grnet.aquarium.user.UserState
 import gr.grnet.aquarium.uid.ConcurrentVMLocalUIDGenerator
-import gr.grnet.aquarium.{AquariumException, Configurable}
-import gr.grnet.aquarium.event.{WalletEntry, ResourceEvent, PolicyEntry}
-import gr.grnet.aquarium.converter.JsonTextFormat
-import gr.grnet.aquarium.util._
-import gr.grnet.aquarium.event.im.{StdIMEvent, IMEventModel}
+import gr.grnet.aquarium.Configurable
+import gr.grnet.aquarium.event.model.{WalletEntry, PolicyEntry}
+import gr.grnet.aquarium.event.model.im.{StdIMEvent, IMEventModel}
+import org.bson.types.ObjectId
+import gr.grnet.aquarium.event.model.resource.{StdResourceEvent, ResourceEventModel}
+import gr.grnet.aquarium.computation.UserState
 
 /**
  * An implementation of various stores that persists data in memory.
@@ -66,6 +66,7 @@ class MemStore extends UserStateStore
   with StoreProvider {
 
   override type IMEvent = MemIMEvent
+  override type ResourceEvent = MemResourceEvent
 
   private[this] val idGen = new ConcurrentVMLocalUIDGenerator(1000)
   
@@ -105,25 +106,34 @@ class MemStore extends UserStateStore
 
 
   //+ UserStateStore
-  def storeUserState(userState: UserState): Maybe[RecordID] = {
-    _userStates = userState.copy(id = idGen.nextUID()) :: _userStates
-    Just(RecordID(_userStates.head._id))
+  def insertUserState(userState: UserState): UserState = {
+    _userStates = userState.copy(_id = new ObjectId()) :: _userStates
+    userState
   }
 
-  def findUserStateByUserId(userId: String) = {
-    _userStates.find(_.userId == userId) match {
-      case Some(userState) ⇒
-        Just(userState)
-      case None ⇒
-        NoVal
+  def findUserStateByUserID(userID: String) = {
+    _userStates.find(_.userID == userID)
+  }
+
+  def findLatestUserStateByUserID(userID: String) = {
+    val goodOnes = _userStates.filter(_.userID == userID)
+
+    goodOnes.sortWith {
+      case (us1, us2) ⇒
+        us1.occurredMillis > us2.occurredMillis
+    } match {
+      case head :: _ ⇒
+        Some(head)
+      case _ ⇒
+        None
     }
   }
 
-  def findLatestUserStateForEndOfBillingMonth(userId: String,
+  def findLatestUserStateForEndOfBillingMonth(userID: String,
                                               yearOfBillingMonth: Int,
-                                              billingMonth: Int): Maybe[UserState] = {
+                                              billingMonth: Int): Option[UserState] = {
     val goodOnes = _userStates.filter { userState ⇒
-        val f1 = userState.userId == userId
+        val f1 = userState.userID == userID
         val f2 = userState.isFullBillingMonthState
         val bm = userState.theFullBillingMonth
         val f3 = (bm ne null) && {
@@ -135,17 +145,17 @@ class MemStore extends UserStateStore
     
     goodOnes.sortWith {
       case (us1, us2) ⇒
-        us1.oldestSnapshotTime > us2.oldestSnapshotTime
+        us1.occurredMillis > us2.occurredMillis
     } match {
       case head :: _ ⇒
-        Just(head)
+        Some(head)
       case _ ⇒
-        NoVal
+        None
     }
   }
 
   def deleteUserState(userId: String) {
-    _userStates.filterNot(_.userId == userId)
+    _userStates.filterNot(_.userID == userId)
   }
   //- UserStateStore
 
@@ -192,21 +202,37 @@ class MemStore extends UserStateStore
   //- WalletEntryStore
 
   //+ ResourceEventStore
+  def createResourceEventFromOther(event: ResourceEventModel): ResourceEvent = {
+    if(event.isInstanceOf[MemResourceEvent]) event.asInstanceOf[MemResourceEvent]
+    else {
+      import event._
+      new StdResourceEvent(
+        id,
+        occurredMillis,
+        receivedMillis,
+        userID,
+        clientID,
+        resource,
+        instanceID,
+        value,
+        eventVersion,
+        details
+      ): MemResourceEvent
+    }
+  }
 
   override def clearResourceEvents() = {
     _resourceEvents = Nil
   }
 
-  def storeResourceEvent(event: ResourceEvent) = {
-    _resourceEvents ::= event
-    RecordID(event.id)
+  def insertResourceEvent(event: ResourceEventModel) = {
+    val localEvent = createResourceEventFromOther(event)
+    _resourceEvents ::= localEvent
+    localEvent
   }
 
   def findResourceEventById(id: String) = {
-    _resourceEvents.find(ev ⇒ ev.id == id) match {
-      case Some(ev) ⇒ Just(ev)
-      case None     ⇒ NoVal
-    }
+    _resourceEvents.find(ev ⇒ ev.id == id)
   }
 
   def findResourceEventsByUserId(userId: String)
@@ -245,7 +271,7 @@ class MemStore extends UserStateStore
     }.toList
   }
 
-  def countOutOfSyncEventsForBillingPeriod(userId: String, startMillis: Long, stopMillis: Long): Maybe[Long] = Maybe {
+  def countOutOfSyncEventsForBillingPeriod(userId: String, startMillis: Long, stopMillis: Long): Long = {
     _resourceEvents.filter { case ev ⇒
       // out of sync events are those that were received in the billing month but occurred in previous (or next?)
       // months
@@ -272,28 +298,20 @@ class MemStore extends UserStateStore
 
   //+ IMEventStore
   def createIMEventFromJson(json: String) = {
-    MemStore.createIMEventFromJson(json)
-  }
-
-  def isLocalIMEvent(event: IMEventModel) = {
-    MemStore.isLocalIMEvent(event)
+    StdIMEvent.fromJsonString(json)
   }
 
   def createIMEventFromOther(event: IMEventModel) = {
-    MemStore.createIMEventFromOther(event)
+    StdIMEvent.fromOther(event)
   }
 
-  def storeUnparsed(json: String) = throw new AquariumException("Not implemented")
-
   def insertIMEvent(event: IMEventModel) = {
     val localEvent = createIMEventFromOther(event)
     imEventById += (event.id -> localEvent)
     localEvent
   }
 
-  def findIMEventById(id: String) = Maybe{imEventById.getOrElse(id, null)}
-
-  def findIMEventsByUserId(userId: String) = imEventById.valuesIterator.filter{v => v.userID == userId}.toList
+  def findIMEventById(id: String) = imEventById.get(id)
   //- IMEventStore
 
   def loadPolicyEntriesAfter(after: Long) =
@@ -318,31 +336,6 @@ class MemStore extends UserStateStore
 }
 
 object MemStore {
-  def createIMEventFromOther(event: IMEventModel) = {
-    if(event.isInstanceOf[MemIMEvent]) event.asInstanceOf[MemIMEvent]
-    else new StdIMEvent(
-      event.id,
-      event.occurredMillis,
-      event.receivedMillis,
-      event.userID,
-      event.clientID,
-      event.isActive,
-      event.role,
-      event.eventVersion,
-      event.eventType,
-      event.details
-    ): MemIMEvent
-  }
-
-  final def createIMEventFromJson(json: String) = {
-    import gr.grnet.aquarium.converter.StdConverters.AllConverters
-    AllConverters.convertEx[MemIMEvent](JsonTextFormat(json))
-  }
-
-  final def createIMEventFromJsonBytes(jsonBytes: Array[Byte]) = {
-    createIMEventFromJson(makeString(jsonBytes))
-  }
-
   final def isLocalIMEvent(event: IMEventModel) = event match {
     case _: MemIMEvent ⇒ true
     case _ ⇒ false