code.grnet.gr Git - aquarium/blob - src/main/scala/gr/grnet/aquarium/store/memory/MemStore.scala

   1 /*
   2  * Copyright 2011-2012 GRNET S.A. All rights reserved.
   3  *
   4  * Redistribution and use in source and binary forms, with or
   5  * without modification, are permitted provided that the following
   6  * conditions are met:
   7  *
   8  *   1. Redistributions of source code must retain the above
   9  *      copyright notice, this list of conditions and the following
  10  *      disclaimer.
  11  *
  12  *   2. Redistributions in binary form must reproduce the above
  13  *      copyright notice, this list of conditions and the following
  14  *      disclaimer in the documentation and/or other materials
  15  *      provided with the distribution.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  18  * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  19  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  20  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  21  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  24  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  25  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  27  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  28  * POSSIBILITY OF SUCH DAMAGE.
  29  *
  30  * The views and conclusions contained in the software and
  31  * documentation are those of the authors and should not be
  32  * interpreted as representing official policies, either expressed
  33  * or implied, of GRNET S.A.
  34  */
  35
  36 package gr.grnet.aquarium.store.memory
  37
  38 import com.ckkloverdos.props.Props
  39 import com.ckkloverdos.maybe.{NoVal, Just}
  40 import gr.grnet.aquarium.store._
  41 import scala.collection.JavaConversions._
  42 import collection.mutable.ConcurrentMap
  43 import java.util.concurrent.ConcurrentHashMap
  44 import gr.grnet.aquarium.uid.ConcurrentVMLocalUIDGenerator
  45 import gr.grnet.aquarium.Configurable
  46 import gr.grnet.aquarium.event.model.PolicyEntry
  47 import gr.grnet.aquarium.event.model.im.{StdIMEvent, IMEventModel}
  48 import org.bson.types.ObjectId
  49 import gr.grnet.aquarium.event.model.resource.{StdResourceEvent, ResourceEventModel}
  50 import gr.grnet.aquarium.computation.UserState
  51 import gr.grnet.aquarium.util.Tags
  52
  53 /**
  54  * An implementation of various stores that persists data in memory.
  55  *
  56  * This is just for testing purposes.
  57  *
  58  * @author Christos KK Loverdos <loverdos@gmail.com>
  59  * @author Georgios Gousios <gousiosg@gmail.com>
  60  */
  61
  62 class MemStore extends UserStateStore
  63   with Configurable with PolicyStore
  64   with ResourceEventStore with IMEventStore
  65   with StoreProvider {
  66
  67   override type IMEvent = MemIMEvent
  68   override type ResourceEvent = MemResourceEvent
  69
  70   private[this] var _userStates     = List[UserState]()
  71   private[this] var _policyEntries  = List[PolicyEntry]()
  72   private[this] var _resourceEvents = List[ResourceEvent]()
  73
  74   private[this] val imEventById: ConcurrentMap[String, MemIMEvent] = new ConcurrentHashMap[String, MemIMEvent]()
  75
  76
  77   def propertyPrefix = None
  78
  79   def configure(props: Props) = {
  80   }
  81
  82   override def toString = {
  83     val map = Map(
  84       Tags.UserStateTag     -> _userStates.size,
  85       Tags.ResourceEventTag -> _resourceEvents.size,
  86       Tags.IMEventTag       -> imEventById.size,
  87       "PolicyEntry"         -> _policyEntries.size
  88     )
  89
  90     "MemStore(%s)" format map
  91   }
  92
  93   //+ StoreProvider
  94   def userStateStore = this
  95
  96   def resourceEventStore = this
  97
  98   def imEventStore = this
  99
 100   def policyStore = this
 101   //- StoreProvider
 102
 103
 104   //+ UserStateStore
 105   def insertUserState(userState: UserState): UserState = {
 106     _userStates = userState.copy(_id = new ObjectId().toString) :: _userStates
 107     userState
 108   }
 109
 110   def findUserStateByUserID(userID: String) = {
 111     _userStates.find(_.userID == userID)
 112   }
 113
 114   def findLatestUserStateForEndOfBillingMonth(userID: String,
 115                                               yearOfBillingMonth: Int,
 116                                               billingMonth: Int): Option[UserState] = {
 117     val goodOnes = _userStates.filter { userState ⇒
 118         val f1 = userState.userID == userID
 119         val f2 = userState.isFullBillingMonthState
 120         val bm = userState.theFullBillingMonth
 121         val f3 = (bm ne null) && {
 122           bm.year == yearOfBillingMonth && bm.month == billingMonth
 123         }
 124
 125         f1 && f2 && f3
 126     }
 127
 128     goodOnes.sortWith {
 129       case (us1, us2) ⇒
 130         us1.occurredMillis > us2.occurredMillis
 131     } match {
 132       case head :: _ ⇒
 133         Some(head)
 134       case _ ⇒
 135         None
 136     }
 137   }
 138
 139   def deleteUserState(userId: String) {
 140     _userStates.filterNot(_.userID == userId)
 141   }
 142   //- UserStateStore
 143
 144   //+ ResourceEventStore
 145   def createResourceEventFromOther(event: ResourceEventModel): ResourceEvent = {
 146     if(event.isInstanceOf[MemResourceEvent]) event.asInstanceOf[MemResourceEvent]
 147     else {
 148       import event._
 149       new StdResourceEvent(
 150         id,
 151         occurredMillis,
 152         receivedMillis,
 153         userID,
 154         clientID,
 155         resource,
 156         instanceID,
 157         value,
 158         eventVersion,
 159         details
 160       ): MemResourceEvent
 161     }
 162   }
 163
 164   override def clearResourceEvents() = {
 165     _resourceEvents = Nil
 166   }
 167
 168   def pingResourceEventStore(): Unit = {
 169     // We are always live and kicking...
 170   }
 171
 172   def insertResourceEvent(event: ResourceEventModel) = {
 173     val localEvent = createResourceEventFromOther(event)
 174     _resourceEvents ::= localEvent
 175     localEvent
 176   }
 177
 178   def findResourceEventById(id: String) = {
 179     _resourceEvents.find(ev ⇒ ev.id == id)
 180   }
 181
 182   def findResourceEventsByUserId(userId: String)
 183                                 (sortWith: Option[(ResourceEvent, ResourceEvent) => Boolean]): List[ResourceEvent] = {
 184     val byUserId = _resourceEvents.filter(_.userID == userId).toArray
 185     val sorted = sortWith match {
 186       case Some(sorter) ⇒
 187         byUserId.sortWith(sorter)
 188       case None ⇒
 189         byUserId
 190     }
 191
 192     sorted.toList
 193   }
 194
 195   def findResourceEventsByUserIdAfterTimestamp(userID: String, timestamp: Long): List[ResourceEvent] = {
 196     _resourceEvents.filter { ev ⇒
 197       ev.userID == userID &&
 198       (ev.occurredMillis > timestamp)
 199     }.toList
 200   }
 201
 202   def findResourceEventHistory(userId: String,
 203                                resName: String,
 204                                instid: Option[String],
 205                                upTo: Long): List[ResourceEvent] = {
 206     Nil
 207   }
 208
 209   def findResourceEventsForReceivedPeriod(userID: String,
 210                                           startTimeMillis: Long,
 211                                           stopTimeMillis: Long): List[ResourceEvent] = {
 212     _resourceEvents.filter { ev ⇒
 213       ev.userID == userID &&
 214       ev.isReceivedWithinMillis(startTimeMillis, stopTimeMillis)
 215     }.toList
 216   }
 217
 218   def countOutOfSyncResourceEventsForBillingPeriod(userID: String, startMillis: Long, stopMillis: Long): Long = {
 219     _resourceEvents.filter { case ev ⇒
 220       ev.userID == userID &&
 221       // out of sync events are those that were received in the billing month but occurred in previous (or next?)
 222       // months
 223       ev.isOutOfSyncForBillingPeriod(startMillis, stopMillis)
 224     }.size.toLong
 225   }
 226
 227   /**
 228    * Finds all relevant resource events for the billing period.
 229    * The relevant events are those:
 230    * a) whose `occurredMillis` is within the given billing period or
 231    * b) whose `receivedMillis` is within the given billing period.
 232    *
 233    * Order them by `occurredMillis`
 234    */
 235   override def findAllRelevantResourceEventsForBillingPeriod(userID: String,
 236                                                              startMillis: Long,
 237                                                              stopMillis: Long): List[ResourceEvent] = {
 238     _resourceEvents.filter { case ev ⇒
 239       ev.userID == userID &&
 240       ev.isOccurredOrReceivedWithinMillis(startMillis, stopMillis)
 241     }.toList sortWith { case (ev1, ev2) ⇒ ev1.occurredMillis <= ev2.occurredMillis }
 242   }
 243   //- ResourceEventStore
 244
 245   //+ IMEventStore
 246   def createIMEventFromJson(json: String) = {
 247     StdIMEvent.fromJsonString(json)
 248   }
 249
 250   def createIMEventFromOther(event: IMEventModel) = {
 251     StdIMEvent.fromOther(event)
 252   }
 253
 254   def pingIMEventStore(): Unit = {
 255   }
 256
 257
 258   def insertIMEvent(event: IMEventModel) = {
 259     val localEvent = createIMEventFromOther(event)
 260     imEventById += (event.id -> localEvent)
 261     localEvent
 262   }
 263
 264   def findIMEventById(id: String) = imEventById.get(id)
 265
 266
 267   /**
 268    * Find the `CREATE` even for the given user. Note that there must be only one such event.
 269    */
 270   def findCreateIMEventByUserID(userID: String): Option[IMEvent] = {
 271     imEventById.valuesIterator.filter { e ⇒
 272       e.userID == userID && e.isCreateUser
 273     }.toList.sortWith { case (e1, e2) ⇒
 274       e1.occurredMillis < e2.occurredMillis
 275     } headOption
 276   }
 277
 278   def findLatestIMEventByUserID(userID: String): Option[IMEvent] = {
 279     imEventById.valuesIterator.filter(_.userID == userID).toList.sortWith {
 280       case (us1, us2) ⇒
 281         us1.occurredMillis > us2.occurredMillis
 282     } headOption
 283   }
 284
 285   def findFirstIsActiveIMEventByUserID(userID: String): Option[IMEvent] = {
 286     imEventById.valuesIterator.filter { case ev ⇒
 287       ev.userID == userID && ev.isActive
 288     }.toList.sortWith { case (ev1, ev2) ⇒
 289       ev1.occurredMillis <= ev2.occurredMillis
 290     } match {
 291       case head :: _ ⇒
 292         Some(head)
 293
 294       case _ ⇒
 295         None
 296     }
 297   }
 298
 299   /**
 300    * Scans events for the given user, sorted by `occurredMillis` in ascending order and runs them through
 301    * the given function `f`.
 302    *
 303    * Any exception is propagated to the caller. The underlying DB resources are properly disposed in any case.
 304    */
 305   def replayIMEventsInOccurrenceOrder(userID: String)(f: (IMEvent) => Unit) = {
 306     imEventById.valuesIterator.filter(_.userID == userID).toSeq.sortWith {
 307       case (ev1, ev2) ⇒ ev1.occurredMillis <= ev2.occurredMillis
 308     } foreach(f)
 309   }
 310   //- IMEventStore
 311
 312   def loadPolicyEntriesAfter(after: Long) =
 313     _policyEntries.filter(p => p.validFrom > after)
 314             .sortWith((a,b) => a.validFrom < b.validFrom)
 315
 316   def storePolicyEntry(policy: PolicyEntry) = {_policyEntries = policy :: _policyEntries; Just(RecordID(policy.id))}
 317
 318   def updatePolicyEntry(policy: PolicyEntry) =
 319     _policyEntries = _policyEntries.foldLeft(List[PolicyEntry]()){
 320       (acc, p) =>
 321         if (p.id == policy.id)
 322           policy :: acc
 323         else
 324           p :: acc
 325   }
 326
 327   def findPolicyEntry(id: String) = {
 328     _policyEntries.find(p => p.id == id)
 329   }
 330 }
 331
 332 object MemStore {
 333   final def isLocalIMEvent(event: IMEventModel) = event match {
 334     case _: MemIMEvent ⇒ true
 335     case _ ⇒ false
 336   }
 337 }