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, Maybe}
  40 import gr.grnet.aquarium.store._
  41 import scala.collection.JavaConversions._
  42 import java.util.Date
  43 import collection.mutable.ConcurrentMap
  44 import java.util.concurrent.ConcurrentHashMap
  45 import gr.grnet.aquarium.user.UserState
  46 import gr.grnet.aquarium.simulation.uid.ConcurrentVMLocalUIDGenerator
  47 import gr.grnet.aquarium.{AquariumException, Configurable}
  48 import gr.grnet.aquarium.events.{IMEvent, WalletEntry, ResourceEvent, PolicyEntry}
  49
  50 /**
  51  * An implementation of various stores that persists data in memory.
  52  *
  53  * This is just for testing purposes.
  54  *
  55  * @author Christos KK Loverdos <loverdos@gmail.com>
  56  * @author Georgios Gousios <gousiosg@gmail.com>
  57  */
  58
  59 class MemStore extends UserStateStore
  60   with Configurable with PolicyStore
  61   with ResourceEventStore with IMEventStore
  62   with WalletEntryStore
  63   with StoreProvider {
  64
  65   private[this] val idGen = new ConcurrentVMLocalUIDGenerator(1000)
  66
  67   private[this] var _userStates     = List[UserState]()
  68   private[this] var _policyEntries  = List[PolicyEntry]()
  69   private[this] var _resourceEvents = List[ResourceEvent]()
  70
  71   private[this] val walletEntriesById: ConcurrentMap[String, WalletEntry] = new ConcurrentHashMap[String, WalletEntry]()
  72   private[this] val imEventById: ConcurrentMap[String, IMEvent] = new ConcurrentHashMap[String, IMEvent]()
  73
  74   def configure(props: Props) = {
  75   }
  76
  77   override def toString = {
  78     val map = Map(
  79       "UserState"     -> _userStates.size,
  80       "ResourceEvent" -> _resourceEvents.size,
  81       "IMEvent"     -> imEventById.size,
  82       "PolicyEntry"   -> _policyEntries.size,
  83       "WalletEntry"   -> walletEntriesById.size
  84     )
  85
  86     "MemStore(%s)" format map
  87   }
  88
  89   //+ StoreProvider
  90   def userStateStore = this
  91
  92   def resourceEventStore = this
  93
  94   def walletEntryStore = this
  95
  96   def imEventStore = this
  97
  98   def policyStore = this
  99   //- StoreProvider
 100
 101
 102   //+ UserStateStore
 103   def storeUserState(userState: UserState): Maybe[RecordID] = {
 104     _userStates = userState.copy(id = idGen.nextUID()) :: _userStates
 105     Just(RecordID(_userStates.head._id))
 106   }
 107
 108   def findUserStateByUserId(userId: String) = {
 109     _userStates.find(_.userId == userId) match {
 110       case Some(userState) ⇒
 111         Just(userState)
 112       case None ⇒
 113         NoVal
 114     }
 115   }
 116
 117   def findLatestUserStateForEndOfBillingMonth(userId: String,
 118                                               yearOfBillingMonth: Int,
 119                                               billingMonth: Int): Maybe[UserState] = {
 120     val goodOnes = _userStates.filter { userState ⇒
 121         val f1 = userState.userId == userId
 122         val f2 = userState.isFullBillingMonthState
 123         val bm = userState.theFullBillingMonth
 124         val f3 = (bm ne null) && {
 125           bm.year == yearOfBillingMonth && bm.month == billingMonth
 126         }
 127
 128         f1 && f2 && f3
 129     }
 130
 131     goodOnes.sortWith {
 132       case (us1, us2) ⇒
 133         us1.oldestSnapshotTime > us2.oldestSnapshotTime
 134     } match {
 135       case head :: _ ⇒
 136         Just(head)
 137       case _ ⇒
 138         NoVal
 139     }
 140   }
 141
 142   def deleteUserState(userId: String) {
 143     _userStates.filterNot(_.userId == userId)
 144   }
 145   //- UserStateStore
 146
 147   //- WalletEntryStore
 148   def storeWalletEntry(entry: WalletEntry): Maybe[RecordID] = {
 149     walletEntriesById.put(entry.id, entry)
 150     Just(RecordID(entry.id))
 151   }
 152
 153   def findWalletEntryById(id: String): Maybe[WalletEntry] = {
 154     Maybe(walletEntriesById.apply(id))
 155   }
 156
 157   def findUserWalletEntries(userId: String): List[WalletEntry] = {
 158     walletEntriesById.valuesIterator.filter(_.userId == userId).toList
 159   }
 160
 161   def findUserWalletEntriesFromTo(userId: String, from: Date, to: Date): List[WalletEntry] = {
 162     walletEntriesById.valuesIterator.filter { we ⇒
 163       val receivedDate = we.receivedDate
 164
 165       we.userId == userId &&
 166       ( (from before receivedDate) || (from == receivedDate) ) &&
 167       ( (to   after  receivedDate) || (to   == receivedDate) )
 168       true
 169     }.toList
 170   }
 171
 172   def findLatestUserWalletEntries(userId: String): Maybe[List[WalletEntry]] = NoVal
 173
 174   def findPreviousEntry(userId: String,
 175                         resource: String,
 176                         instanceId: String,
 177                         finalized: Option[Boolean]): List[WalletEntry] = Nil
 178
 179   def findWalletEntriesAfter(userId: String, from: Date): List[WalletEntry] = {
 180     walletEntriesById.valuesIterator.filter { we ⇒
 181       val occurredDate = we.occurredDate
 182
 183       we.userId == userId &&
 184             ( (from before occurredDate) || (from == occurredDate) )
 185     }.toList
 186   }
 187   //- WalletEntryStore
 188
 189   //+ ResourceEventStore
 190
 191   override def clearResourceEvents() = {
 192     _resourceEvents = Nil
 193   }
 194
 195   def storeResourceEvent(event: ResourceEvent) = {
 196     _resourceEvents ::= event
 197     Just(RecordID(event.id))
 198   }
 199
 200   def findResourceEventById(id: String) = {
 201     _resourceEvents.find(ev ⇒ ev.id == id) match {
 202       case Some(ev) ⇒ Just(ev)
 203       case None     ⇒ NoVal
 204     }
 205   }
 206
 207   def findResourceEventsByUserId(userId: String)
 208                                 (sortWith: Option[(ResourceEvent, ResourceEvent) => Boolean]): List[ResourceEvent] = {
 209     val byUserId = _resourceEvents.filter(_.userID == userId).toArray
 210     val sorted = sortWith match {
 211       case Some(sorter) ⇒
 212         byUserId.sortWith(sorter)
 213       case None ⇒
 214         byUserId
 215     }
 216
 217     sorted.toList
 218   }
 219
 220   def findResourceEventsByUserIdAfterTimestamp(userId: String, timestamp: Long): List[ResourceEvent] = {
 221     _resourceEvents.filter { ev ⇒
 222       ev.userID == userId &&
 223       (ev.occurredMillis > timestamp)
 224     }.toList
 225   }
 226
 227   def findResourceEventHistory(userId: String,
 228                                resName: String,
 229                                instid: Option[String],
 230                                upTo: Long): List[ResourceEvent] = {
 231     Nil
 232   }
 233
 234   def findResourceEventsForReceivedPeriod(userId: String,
 235                                           startTimeMillis: Long,
 236                                           stopTimeMillis: Long): List[ResourceEvent] = {
 237     _resourceEvents.filter { ev ⇒
 238       ev.userID == userId &&
 239       ev.isReceivedWithinMillis(startTimeMillis, stopTimeMillis)
 240     }.toList
 241   }
 242
 243   def countOutOfSyncEventsForBillingPeriod(userId: String, startMillis: Long, stopMillis: Long): Maybe[Long] = Maybe {
 244     _resourceEvents.filter { case ev ⇒
 245       // out of sync events are those that were received in the billing month but occurred in previous (or next?)
 246       // months
 247       ev.isOutOfSyncForBillingPeriod(startMillis, stopMillis)
 248     }.size.toLong
 249   }
 250
 251   /**
 252    * Finds all relevant resource events for the billing period.
 253    * The relevant events are those:
 254    * a) whose `occurredMillis` is within the given billing period or
 255    * b) whose `receivedMillis` is within the given billing period.
 256    *
 257    * Order them by `occurredMillis`
 258    */
 259   override def findAllRelevantResourceEventsForBillingPeriod(userId: String,
 260                                                              startMillis: Long,
 261                                                              stopMillis: Long): List[ResourceEvent] = {
 262     _resourceEvents.filter { case ev ⇒
 263       ev.isOccurredOrReceivedWithinMillis(startMillis, stopMillis)
 264     }.toList sortWith { case (ev1, ev2) ⇒ ev1.occurredMillis <= ev2.occurredMillis }
 265   }
 266   //- ResourceEventStore
 267
 268   //+ IMEventStore
 269   def storeUnparsed(json: String) = throw new AquariumException("Not implemented")
 270
 271   def storeIMEvent(event: IMEvent) = {imEventById += (event.id -> event); Just(RecordID(event.id))}
 272
 273   def findIMEventById(id: String) = Maybe{imEventById.getOrElse(id, null)}
 274
 275   def findIMEventsByUserId(userId: String) = imEventById.valuesIterator.filter{v => v.userID == userId}.toList
 276   //- IMEventStore
 277
 278   def loadPolicyEntriesAfter(after: Long) =
 279     _policyEntries.filter(p => p.validFrom > after)
 280             .sortWith((a,b) => a.validFrom < b.validFrom)
 281
 282   def storePolicyEntry(policy: PolicyEntry) = {_policyEntries = policy :: _policyEntries; Just(RecordID(policy.id))}
 283
 284   def updatePolicyEntry(policy: PolicyEntry) =
 285     _policyEntries = _policyEntries.foldLeft(List[PolicyEntry]()){
 286       (acc, p) =>
 287         if (p.id == policy.id)
 288           policy :: acc
 289         else
 290           p :: acc
 291   }
 292
 293   def findPolicyEntry(id: String) = _policyEntries.find(p => p.id == id) match {
 294     case Some(x) => Just(x)
 295     case None => NoVal
 296   }
 297 }