package gr.grnet.aquarium.store.mongodb
-import gr.grnet.aquarium.util.Loggable
import com.mongodb.util.JSON
import gr.grnet.aquarium.user.UserState
import gr.grnet.aquarium.user.UserState.{JsonNames => UserStateJsonNames}
-import gr.grnet.aquarium.util.displayableObjectInfo
import gr.grnet.aquarium.util.json.JsonSupport
import collection.mutable.ListBuffer
+import gr.grnet.aquarium.events.im.IMEventModel.{Names => IMEventNames}
import gr.grnet.aquarium.store._
-import gr.grnet.aquarium.logic.events.ResourceEvent.{JsonNames => ResourceJsonNames}
-import gr.grnet.aquarium.logic.events.UserEvent.{JsonNames => UserEventJsonNames}
-import gr.grnet.aquarium.logic.events.WalletEntry.{JsonNames => WalletJsonNames}
-import gr.grnet.aquarium.logic.events.PolicyEntry.{JsonNames => PolicyJsonNames}
+import gr.grnet.aquarium.events.ResourceEvent.{JsonNames => ResourceJsonNames}
+import gr.grnet.aquarium.events.WalletEntry.{JsonNames => WalletJsonNames}
+import gr.grnet.aquarium.events.PolicyEntry.{JsonNames => PolicyJsonNames}
import java.util.Date
import gr.grnet.aquarium.logic.accounting.Policy
-import gr.grnet.aquarium.logic.events._
import com.mongodb._
-import com.ckkloverdos.maybe.{NoVal, Maybe}
-import gr.grnet.aquarium.logic.accounting.dsl.{DSLResource, Timeslot, DSLPolicy, DSLComplexResource}
import org.bson.types.ObjectId
+import gr.grnet.aquarium.events._
+import com.ckkloverdos.maybe.{NoVal, Maybe}
+import im.IMEventModel
+import gr.grnet.aquarium.util._
+import gr.grnet.aquarium.converter.StdConverters
/**
* Mongodb implementation of the various aquarium stores.
extends ResourceEventStore
with UserStateStore
with WalletEntryStore
- with UserEventStore
+ with IMEventStore
with PolicyStore
with Loggable {
- private[store] lazy val resourceEvents = getCollection(MongoDBStore.RESOURCE_EVENTS_COLLECTION)
- private[store] lazy val userStates = getCollection(MongoDBStore.USER_STATES_COLLECTION)
- private[store] lazy val userEvents = getCollection(MongoDBStore.USER_EVENTS_COLLECTION)
- private[store] lazy val unparsedUserEvents = getCollection(MongoDBStore.UNPARSED_USER_EVENTS_COLLECTION)
- private[store] lazy val walletEntries = getCollection(MongoDBStore.WALLET_ENTRIES_COLLECTION)
-// private[store] lazy val policies = getCollection(MongoDBStore.POLICIES_COLLECTION)
- private[store] lazy val policyEntries = getCollection(MongoDBStore.POLICY_ENTRIES_COLLECTION)
+ override type IMEvent = MongoDBIMEvent
+
+ private[store] lazy val resourceEvents = getCollection(MongoDBStore.RESOURCE_EVENTS_COLLECTION)
+ private[store] lazy val userStates = getCollection(MongoDBStore.USER_STATES_COLLECTION)
+ private[store] lazy val imEvents = getCollection(MongoDBStore.IM_EVENTS_COLLECTION)
+ private[store] lazy val unparsedIMEvents = getCollection(MongoDBStore.UNPARSED_IM_EVENTS_COLLECTION)
+ private[store] lazy val walletEntries = getCollection(MongoDBStore.WALLET_ENTRIES_COLLECTION)
+ private[store] lazy val policyEntries = getCollection(MongoDBStore.POLICY_ENTRIES_COLLECTION)
private[this] def getCollection(name: String): DBCollection = {
val db = mongo.getDB(database)
db.getCollection(name)
}
- private[this] def _sortByTimestampAsc[A <: AquariumEvent](one: A, two: A): Boolean = {
+ private[this] def _sortByTimestampAsc[A <: AquariumEventModel](one: A, two: A): Boolean = {
if (one.occurredMillis > two.occurredMillis) false
else if (one.occurredMillis < two.occurredMillis) true
else true
}
- private[this] def _sortByTimestampDesc[A <: AquariumEvent](one: A, two: A): Boolean = {
+ private[this] def _sortByTimestampDesc[A <: AquariumEventSkeleton](one: A, two: A): Boolean = {
if (one.occurredMillis < two.occurredMillis) false
else if (one.occurredMillis > two.occurredMillis) true
else true
}
//+ResourceEventStore
- def storeResourceEvent(event: ResourceEvent): Maybe[RecordID] =
- MongoDBStore.storeAquariumEvent(event, resourceEvents)
+ def storeResourceEvent(event: ResourceEvent) = {
+ MongoDBStore.storeAny[ResourceEvent](
+ event,
+ resourceEvents,
+ ResourceJsonNames.id,
+ (e) => e.id,
+ MongoDBStore.jsonSupportToDBObject)
+ }
def findResourceEventById(id: String): Maybe[ResourceEvent] =
MongoDBStore.findById(id, resourceEvents, MongoDBStore.dbObjectToResourceEvent)
//- UserStateStore
//+WalletEntryStore
- def storeWalletEntry(entry: WalletEntry): Maybe[RecordID] =
- MongoDBStore.storeAquariumEvent(entry, walletEntries)
+ def storeWalletEntry(entry: WalletEntry): Maybe[RecordID] = {
+ Maybe {
+ MongoDBStore.storeAny[WalletEntry](
+ entry,
+ walletEntries,
+ ResourceJsonNames.id,
+ (e) => e.id,
+ MongoDBStore.jsonSupportToDBObject)
+ }
+ }
def findWalletEntryById(id: String): Maybe[WalletEntry] =
MongoDBStore.findById[WalletEntry](id, walletEntries, MongoDBStore.dbObjectToWalletEntry)
}
//-WalletEntryStore
- //+UserEventStore
- def storeUnparsed(json: String): Maybe[RecordID] = {
- MongoDBStore.storeJustJson(json, unparsedUserEvents)
+ //+IMEventStore
+ def isLocalIMEvent(event: IMEventModel) = {
+ MongoDBStore.isLocalIMEvent(event)
}
- def storeUserEvent(event: UserEvent): Maybe[RecordID] =
- MongoDBStore.storeAny[UserEvent](event, userEvents, UserEventJsonNames.userID,
- _.userID, MongoDBStore.jsonSupportToDBObject)
+ def createIMEventFromJson(json: String) = {
+ MongoDBStore.createIMEventFromJson(json)
+ }
+ def createIMEventFromOther(event: IMEventModel) = {
+ MongoDBStore.createIMEventFromOther(event)
+ }
+
+ def storeUnparsed(json: String): Maybe[RecordID] = {
+ MongoDBStore.storeJustJson(json, unparsedIMEvents)
+ }
- def findUserEventById(id: String): Maybe[UserEvent] =
- MongoDBStore.findById[UserEvent](id, userEvents, MongoDBStore.dbObjectToUserEvent)
+ def storeIMEvent(_event: IMEventModel): RecordID = {
+ val event = createIMEventFromOther(_event)
+ MongoDBStore.storeAny[IMEvent](
+ event,
+ imEvents,
+ IMEventNames.userID,
+ _.userID,
+ MongoDBStore.jsonSupportToDBObject
+ )
+ }
+
+ def findIMEventById(id: String): Maybe[IMEvent] =
+ MongoDBStore.findById[IMEvent](id, imEvents, MongoDBStore.dbObjectToIMEvent)
- def findUserEventsByUserId(userId: String): List[UserEvent] = {
- val query = new BasicDBObject(UserEventJsonNames.userID, userId)
- MongoDBStore.runQuery(query, userEvents)(MongoDBStore.dbObjectToUserEvent)(Some(_sortByTimestampAsc))
+ def findIMEventsByUserId(userId: String): List[IMEvent] = {
+ val query = new BasicDBObject(IMEventNames.userID, userId)
+ MongoDBStore.runQuery(query, imEvents)(MongoDBStore.dbObjectToIMEvent)(Some(_sortByTimestampAsc))
}
- //-UserEventStore
+ //-IMEventStore
//+PolicyStore
def loadPolicyEntriesAfter(after: Long): List[PolicyEntry] = {
}
/**
- * Collection holding the [[gr.grnet.aquarium.logic.events.ResourceEvent]]s.
+ * Collection holding the [[gr.grnet.aquarium.events.ResourceEvent]]s.
*
* Resource events are coming from all systems handling billable resources.
*/
/**
* Collection holding the snapshots of [[gr.grnet.aquarium.user.UserState]].
*
- * [[gr.grnet.aquarium.user.UserState]] is held internally within [[gr.grnet.aquarium.user.actor.UserActor]]s.
+ * [[gr.grnet.aquarium.user.UserState]] is held internally within [[gr.grnet.aquarium.actor.service.user .UserActor]]s.
*/
final val USER_STATES_COLLECTION = "userstates"
/**
- * Collection holding [[gr.grnet.aquarium.logic.events.UserEvent]]s.
+ * Collection holding [[gr.grnet.aquarium.events.im.IMEventModel]]s.
*
* User events are coming from the IM module (external).
*/
- final val USER_EVENTS_COLLECTION = "userevents"
+ final val IM_EVENTS_COLLECTION = "imevents"
/**
- * Collection holding [[gr.grnet.aquarium.logic.events.UserEvent]]s that could not be parsed to normal objects.
+ * Collection holding [[gr.grnet.aquarium.events.im.IMEventModel]]s that could not be parsed to normal objects.
*
* We of course assume at least a valid JSON representation.
*
* User events are coming from the IM module (external).
*/
- final val UNPARSED_USER_EVENTS_COLLECTION = "unparsed_userevents"
+ final val UNPARSED_IM_EVENTS_COLLECTION = "unparsed_imevents"
/**
- * Collection holding [[gr.grnet.aquarium.logic.events.WalletEntry]].
+ * Collection holding [[gr.grnet.aquarium.events.WalletEntry]].
*
* Wallet entries are generated internally in Aquarium.
*/
// final val POLICIES_COLLECTION = "policies"
/**
- * Collection holding [[gr.grnet.aquarium.logic.events.PolicyEntry]].
+ * Collection holding [[gr.grnet.aquarium.events.PolicyEntry]].
*/
final val POLICY_ENTRIES_COLLECTION = "policyEntries"
WalletEntry.fromJson(JSON.serialize(dbObj))
}
- def dbObjectToUserEvent(dbObj: DBObject): UserEvent = {
- UserEvent.fromJson(JSON.serialize(dbObj))
+ def dbObjectToIMEvent(dbObj: DBObject): MongoDBIMEvent = {
+ MongoDBIMEvent.fromJson(JSON.serialize(dbObj))
}
def dbObjectToPolicyEntry(dbObj: DBObject): PolicyEntry = {
PolicyEntry.fromJson(JSON.serialize(dbObj))
}
- def findById[A >: Null <: AquariumEvent](id: String, collection: DBCollection, deserializer: (DBObject) => A) : Maybe[A] = Maybe {
+ def findById[A >: Null <: AnyRef](id: String, collection: DBCollection, deserializer: (DBObject) => A) : Maybe[A] =
+ Maybe {
val query = new BasicDBObject(ResourceJsonNames.id, id)
val cursor = collection find query
}
}
- def runQuery[A <: AquariumEvent](query: DBObject, collection: DBCollection, orderBy: DBObject = null)
+ def runQuery[A <: AquariumEventModel](query: DBObject, collection: DBCollection, orderBy: DBObject = null)
(deserializer: (DBObject) => A)
(sortWith: Option[(A, A) => Boolean]): List[A] = {
val cursor0 = collection find query
}
}
- def storeAquariumEvent[A <: AquariumEvent](event: A, collection: DBCollection) : Maybe[RecordID] = {
- storeAny[A](event, collection, ResourceJsonNames.id, (e) => e.id, MongoDBStore.jsonSupportToDBObject)
- }
-
def storeUserState(userState: UserState, collection: DBCollection): Maybe[RecordID] = {
- storeAny[UserState](userState, collection, ResourceJsonNames.userId, _.userId, MongoDBStore.jsonSupportToDBObject)
+ Maybe(storeAny[UserState](userState, collection, ResourceJsonNames.userId, _.userId, MongoDBStore.jsonSupportToDBObject))
}
def storePolicyEntry(policyEntry: PolicyEntry, collection: DBCollection): Maybe[RecordID] = {
- storeAny[PolicyEntry](policyEntry, collection, PolicyJsonNames.id, _.id, MongoDBStore.jsonSupportToDBObject)
+ Maybe(storeAny[PolicyEntry](policyEntry, collection, PolicyJsonNames.id, _.id, MongoDBStore.jsonSupportToDBObject))
}
def storeJustJson(json: String, collection: DBCollection): Maybe[RecordID] = {
collection: DBCollection,
idName: String,
idValueProvider: (A) => String,
- serializer: (A) => DBObject) : Maybe[RecordID] = {
- import com.ckkloverdos.maybe.effect
+ serializer: (A) => DBObject) : RecordID = {
- Maybe {
- val dbObj = serializer apply any
- val writeResult = collection insert dbObj
- writeResult.getLastError().throwOnError()
+ val dbObject = serializer apply any
+ val _id = new ObjectId()
+ dbObject.put("_id", _id)
+ val writeResult = collection.insert(dbObject, WriteConcern.JOURNAL_SAFE)
+ writeResult.getLastError().throwOnError()
- // Get back to retrieve unique id
- val cursor = collection.find(new BasicDBObject(idName, idValueProvider(any)))
- cursor
- } flatMap { cursor ⇒
- effect {
- if(cursor.hasNext)
- RecordID(cursor.next().get(JsonNames._id).toString)
- else
- throw new StoreException("Could not store %s to %s".format(any, collection))
- } {} { cursor.close() }
- }
+ RecordID(dbObject.get("_id").toString)
}
- def jsonSupportToDBObject(any: JsonSupport): DBObject = {
- jsonStringToDBObject(any.toJson)
+ def jsonSupportToDBObject(jsonSupport: JsonSupport): DBObject = {
+ StdConverters.StdConverters.convertEx[DBObject](jsonSupport)
}
- def jsonStringToDBObject(json: String): DBObject = {
- JSON.parse(json) match {
- case dbObject: DBObject ⇒
- dbObject
- case _ ⇒
- throw new StoreException("Could not transform %s -> %s".format(json.getClass, classOf[DBObject].getName))
- }
+ def jsonStringToDBObject(jsonString: String): DBObject = {
+ StdConverters.StdConverters.convertEx[DBObject](jsonString)
+ }
+
+ final def isLocalIMEvent(event: IMEventModel) = event match {
+ case _: MongoDBIMEvent ⇒ true
+ case _ ⇒ false
+ }
+
+ final def createIMEventFromJson(json: String) = {
+ MongoDBIMEvent.fromJson(json)
+ }
+
+ final def createIMEventFromOther(event: IMEventModel) = {
+ MongoDBIMEvent.fromOther(event)
+ }
+
+ final def createIMEventFromJsonBytes(jsonBytes: Array[Byte]) = {
+ MongoDBIMEvent.fromJsonBytes(jsonBytes)
}
}
projects / aquarium / blobdiff