package gr.grnet.aquarium
-import java.io.File
-
-import com.ckkloverdos.convert.Converters.{DefaultConverters => TheDefaultConverters}
+import com.ckkloverdos.convert.Converters
+import com.ckkloverdos.env.Env
+import com.ckkloverdos.key.{IntKey, StringKey, LongKey, TypedKeySkeleton, TypedKey, BooleanKey}
import com.ckkloverdos.maybe._
import com.ckkloverdos.props.Props
import com.ckkloverdos.sys.SysProp
-
-import gr.grnet.aquarium.util.{Lifecycle, Loggable, shortNameOfClass, shortClassNameOf}
-import gr.grnet.aquarium.store._
-import gr.grnet.aquarium.service._
-import gr.grnet.aquarium.converter.StdConverters
+import connector.rabbitmq.RabbitMQProducer
+import gr.grnet.aquarium.charging.{ChargingService, ChargingBehavior}
+import gr.grnet.aquarium.message.avro.gen.{UserAgreementMsg, FullPriceTableMsg, IMEventMsg, ResourceTypeMsg, PolicyMsg}
+import gr.grnet.aquarium.message.avro.{MessageHelpers, MessageFactory, ModelFactory, AvroHelpers}
+import gr.grnet.aquarium.policy.{AdHocFullPriceTableRef, FullPriceTableModel, PolicyModel, CachingPolicyStore, PolicyDefinedFullPriceTableRef, UserAgreementModel, ResourceType}
+import gr.grnet.aquarium.service.event.AquariumCreatedEvent
+import gr.grnet.aquarium.service.{StoreWatcherService, RabbitMQService, TimerService, EventBusService, AkkaService}
+import gr.grnet.aquarium.store.StoreProvider
+import gr.grnet.aquarium.util.date.TimeHelpers
+import gr.grnet.aquarium.util.{Loggable, Lifecycle}
+import java.io.File
import java.util.concurrent.atomic.AtomicBoolean
-import gr.grnet.aquarium.ResourceLocator._
-import gr.grnet.aquarium.computation.UserStateComputations
-import gr.grnet.aquarium.logic.accounting.algorithm.{SimpleCostPolicyAlgorithmCompiler, CostPolicyAlgorithmCompiler}
-import gr.grnet.aquarium.logic.accounting.dsl.DSLResourcesMap
-import gr.grnet.aquarium.logic.accounting.Policy
-import org.slf4j.{Logger, LoggerFactory}
+import org.slf4j.{LoggerFactory, Logger}
+import gr.grnet.aquarium.event.CreditsModel
+import gr.grnet.aquarium.charging.state.UserStateBootstrap
+import java.util.{Map ⇒ JMap}
+import java.util.{HashMap ⇒ JHashMap}
/**
- * This is the Aquarium entry point.
- *
- * Responsible to load all of application configuration and provide the relevant services.
*
- * @author Christos KK Loverdos <loverdos@gmail.com>.
+ * @author Christos KK Loverdos <loverdos@gmail.com>
*/
-final class Aquarium(val props: Props) extends Lifecycle with Loggable { aquariumSelf ⇒
- import Aquarium.Keys
+
+final class Aquarium(env: Env) extends Lifecycle with Loggable {
+
+ import Aquarium.EnvKeys
+
+ @volatile private[this] var _chargingBehaviorMap = Map[String, ChargingBehavior]()
+
+ // Caching value for the latest resource mapping
+ @volatile private[this] var _resourceMapping = apply(EnvKeys.defaultPolicyMsg).getResourceMapping
+
+ private[this] lazy val cachingPolicyStore = new CachingPolicyStore(
+ apply(EnvKeys.defaultPolicyMsg),
+ apply(EnvKeys.storeProvider).policyStore
+ )
private[this] val _isStopping = new AtomicBoolean(false)
+ override def toString = "%s/v%s".format(getClass.getName, version)
+
def isStopping() = _isStopping.get()
@inline
getClientLogger(client).warn(fmt.format(args: _*))
}
- /**
- * Reflectively provide a new instance of a class and configure it appropriately.
- */
- private[this] def newInstance[C : Manifest](_className: String = ""): C = {
- val className = _className match {
- case "" ⇒
- manifest[C].erasure.getName
-
- case name ⇒
- name
+ @throws(classOf[AquariumInternalError])
+ def apply[T: Manifest](key: TypedKey[T]): T = {
+ try {
+ env.getEx(key)
+ } catch {
+ case e: Exception ⇒
+ throw new AquariumInternalError("Could not locate %s in Aquarium environment".format(key))
}
+ }
- val instanceM = MaybeEither(defaultClassLoader.loadClass(className).newInstance().asInstanceOf[C])
- instanceM match {
- case Just(instance) ⇒ instance match {
- case configurable: Configurable ⇒
- val localProps = configurable.propertyPrefix match {
- case Some(prefix) ⇒
- props.subsetForKeyPrefix(prefix)
-
- case None ⇒
- props
- }
+ private[this] lazy val _allServices: Seq[_ <: Lifecycle] = Aquarium.ServiceKeys.map(this.apply(_))
- logger.debug("Configuring {} with props", configurable.getClass.getName)
- MaybeEither(configurable configure localProps) match {
- case Just(_) ⇒
- logger.info("Configured {} with props", configurable.getClass.getName)
- instance
+ private[this] def startServices(): Unit = {
+ for(service ← _allServices) {
+ logStartingF(service.toString) {
+ service.start()
+ } {}
+ }
+ }
- case Failed(e) ⇒
- throw new AquariumInternalError("Could not configure instance of %s".format(className), e)
- }
+ private[this] def stopServices(): Unit = {
+ val services = _allServices.reverse
- case _ ⇒
- instance
- }
+ for(service ← services) {
+ logStoppingF(service.toString) {
+ safeUnit(service.stop())
+ } {}
+ }
+ }
- case Failed(e) ⇒
- throw new AquariumInternalError("Could not instantiate %s".format(className), e)
+ private[this] def showBasicConfiguration(): Unit = {
+ for(folder ← this.eventsStoreFolder) {
+ logger.info("{} = {}", EnvKeys.eventsStoreFolder.name, folder)
}
+ this.eventsStoreFolder.throwMe // on error
+ logger.info("default policy = {}", AvroHelpers.jsonStringOfSpecificRecord(defaultPolicyMsg))
}
- private[this] lazy val _algorithmCompiler: CostPolicyAlgorithmCompiler = SimpleCostPolicyAlgorithmCompiler
+ private[this] def addShutdownHooks(): Unit = {
+ Runtime.getRuntime.addShutdownHook(new Thread(new Runnable {
+ def run = {
+ if(!_isStopping.get()) {
+ logStoppingF("Aquarium") {
+ stop()
+ } {}
+ }
+ }
+ }))
+ }
- private[this] lazy val _userStateComputations = new UserStateComputations(aquariumSelf)
+ def start(): Unit = {
+ this._isStopping.set(false)
+ showBasicConfiguration()
+ addShutdownHooks()
+ startServices()
+ }
- private[this] lazy val _actorProvider = newInstance[RoleableActorProviderService](props(Keys.actor_provider_class))
+ def stop(): Unit = {
+ this._isStopping.set(true)
+ stopServices()
+ }
/**
- * Initializes a store provider, according to the value configured
- * in the configuration file. The
+ * Stops Aquarium after the given millis. Used during testing.
*/
- private[this] lazy val _storeProvider = newInstance[StoreProvider](props(Keys.store_provider_class))
-
- private[this] lazy val _restService = newInstance[Lifecycle](props(Keys.rest_service_class))
-
- private[this] lazy val _userStateStoreM: Maybe[UserStateStore] = {
- // If there is a specific `UserStateStore` implementation specified in the
- // properties, then this implementation overrides the user store given by
- // `StoreProvider`.
- props.get(Keys.user_state_store_class) map { className ⇒
- val instance = newInstance[UserStateStore](className)
- logger.info("Overriding %s provisioning. Implementation given by: %s".format(
- shortNameOfClass(classOf[UserStateStore]),
- instance.getClass))
- instance
- }
+ def stopAfterMillis(millis: Long) {
+ Thread sleep millis
+ stop()
}
- private[this] lazy val _resourceEventStoreM: Maybe[ResourceEventStore] = {
- // If there is a specific `EventStore` implementation specified in the
- // properties, then this implementation overrides the event store given by
- // `StoreProvider`.
- props.get(Keys.resource_event_store_class) map { className ⇒
- val instance = newInstance[ResourceEventStore](className)
- logger.info("Overriding EventStore provisioning. Implementation given by: %s".format(instance.getClass))
- instance
- }
+ /**
+ * Reflectively provide a new instance of a class and configure it appropriately.
+ */
+ def newInstance[C <: AnyRef](_class: Class[C]): C = {
+ newInstance(_class.getName)
}
- private[this] lazy val _imEventStoreM: Maybe[IMEventStore] = {
- props.get(Keys.user_event_store_class) map { className ⇒
- val instance = newInstance[IMEventStore](className)
- logger.info("Overriding IMEventStore provisioning. Implementation given by: %s".format(instance.getClass))
- instance
- }
- }
+ /**
+ * Reflectively provide a new instance of a class and configure it appropriately.
+ */
+ def newInstance[C <: AnyRef](className: String): C = {
+ val originalProps = apply(EnvKeys.originalProps)
- private[this] lazy val _policyStoreM: Maybe[PolicyStore] = {
- props.get(Keys.policy_store_class) map {
- className ⇒
- val instance = newInstance[PolicyStore](className)
- logger.info("Overriding PolicyStore provisioning. Implementation given by: %s".format(instance.getClass))
- instance
- }
- }
+ val instanceM = MaybeEither(defaultClassLoader.loadClass(className).newInstance().asInstanceOf[C])
+ instanceM match {
+ case Just(instance) ⇒
+// eventBus.addSubscriber[C](instance)
+ instance match {
+ case aquariumAware: AquariumAware ⇒
+ aquariumAware.awareOfAquarium(AquariumCreatedEvent(this))
- private[this] lazy val _eventsStoreFolder: Maybe[File] = {
- props.get(Keys.events_store_folder) map {
- folderName ⇒
- logger.info("{} = {}", Keys.events_store_folder, folderName)
-
- val canonicalFolder = {
- val folder = new File(folderName)
- if(folder.isAbsolute) {
- folder.getCanonicalFile
- } else {
- logger.info("{} is not absolute, making it relative to Aquarium Home", Keys.events_store_folder)
- new File(ResourceLocator.Homes.Folders.AquariumHome, folderName).getCanonicalFile
- }
+ case _ ⇒
}
- val canonicalPath = canonicalFolder.getCanonicalPath
+ instance match {
+ case configurable: Configurable if (originalProps ne null) ⇒
+ val localProps = configurable.propertyPrefix match {
+ case somePrefix @ Some(prefix) ⇒
+ if(prefix.length == 0) {
+ logger.warn(
+ "Property prefix for %s is %s. Consider using None".format(instance, somePrefix))
+ }
- if(canonicalFolder.exists() && !canonicalFolder.isDirectory) {
- throw new AquariumInternalError("%s = %s is not a folder".format(Keys.events_store_folder, canonicalFolder))
- }
+ originalProps.subsetForKeyPrefix(prefix)
+
+ case None ⇒
+ originalProps
+ }
+
+ logger.debug("Configuring {} with props (prefix={})", configurable.getClass.getName, configurable.propertyPrefix)
+ MaybeEither(configurable configure localProps) match {
+ case Just(_) ⇒
+ logger.info("Configured {} with props (prefix={})", configurable.getClass.getName, configurable.propertyPrefix)
+
+ case Failed(e) ⇒
+ throw new AquariumInternalError("Could not configure instance of %s".format(className), e)
+ }
- // Now, events folder must be outside AQUARIUM_HOME, since AQUARIUM_HOME can be wiped out for an upgrade but
- // we still want to keep the events.
- val ahCanonicalPath = ResourceLocator.Homes.Folders.AquariumHome.getCanonicalPath
- if(canonicalPath.startsWith(ahCanonicalPath)) {
- throw new AquariumException(
- "%s = %s is under Aquarium Home = %s".format(
- Keys.events_store_folder,
- canonicalFolder,
- ahCanonicalPath
- ))
+ case _ ⇒
}
- canonicalFolder.mkdirs()
+ instance
- canonicalFolder
+ case Failed(e) ⇒
+ throw new AquariumInternalError("Could not instantiate %s".format(className), e)
}
+
}
- private[this] lazy val _events_store_save_rc_events = props.getBoolean(Keys.events_store_save_rc_events).getOr(false)
+ /**
+ * @deprecated Use `currentResourceMapping` instead
+ */
+ def resourceMappingAtMillis(millis: Long): JMap[String, ResourceTypeMsg] = {
+ val policyMspOpt = policyStore.loadPolicyAt(millis)
+ if(policyMspOpt.isEmpty) {
+ throw new AquariumInternalError(
+ "Cannot get resource mapping. Not even the default policy found for time %s",
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(millis)
+ )
+ }
- private[this] lazy val _events_store_save_im_events = props.getBoolean(Keys.events_store_save_im_events).getOr(false)
+ val policyMsg = policyMspOpt.get
+ policyMsg.getResourceMapping
+ }
- private[this] lazy val _converters = StdConverters.AllConverters
+ /**
+ * Provides the current resource mapping. This value is cached.
+ *
+ * NOTE: The assumption is that the resource mapping is always updated with new keys,
+ * that is we allow only the addition of new resource types.
+ */
+ def currentResourceMapping = {
+ this._resourceMapping synchronized this._resourceMapping
+ }
+
+ // def resourceTypesMapAtMillis(millis: Long): Map[String, ResourceType] = {
+// val policyMspOpt = policyStore.loadPolicyAt(millis)
+// if(policyMspOpt.isEmpty) {
+// throw new AquariumInternalError(
+// "Cannot get resource types map. Not even the default policy found for time %s",
+// TimeHelpers.toYYYYMMDDHHMMSSSSS(millis)
+// )
+// }
+//
+// val policyMsg = policyMspOpt.get
+// // TODO optimize
+// ModelFactory.newPolicyModel(policyMsg).resourceTypesMap
+// }
+//
+// def currentResourceTypesMap: Map[String, ResourceType] = {
+// resourceTypesMapAtMillis(TimeHelpers.nowMillis())
+// }
+
+ def unsafeValidPolicyModelAt(referenceTimeMillis: Long): PolicyModel = {
+ policyStore.loadPolicyAt(referenceTimeMillis) match {
+ case None ⇒
+ throw new AquariumInternalError(
+ "No policy found at %s".format(TimeHelpers.toYYYYMMDDHHMMSSSSS(referenceTimeMillis))
+ )
- private[this] lazy val _timerService: TimerService = newInstance[SimpleTimerService]()
+ case Some(policyMsg) ⇒
+ ModelFactory.newPolicyModel(policyMsg)
+ }
+ }
- private[this] lazy val _akka = newInstance[AkkaService]()
+ def unsafeValidPolicyAt(referenceTimeMillis: Long): PolicyMsg = {
+ unsafeValidPolicyModelAt(referenceTimeMillis).msg
+ }
- private[this] lazy val _eventBus = newInstance[EventBusService]()
+ def unsafeFullPriceTableModelForRoleAt(role: String, referenceTimeMillis: Long): FullPriceTableModel = {
+ val policyModelAtReferenceTime = unsafeValidPolicyModelAt(referenceTimeMillis)
- private[this] lazy val _rabbitmqService = newInstance[RabbitMQService]()
+ policyModelAtReferenceTime.roleMapping.get(role) match {
+ case None ⇒
+ throw new AquariumInternalError("Unknown price table for role %s at %s".format(
+ role,
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(referenceTimeMillis)
+ ))
- private[this] lazy val _storeWatcherService = newInstance[StoreWatcherService]()
+ case Some(fullPriceTable) ⇒
+ fullPriceTable
+ }
+ }
- private[this] lazy val _allServices = List(
- _timerService,
- _akka,
- _actorProvider,
- _eventBus,
- _restService,
- _rabbitmqService,
- _storeWatcherService
- )
+ def unsafeFullPriceTableForRoleAt(role: String, referenceTimeMillis: Long): FullPriceTableMsg = {
+ val policyAtReferenceTime = unsafeValidPolicyAt(referenceTimeMillis)
+ policyAtReferenceTime.getRoleMapping.get(role) match {
+ case null ⇒
+ throw new AquariumInternalError("Unknown price table for role %s at %s".format(
+ role,
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(referenceTimeMillis)
+ ))
- def get(key: String, default: String = ""): String = props.getOr(key, default)
+ case fullPriceTable ⇒
+ fullPriceTable
+ }
+ }
- def defaultClassLoader = Thread.currentThread().getContextClassLoader
+ def unsafeFullPriceTableModelForAgreement(
+ userAgreementModel: UserAgreementModel,
+ knownPolicyModel: PolicyModel
+ ): FullPriceTableModel = {
+ val policyModel = knownPolicyModel match {
+ case null ⇒
+ unsafeValidPolicyModelAt(userAgreementModel.validFromMillis)
- /**
- * FIXME: This must be ditched.
- *
- * Find a file whose location can be overiden in
- * the configuration file (e.g. policy.yaml)
- *
- * @param name Name of the file to search for
- * @param prop Name of the property that defines the file path
- * @param default Name to return if no file is found
- */
- def findConfigFile(name: String, prop: String, default: String): File = {
- // Check for the configured value first
- val configured = new File(get(prop))
- if (configured.exists)
- return configured
-
- // Look into the configuration context
- ResourceLocator.getResource(name) match {
- case Just(policyResource) ⇒
- val path = policyResource.url.getPath
- new File(path)
- case _ ⇒
- new File(default)
+ case policyModel ⇒
+ policyModel
}
- }
- private[this] def startServices(): Unit = {
- for(service ← _allServices) {
- logStartingF(service.toString) {
- service.start()
- } {}
+ userAgreementModel.fullPriceTableRef match {
+ case PolicyDefinedFullPriceTableRef ⇒
+ val role = userAgreementModel.role
+ policyModel.roleMapping.get(role) match {
+ case None ⇒
+ throw new AquariumInternalError("Unknown role %s while computing full price table for user %s at %s",
+ role,
+ userAgreementModel.userID,
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(userAgreementModel.validFromMillis)
+ )
+
+ case Some(fullPriceTable) ⇒
+ fullPriceTable
+ }
+
+ case AdHocFullPriceTableRef(fullPriceTable) ⇒
+ fullPriceTable
}
}
- private[this] def stopServices(): Unit = {
- val services = _allServices.reverse
+ def unsafeFullPriceTableForAgreement(
+ userAgreement: UserAgreementMsg,
+ knownPolicyModel: PolicyModel
+ ): FullPriceTableMsg = {
- for(service ← services) {
- logStoppingF(service.toString) {
- safeUnit(service.stop())
- } {}
+ val policyModel = knownPolicyModel match {
+ case null ⇒
+ unsafeValidPolicyModelAt(userAgreement.getValidFromMillis)
+
+ case policyModel ⇒
+ policyModel
}
- }
- def stopWithDelay(millis: Long) {
- Thread sleep millis
- stop()
+ unsafeFullPriceTableForAgreement(userAgreement, policyModel.msg)
}
- private[this] def configure(): Unit = {
- logger.info("Aquarium Home = %s".format(
- if(Homes.Folders.AquariumHome.isAbsolute)
- Homes.Folders.AquariumHome
- else
- "%s [=%s]".format(Homes.Folders.AquariumHome, Homes.Folders.AquariumHome.getCanonicalPath)
- ))
+ def unsafeFullPriceTableForAgreement(
+ userAgreement: UserAgreementMsg,
+ knownPolicy: PolicyMsg
+ ): FullPriceTableMsg = {
+ val policy = knownPolicy match {
+ case null ⇒
+ unsafeValidPolicyAt(userAgreement.getValidFromMillis)
- for(folder ← this.eventsStoreFolder) {
- logger.info("{} = {}", Aquarium.Keys.events_store_folder, folder)
+ case policy ⇒
+ policy
}
- this.eventsStoreFolder.throwMe // on error
- for(prop ← Aquarium.PropsToShow) {
- logger.info("{} = {}", prop.name, prop.rawValue)
+ val role = userAgreement.getRole
+ userAgreement.getFullPriceTableRef match {
+ case null ⇒
+ policy.getRoleMapping.get(role) match {
+ case null ⇒
+ throw new AquariumInternalError("Unknown role %s while computing full price table for user %s at %s",
+ role,
+ userAgreement.getUserID,
+ TimeHelpers.toYYYYMMDDHHMMSSSSS(userAgreement.getValidFromMillis)
+ )
+
+ case fullPriceTable ⇒
+ fullPriceTable
+ }
+
+ case fullPriceTable ⇒
+ fullPriceTable
}
+ }
+
+ /**
+ * Computes the initial user agreement for the given role and reference time. Also,
+ * records the ID from a potential related IMEvent.
+ *
+ * @param imEvent The IMEvent that creates the user
+ */
+ def initialUserAgreement(imEvent: IMEventMsg): UserAgreementModel = {
+ require(MessageHelpers.isIMEventCreate(imEvent))
- logger.info("CONF_HERE = {}", HERE)
- logger.info("{} = {}", ResourceLocator.ResourceNames.AQUARIUM_PROPERTIES, ResourceLocator.Resources.AquariumPropertiesResource)
- logger.info("{} = {}", ResourceLocator.ResourceNames.LOGBACK_XML, ResourceLocator.Resources.LogbackXMLResource)
- logger.info("{} = {}", ResourceLocator.ResourceNames.POLICY_YAML, ResourceLocator.Resources.PolicyYAMLResource)
+ val role = imEvent.getRole
+ val referenceTimeMillis = imEvent.getOccurredMillis
+
+ // Just checking
+ assert(null ne unsafeFullPriceTableModelForRoleAt(role, referenceTimeMillis))
+
+ ModelFactory.newUserAgreementModelFromIMEvent(imEvent)
}
- private[this] def addShutdownHooks(): Unit = {
- Runtime.getRuntime.addShutdownHook(new Thread(new Runnable {
- def run = {
- if(!_isStopping.get()) {
- logStoppingF("Aquarium") {
- stop()
- } {}
- }
- }
- }))
+ def initialUserBalance(role: String, referenceTimeMillis: Long): CreditsModel.Type = {
+ // FIXME: Where is the mapping?
+ CreditsModel.from(0.0)
}
- def start() = {
- this._isStopping.set(false)
- configure()
- addShutdownHooks()
- startServices()
+ def getUserStateBootstrap(imEvent: IMEventMsg): UserStateBootstrap = {
+ UserStateBootstrap(
+ this.initialUserAgreement(imEvent),
+ this.initialUserBalance(imEvent.getRole, imEvent.getOccurredMillis)
+ )
}
- def stop() = {
- this._isStopping.set(true)
- stopServices()
+ def chargingBehaviorOf(resourceType: ResourceTypeMsg): ChargingBehavior = {
+ // A resource type never changes charging behavior. By definition.
+ val className = resourceType.getChargingBehaviorClass
+ _chargingBehaviorMap.get(className) match {
+ case Some(chargingBehavior) ⇒
+ chargingBehavior
+
+ case _ ⇒
+ try {
+ _chargingBehaviorMap synchronized {
+ val chargingBehavior = newInstance[ChargingBehavior](className)
+ _chargingBehaviorMap = _chargingBehaviorMap.updated(className, chargingBehavior)
+ chargingBehavior
+ }
+ }
+ catch {
+ case e: Exception ⇒
+ throw new AquariumInternalError("Could not load charging behavior %s".format(className), e)
+ }
+ }
}
- def algorithmCompiler = _algorithmCompiler
+ def defaultPolicyMsg = apply(EnvKeys.defaultPolicyMsg)
- def userStateComputations = _userStateComputations
+ def defaultClassLoader = apply(EnvKeys.defaultClassLoader)
- def converters = _converters
-
- def actorProvider = _actorProvider
+ def resourceEventStore = apply(EnvKeys.storeProvider).resourceEventStore
- def eventBus = _eventBus
+ def imEventStore = apply(EnvKeys.storeProvider).imEventStore
- def timerService = _timerService
+ def userStateStore = apply(EnvKeys.storeProvider).userStateStore
- def userStateStore = {
- _userStateStoreM match {
- case Just(us) ⇒ us
- case _ ⇒ storeProvider.userStateStore
- }
- }
+ def policyStore = this.cachingPolicyStore
- def resourceEventStore = {
- _resourceEventStoreM match {
- case Just(es) ⇒ es
- case _ ⇒ storeProvider.resourceEventStore
- }
- }
+ def eventsStoreFolder = apply(EnvKeys.eventsStoreFolder)
- def imEventStore = {
- _imEventStoreM match {
- case Just(es) ⇒ es
- case _ ⇒ storeProvider.imEventStore
- }
- }
+ def eventBus = apply(EnvKeys.eventBus)
- def policyStore = {
- _policyStoreM match {
- case Just(es) ⇒ es
- case _ ⇒ storeProvider.policyStore
- }
- }
+ def chargingService = apply(EnvKeys.chargingService)
- def storeProvider = _storeProvider
+ def userStateTimestampThreshold = apply(EnvKeys.userStateTimestampThreshold)
- def currentResourcesMap: DSLResourcesMap = {
- // FIXME: Get rid of this singleton stuff
- Policy.policy.resourcesMap
- }
+ def adminCookie = apply(EnvKeys.adminCookie)
- def initialAgreementForRole(role: String, referenceTimeMillis: Long): String = {
- // FIXME: Where is the mapping?
- "default"
- }
+ def converters = apply(EnvKeys.converters)
- def initialBalanceForRole(role: String, referenceTimeMillis: Long): Double = {
- // FIXME: Where is the mapping?
- 10000.0
- }
+ def saveResourceEventsToEventsStoreFolder = apply(EnvKeys.eventsStoreSaveRCEvents)
- def defaultInitialUserRole: String = {
- // FIXME: Read from properties?
- "default"
- }
-
- def withStoreProviderClass[C <: StoreProvider](spc: Class[C]): Aquarium = {
- val map = this.props.map
- val newMap = map.updated(Keys.store_provider_class, spc.getName)
- val newProps = new Props(newMap)
- new Aquarium(newProps)
- }
+ def saveIMEventsToEventsStoreFolder = apply(EnvKeys.eventsStoreSaveIMEvents)
- def eventsStoreFolder = _eventsStoreFolder
+ def timerService = apply(EnvKeys.timerService)
- def saveResourceEventsToEventsStoreFolder = _events_store_save_rc_events
+ def restPort = apply(EnvKeys.restPort)
- def saveIMEventsToEventsStoreFolder = _events_store_save_im_events
+ def akkaService = apply(EnvKeys.akkaService)
- def adminCookie: MaybeOption[String] = props.get(Aquarium.Keys.admin_cookie) match {
- case just @ Just(_) ⇒ just
- case _ ⇒ NoVal
- }
+ def version = apply(EnvKeys.version)
}
object Aquarium {
SysProp.FileEncoding
)
- implicit val DefaultConverters = TheDefaultConverters
-
- final val PolicyConfName = ResourceLocator.ResourceNames.POLICY_YAML
-
- final val RolesAgreementsName = ResourceLocator.ResourceNames.ROLE_AGREEMENTS_MAP
-
- final lazy val AquariumPropertiesResource = ResourceLocator.Resources.AquariumPropertiesResource
-
- final lazy val AquariumProperties = {
- val maybeProps = Props(AquariumPropertiesResource)
- maybeProps match {
- case Just(props) ⇒
- props
-
- case NoVal ⇒
- throw new AquariumInternalError(
- "Could not load %s from %s".format(
- ResourceLocator.ResourceNames.AQUARIUM_PROPERTIES,
- AquariumPropertiesResource))
-
-
- case Failed(e) ⇒
- throw new AquariumInternalError(
- "Could not load %s from %s".format(
- ResourceLocator.ResourceNames.AQUARIUM_PROPERTIES,
- AquariumPropertiesResource),
- e)
- }
- }
-
- /**
- * The main [[gr.grnet.aquarium.Aquarium]] instance.
- */
- final lazy val Instance = {
- Maybe(new Aquarium(AquariumProperties)) match {
- case Just(masterConf) ⇒
- masterConf
-
- case NoVal ⇒
- throw new AquariumInternalError(
- "Could not create Aquarium configuration from %s".format(
- AquariumPropertiesResource))
-
- case Failed(e) ⇒
- throw new AquariumInternalError(
- "Could not create Aquarium configuration from %s".format(
- AquariumPropertiesResource),
- e)
- }
- }
-
- /**
- * Defines the names of all the known keys inside the master properties file.
- */
- final object Keys {
+ object HTTP {
+ final val RESTAdminHeaderName = "X-Aquarium-Admin-Cookie"
+ final val RESTAdminHeaderNameLowerCase = RESTAdminHeaderName.toLowerCase
+ }
+
+ final class AquariumEnvKey[T: Manifest](override val name: String) extends TypedKeySkeleton[T](name) {
+ override def toString = "%s(%s)".format(manifest[T], name)
+ }
+
+ final val ServiceKeys: List[TypedKey[_ <: Lifecycle]] = List(
+ EnvKeys.timerService,
+ EnvKeys.akkaService,
+ EnvKeys.eventBus,
+ EnvKeys.restService,
+ EnvKeys.rabbitMQService,
+ EnvKeys.storeWatcherService,
+ EnvKeys.rabbitMQProducer
+ )
+ object EnvKeys {
/**
* The Aquarium version. Will be reported in any due occasion.
*/
- final val version = "version"
+ final val version = StringKey("version")
- /**
- * The fully qualified name of the class that implements the `RoleableActorProviderService`.
- * Will be instantiated reflectively and should have a public default constructor.
- */
- final val actor_provider_class = "actor.provider.class"
-
- /**
- * The class that initializes the REST service
- */
- final val rest_service_class = "rest.service.class"
+ final val originalProps: TypedKey[Props] =
+ new AquariumEnvKey[Props]("originalProps")
/**
* The fully qualified name of the class that implements the `StoreProvider`.
* Will be instantiated reflectively and should have a public default constructor.
*/
- final val store_provider_class = "store.provider.class"
-
- /**
- * The class that implements the User store
- */
- final val user_state_store_class = "user.state.store.class"
+ final val storeProvider: TypedKey[StoreProvider] =
+ new AquariumEnvKey[StoreProvider]("store.provider.class")
/**
- * The class that implements the resource event store
- */
- final val resource_event_store_class = "resource.event.store.class"
-
- /**
- * The class that implements the IM event store
- */
- final val user_event_store_class = "user.event.store.class"
-
- /**
- * The class that implements the wallet entries store
- */
- final val policy_store_class = "policy.store.class"
-
-
- /** The lower mark for the UserActors' LRU.
- *
- * The terminology is borrowed from the (also borrowed) Apache-lucene-solr-based implementation.
+ * If a value is given to this property, then it represents a folder where all events coming to aquarium are
+ * saved.
*
+ * This is for debugging purposes.
*/
- final val user_actors_lru_lower_mark = "user.actors.LRU.lower.mark"
+ final val eventsStoreFolder: TypedKey[Option[File]] =
+ new AquariumEnvKey[Option[File]]("events.store.folder")
/**
- * The upper mark for the UserActors' LRU.
+ * If this is `true` and `events.store.folder` is defined, then all resource events are
+ * also stored in `events.store.folder`.
*
- * The terminology is borrowed from the (also borrowed) Apache-lucene-solr-based implementation.
+ * This is for debugging purposes.
*/
- final val user_actors_lru_upper_mark = "user.actors.LRU.upper.mark"
- /**
- * REST service listening port.
- *
- * Default is 8080.
- */
- final val rest_port = "rest.port"
+ final val eventsStoreSaveRCEvents = BooleanKey("events.store.save.rc.events")
/**
- * Location of the Aquarium accounting policy config file
+ * If this is `true` and `events.store.folder` is defined, then all IM events are
+ * also stored in `events.store.folder`.
+ *
+ * This is for debugging purposes.
*/
- final val aquarium_policy = "aquarium.policy"
+ final val eventsStoreSaveIMEvents = BooleanKey("events.store.save.im.events")
/**
- * Location of the role-agreement mapping file
- */
- final val aquarium_role_agreement_map = "aquarium.role-agreement.map"
-
- /**
* A time period in milliseconds for which we can tolerate stale parts regarding user state.
*
* The smaller the value, the more accurate the user credits and other state parts are.
* the timestamp of the last known balance amount by this value, then a re-computation for
* the balance is triggered.
*/
- final val user_state_timestamp_threshold = "user.state.timestamp.threshold"
+ final val userStateTimestampThreshold = LongKey("user.state.timestamp.threshold")
/**
- * The time unit is the lowest billable time period.
- * For example, with a time unit of ten seconds, if a VM is started up and shut down in nine
- * seconds, then the user will be billed for ten seconds.
- *
- * This is an overall constant. We use it as a property in order to prepare ourselves for
- * multi-cloud setup, where the same Aquarium instance is used to bill several distinct cloud
- * infrastructures.
+ * REST service listening port.
*/
- final val time_unit_in_millis = "time.unit.in.seconds"
+ final val restPort = IntKey("rest.port")
- /**
- * If a value is given to this property, then it represents a folder where all events coming to aquarium are
- * saved.
- */
- final val events_store_folder = "events.store.folder"
+ final val restShutdownTimeoutMillis = LongKey("rest.shutdown.timeout.millis")
/**
- * If this is `true` and `events.store.folder` is defined, then all resource events are
- * also stored in `events.store.folder`.
- *
- * This is for debugging purposes.
+ * A cookie used in every administrative REST API call, so that Aquarium knows it comes from
+ * an authorised client.
*/
- final val events_store_save_rc_events = "events.store.save.rc.events"
+ final val adminCookie: TypedKey[Option[String]] =
+ new AquariumEnvKey[Option[String]]("admin.cookie")
/**
- * If this is `true` and `events.store.folder` is defined, then all IM events are
- * also stored in `events.store.folder`.
- *
- * This is for debugging purposes.
+ * The class that initializes the REST service
*/
- final val events_store_save_im_events = "events.store.save.im.events"
+ final val restService: TypedKey[Lifecycle] =
+ new AquariumEnvKey[Lifecycle]("rest.service.class")
- /**
- * If set to `true`, then an IM event that cannot be parsed to [[gr.grnet.aquarium.event.model.im.IMEventModel]] is
- * saved to the [[gr.grnet.aquarium.store.IMEventStore]].
- */
- final val save_unparsed_event_im = "save.unparsed.event.im"
+ final val akkaService: TypedKey[AkkaService] =
+ new AquariumEnvKey[AkkaService]("akka.service")
- /**
- * A cookie used in every administrative REST API call, so that Aquarium knows it comes from
- * an authorised client.
- */
- final val admin_cookie = "admin.cookie"
- }
+ final val eventBus: TypedKey[EventBusService] =
+ new AquariumEnvKey[EventBusService]("event.bus.service")
- object HTTP {
- final val RESTAdminHeaderName = "X-Aquarium-Admin-Cookie"
- final val RESTAdminHeaderNameLowerCase = RESTAdminHeaderName.toLowerCase
+ final val timerService: TypedKey[TimerService] =
+ new AquariumEnvKey[TimerService]("timer.service")
+
+ final val rabbitMQService: TypedKey[RabbitMQService] =
+ new AquariumEnvKey[RabbitMQService]("rabbitmq.service")
+
+ final val rabbitMQProducer: TypedKey[RabbitMQProducer] =
+ new AquariumEnvKey[RabbitMQProducer]("rabbitmq.client")
+
+ final val storeWatcherService: TypedKey[StoreWatcherService] =
+ new AquariumEnvKey[StoreWatcherService]("store.watcher.service")
+
+ final val converters: TypedKey[Converters] =
+ new AquariumEnvKey[Converters]("converters")
+
+ final val chargingService: TypedKey[ChargingService] =
+ new AquariumEnvKey[ChargingService]("charging.service")
+
+ final val defaultClassLoader: TypedKey[ClassLoader] =
+ new AquariumEnvKey[ClassLoader]("default.class.loader")
+
+ final val defaultPolicyMsg: TypedKey[PolicyMsg] =
+ new AquariumEnvKey[PolicyMsg]("default.policy.msg")
}
}
projects / aquarium / blobdiff