2 * Copyright 2011-2012 GRNET S.A. All rights reserved.
4 * Redistribution and use in source and binary forms, with or
5 * without modification, are permitted provided that the following
8 * 1. Redistributions of source code must retain the above
9 * copyright notice, this list of conditions and the following
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.
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.
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.
36 package gr.grnet.aquarium.service
38 import com.ckkloverdos.props.Props
39 import com.google.common.eventbus.Subscribe
40 import gr.grnet.aquarium.{AquariumAwareSkeleton, Configurable}
41 import gr.grnet.aquarium.converter.StdConverters
42 import gr.grnet.aquarium.connector.rabbitmq.RabbitMQConsumer
43 import gr.grnet.aquarium.util.{Tags, Loggable, Lifecycle}
44 import gr.grnet.aquarium.util.sameTags
45 import gr.grnet.aquarium.service.event.{AquariumCreatedEvent, StoreIsAliveBusEvent, StoreIsDeadBusEvent}
46 import gr.grnet.aquarium.connector.rabbitmq.service.PayloadHandlerPostNotifier
47 import gr.grnet.aquarium.connector.rabbitmq.conf.RabbitMQKeys.RabbitMQConfKeys
48 import gr.grnet.aquarium.connector.rabbitmq.conf.RabbitMQKeys
49 import gr.grnet.aquarium.connector.handler.{SynchronousPayloadHandlerExecutor, ResourceEventPayloadHandler, IMEventPayloadHandler}
52 * The service that is responsible to handle `RabbitMQ` connecrivity.
54 * @author Christos KK Loverdos <loverdos@gmail.com>
57 class RabbitMQService extends Loggable with Lifecycle with Configurable with AquariumAwareSkeleton {
58 @volatile private[this] var _props: Props = Props()(StdConverters.AllConverters)
59 @volatile private[this] var _consumers = List[RabbitMQConsumer]()
61 def propertyPrefix = Some(RabbitMQKeys.PropertiesPrefix)
63 def eventBus = aquarium.eventBus
65 def resourceEventStore = aquarium.resourceEventStore
67 def imEventStore = aquarium.imEventStore
69 def converters = aquarium.converters
72 * Configure this instance with the provided properties.
74 * If `propertyPrefix` is defined, then `props` contains only keys that start with the given prefix.
76 def configure(props: Props) = {
81 override def awareOfAquariumEx(event: AquariumCreatedEvent) {
82 super.awareOfAquariumEx(event)
84 aquarium.eventBus.addSubscriber(this)
89 private[this] def doSetup(): Unit = {
90 val postNotifier = new PayloadHandlerPostNotifier(logger)
92 val rcHandler = new ResourceEventPayloadHandler(aquarium, logger)
94 val imHandler = new IMEventPayloadHandler(aquarium, logger)
96 val futureExecutor = new SynchronousPayloadHandlerExecutor
98 // (e)xchange:(r)outing key:(q)
100 // These two are to trigger an error if the property does not exist
101 locally(_props(RabbitMQConfKeys.rcevents_queues))
102 locally(_props(RabbitMQConfKeys.imevents_queues))
104 val all_rc_ERQs = _props.getTrimmedList(RabbitMQConfKeys.rcevents_queues)
106 val rcConsumerConfs_ = for(oneERQ ← all_rc_ERQs) yield {
107 RabbitMQKeys.makeRabbitMQConsumerConf(Tags.ResourceEventTag, _props, oneERQ)
109 val rcConsumerConfs = rcConsumerConfs_.toSet.toList
110 if(rcConsumerConfs.size != rcConsumerConfs_.size) {
112 "Duplicate %s consumer info in %s=%s".format(
113 RabbitMQKeys.PropertiesPrefix,
114 RabbitMQConfKeys.rcevents_queues,
115 _props(RabbitMQConfKeys.rcevents_queues)))
118 val all_im_ERQs = _props.getTrimmedList(RabbitMQConfKeys.imevents_queues)
119 val imConsumerConfs_ = for(oneERQ ← all_im_ERQs) yield {
120 RabbitMQKeys.makeRabbitMQConsumerConf(Tags.IMEventTag, _props, oneERQ)
122 val imConsumerConfs = imConsumerConfs_.toSet.toList
123 if(imConsumerConfs.size != imConsumerConfs_.size) {
125 "Duplicate %s consumer info in %s=%s".format(
126 RabbitMQKeys.PropertiesPrefix,
127 RabbitMQConfKeys.imevents_queues,
128 _props(RabbitMQConfKeys.imevents_queues)))
131 val rcConsumers = for(rccc ← rcConsumerConfs) yield {
132 logger.info("Declaring %s consumer {exchange=%s, routingKey=%s, queue=%s}".format(
133 RabbitMQKeys.PropertiesPrefix,
138 new RabbitMQConsumer(
147 val imConsumers = for(imcc ← imConsumerConfs) yield {
148 logger.info("Declaring %s consumer {exchange=%s, routingKey=%s, queue=%s}".format(
149 RabbitMQKeys.PropertiesPrefix,
154 new RabbitMQConsumer(
163 this._consumers = rcConsumers ++ imConsumers
165 val lg: (String ⇒ Unit) = if(this._consumers.size == 0) logger.warn(_) else logger.debug(_)
166 lg("Got %s consumers".format(this._consumers.size))
168 this._consumers.foreach(logger.debug("Configured {}", _))
176 for(consumer ← this._consumers) {
177 logStartingF(consumer.toString) {
182 for(consumer ← this._consumers) {
183 if(!consumer.isAlive()) {
184 logger.warn("Consumer not started yet %s".format(consumer))
194 for(consumer ← this._consumers) {
195 logStoppingF(consumer.toString) {
202 def handleStoreFailure(event: StoreIsDeadBusEvent): Unit = {
203 val eventTag = event.tag
205 val consumersForTag = this._consumers.filter(consumer ⇒ sameTags(consumer.conf.tag, eventTag))
206 for(consumer ← consumersForTag) {
207 if(consumer.isAlive()) {
208 // Our store is down, so we cannot accept messages anymore
209 logger.info("Shutting down %s, since store for %s is down".format(consumer, eventTag))
210 consumer.setAllowReconnects(false)
217 def handleStoreRevival(event: StoreIsAliveBusEvent): Unit = {
218 val eventTag = event.tag
220 val consumersForTag = this._consumers.filter(consumer ⇒ sameTags(consumer.conf.tag, eventTag))
221 for(consumer ← consumersForTag) {
222 if(!consumer.isAlive() && !aquarium.isStopping()) {
223 // out store is up, so we can again accept messages
224 logger.info("Starting up %s, since store for %s is alive".format(consumer, eventTag))
225 consumer.setAllowReconnects(true)