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.uid.ConcurrentVMLocalUIDGenerator
  47 import gr.grnet.aquarium.{AquariumException, Configurable}
  48 import gr.grnet.aquarium.event.{WalletEntry, ResourceEvent, PolicyEntry}
  49 import gr.grnet.aquarium.converter.JsonTextFormat
  50 import gr.grnet.aquarium.util._
  51 import gr.grnet.aquarium.event.im.{StdIMEvent, IMEventModel}
  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 WalletEntryStore
  66   with StoreProvider {
  67
  68   override type IMEvent = MemIMEvent
  69
  70   private[this] val idGen = new ConcurrentVMLocalUIDGenerator(1000)
  71
  72   private[this] var _userStates     = List[UserState]()
  73   private[this] var _policyEntries  = List[PolicyEntry]()
  74   private[this] var _resourceEvents = List[ResourceEvent]()
  75
  76   private[this] val walletEntriesById: ConcurrentMap[String, WalletEntry] = new ConcurrentHashMap[String, WalletEntry]()
  77   private[this] val imEventById: ConcurrentMap[String, MemIMEvent] = new ConcurrentHashMap[String, MemIMEvent]()
  78
  79   def configure(props: Props) = {
  80   }
  81
  82   override def toString = {
  83     val map = Map(
  84       "UserState"     -> _userStates.size,
  85       "ResourceEvent" -> _resourceEvents.size,
  86       "IMEvent"     -> imEventById.size,
  87       "PolicyEntry"   -> _policyEntries.size,
  88       "WalletEntry"   -> walletEntriesById.size
  89     )
  90
  91     "MemStore(%s)" format map
  92   }
  93
  94   //+ StoreProvider
  95   def userStateStore = this
  96
  97   def resourceEventStore = this
  98
  99   def walletEntryStore = this
 100
 101   def imEventStore = this
 102
 103   def policyStore = this
 104   //- StoreProvider
 105
 106
 107   //+ UserStateStore
 108   def storeUserState(userState: UserState): Maybe[RecordID] = {
 109     _userStates = userState.copy(id = idGen.nextUID()) :: _userStates
 110     Just(RecordID(_userStates.head._id))
 111   }
 112
 113   def findUserStateByUserId(userId: String) = {
 114     _userStates.find(_.userId == userId) match {
 115       case Some(userState) ⇒
 116         Just(userState)
 117       case None ⇒
 118         NoVal
 119     }
 120   }
 121
 122   def findLatestUserStateForEndOfBillingMonth(userId: String,
 123                                               yearOfBillingMonth: Int,
 124                                               billingMonth: Int): Maybe[UserState] = {
 125     val goodOnes = _userStates.filter { userState ⇒
 126         val f1 = userState.userId == userId
 127         val f2 = userState.isFullBillingMonthState
 128         val bm = userState.theFullBillingMonth
 129         val f3 = (bm ne null) && {
 130           bm.year == yearOfBillingMonth && bm.month == billingMonth
 131         }
 132
 133         f1 && f2 && f3
 134     }
 135
 136     goodOnes.sortWith {
 137       case (us1, us2) ⇒
 138         us1.oldestSnapshotTime > us2.oldestSnapshotTime
 139     } match {
 140       case head :: _ ⇒
 141         Just(head)
 142       case _ ⇒
 143         NoVal
 144     }
 145   }
 146
 147   def deleteUserState(userId: String) {
 148     _userStates.filterNot(_.userId == userId)
 149   }
 150   //- UserStateStore
 151
 152   //- WalletEntryStore
 153   def storeWalletEntry(entry: WalletEntry): Maybe[RecordID] = {
 154     walletEntriesById.put(entry.id, entry)
 155     Just(RecordID(entry.id))
 156   }
 157
 158   def findWalletEntryById(id: String): Maybe[WalletEntry] = {
 159     Maybe(walletEntriesById.apply(id))
 160   }
 161
 162   def findUserWalletEntries(userId: String): List[WalletEntry] = {
 163     walletEntriesById.valuesIterator.filter(_.userId == userId).toList
 164   }
 165
 166   def findUserWalletEntriesFromTo(userId: String, from: Date, to: Date): List[WalletEntry] = {
 167     walletEntriesById.valuesIterator.filter { we ⇒
 168       val receivedDate = we.receivedDate
 169
 170       we.userId == userId &&
 171       ( (from before receivedDate) || (from == receivedDate) ) &&
 172       ( (to   after  receivedDate) || (to   == receivedDate) )
 173       true
 174     }.toList
 175   }
 176
 177   def findLatestUserWalletEntries(userId: String): Maybe[List[WalletEntry]] = NoVal
 178
 179   def findPreviousEntry(userId: String,
 180                         resource: String,
 181                         instanceId: String,
 182                         finalized: Option[Boolean]): List[WalletEntry] = Nil
 183
 184   def findWalletEntriesAfter(userId: String, from: Date): List[WalletEntry] = {
 185     walletEntriesById.valuesIterator.filter { we ⇒
 186       val occurredDate = we.occurredDate
 187
 188       we.userId == userId &&
 189             ( (from before occurredDate) || (from == occurredDate) )
 190     }.toList
 191   }
 192   //- WalletEntryStore
 193
 194   //+ ResourceEventStore
 195
 196   override def clearResourceEvents() = {
 197     _resourceEvents = Nil
 198   }
 199
 200   def storeResourceEvent(event: ResourceEvent) = {
 201     _resourceEvents ::= event
 202     RecordID(event.id)
 203   }
 204
 205   def findResourceEventById(id: String) = {
 206     _resourceEvents.find(ev ⇒ ev.id == id) match {
 207       case Some(ev) ⇒ Just(ev)
 208       case None     ⇒ NoVal
 209     }
 210   }
 211
 212   def findResourceEventsByUserId(userId: String)
 213                                 (sortWith: Option[(ResourceEvent, ResourceEvent) => Boolean]): List[ResourceEvent] = {
 214     val byUserId = _resourceEvents.filter(_.userID == userId).toArray
 215     val sorted = sortWith match {
 216       case Some(sorter) ⇒
 217         byUserId.sortWith(sorter)
 218       case None ⇒
 219         byUserId
 220     }
 221
 222     sorted.toList
 223   }
 224
 225   def findResourceEventsByUserIdAfterTimestamp(userId: String, timestamp: Long): List[ResourceEvent] = {
 226     _resourceEvents.filter { ev ⇒
 227       ev.userID == userId &&
 228       (ev.occurredMillis > timestamp)
 229     }.toList
 230   }
 231
 232   def findResourceEventHistory(userId: String,
 233                                resName: String,
 234                                instid: Option[String],
 235                                upTo: Long): List[ResourceEvent] = {
 236     Nil
 237   }
 238
 239   def findResourceEventsForReceivedPeriod(userId: String,
 240                                           startTimeMillis: Long,
 241                                           stopTimeMillis: Long): List[ResourceEvent] = {
 242     _resourceEvents.filter { ev ⇒
 243       ev.userID == userId &&
 244       ev.isReceivedWithinMillis(startTimeMillis, stopTimeMillis)
 245     }.toList
 246   }
 247
 248   def countOutOfSyncEventsForBillingPeriod(userId: String, startMillis: Long, stopMillis: Long): Maybe[Long] = Maybe {
 249     _resourceEvents.filter { case ev ⇒
 250       // out of sync events are those that were received in the billing month but occurred in previous (or next?)
 251       // months
 252       ev.isOutOfSyncForBillingPeriod(startMillis, stopMillis)
 253     }.size.toLong
 254   }
 255
 256   /**
 257    * Finds all relevant resource events for the billing period.
 258    * The relevant events are those:
 259    * a) whose `occurredMillis` is within the given billing period or
 260    * b) whose `receivedMillis` is within the given billing period.
 261    *
 262    * Order them by `occurredMillis`
 263    */
 264   override def findAllRelevantResourceEventsForBillingPeriod(userId: String,
 265                                                              startMillis: Long,
 266                                                              stopMillis: Long): List[ResourceEvent] = {
 267     _resourceEvents.filter { case ev ⇒
 268       ev.isOccurredOrReceivedWithinMillis(startMillis, stopMillis)
 269     }.toList sortWith { case (ev1, ev2) ⇒ ev1.occurredMillis <= ev2.occurredMillis }
 270   }
 271   //- ResourceEventStore
 272
 273   //+ IMEventStore
 274   def createIMEventFromJson(json: String) = {
 275     MemStore.createIMEventFromJson(json)
 276   }
 277
 278   def isLocalIMEvent(event: IMEventModel) = {
 279     MemStore.isLocalIMEvent(event)
 280   }
 281
 282   def createIMEventFromOther(event: IMEventModel) = {
 283     MemStore.createIMEventFromOther(event)
 284   }
 285
 286   def storeUnparsed(json: String) = throw new AquariumException("Not implemented")
 287
 288   def storeIMEvent(event: IMEventModel) = {imEventById += (event.id -> createIMEventFromOther(event)); RecordID(event.id)}
 289
 290   def findIMEventById(id: String) = Maybe{imEventById.getOrElse(id, null)}
 291
 292   def findIMEventsByUserId(userId: String) = imEventById.valuesIterator.filter{v => v.userID == userId}.toList
 293   //- IMEventStore
 294
 295   def loadPolicyEntriesAfter(after: Long) =
 296     _policyEntries.filter(p => p.validFrom > after)
 297             .sortWith((a,b) => a.validFrom < b.validFrom)
 298
 299   def storePolicyEntry(policy: PolicyEntry) = {_policyEntries = policy :: _policyEntries; Just(RecordID(policy.id))}
 300
 301   def updatePolicyEntry(policy: PolicyEntry) =
 302     _policyEntries = _policyEntries.foldLeft(List[PolicyEntry]()){
 303       (acc, p) =>
 304         if (p.id == policy.id)
 305           policy :: acc
 306         else
 307           p :: acc
 308   }
 309
 310   def findPolicyEntry(id: String) = _policyEntries.find(p => p.id == id) match {
 311     case Some(x) => Just(x)
 312     case None => NoVal
 313   }
 314 }
 315
 316 object MemStore {
 317   def createIMEventFromOther(event: IMEventModel) = {
 318     if(event.isInstanceOf[MemIMEvent]) event.asInstanceOf[MemIMEvent]
 319     else new StdIMEvent(
 320       event.id,
 321       event.occurredMillis,
 322       event.receivedMillis,
 323       event.userID,
 324       event.clientID,
 325       event.isActive,
 326       event.role,
 327       event.eventVersion,
 328       event.eventType,
 329       event.details
 330     ): MemIMEvent
 331   }
 332
 333   final def createIMEventFromJson(json: String) = {
 334     import gr.grnet.aquarium.converter.StdConverters.StdConverters
 335     StdConverters.convertEx[MemIMEvent](JsonTextFormat(json))
 336   }
 337
 338   final def createIMEventFromJsonBytes(jsonBytes: Array[Byte]) = {
 339     createIMEventFromJson(makeString(jsonBytes))
 340   }
 341
 342   final def isLocalIMEvent(event: IMEventModel) = event match {
 343     case _: MemIMEvent ⇒ true
 344     case _ ⇒ false
 345   }
 346 }