11package com .flipcast .rmq
22
3- import scala .collection .mutable
4- import com .github .sstone .amqp .{Amqp , RabbitMQConnection }
3+ import com .github .sstone .amqp .{ChannelOwner , Consumer , ConnectionOwner , Amqp }
54import java .util .concurrent .atomic .AtomicInteger
65import com .flipcast .Flipcast
7- import akka .actor .{ActorRef , ActorSystem }
8- import com .rabbitmq .client .{AMQP , Address }
9- import scala .concurrent .duration .Duration
10- import java .util .concurrent .TimeUnit ._
6+ import akka .actor .{PoisonPill , ActorRef , ActorSystem }
7+ import com .rabbitmq .client .{ConnectionFactory , AMQP , Address }
118import com .github .sstone .amqp .Amqp ._
129import akka .event .slf4j .Logger
1310import com .github .sstone .amqp .Amqp .DeclareQueue
@@ -16,6 +13,8 @@ import com.github.sstone.amqp.Amqp.QueueParameters
1613import com .github .sstone .amqp .Amqp .DeclareExchange
1714import com .github .sstone .amqp .Amqp .QueueBind
1815import java .util .concurrent .{LinkedBlockingQueue , TimeUnit , ThreadPoolExecutor }
16+ import scala .concurrent .duration ._
17+ import collection .JavaConverters ._
1918
2019/**
2120 * RabbitMQ connection helper
@@ -37,14 +36,14 @@ object ConnectionHelper {
3736
3837
3938 /**
40- * Keep track of open connections
39+ * RMQ Connection
4140 */
42- val connections : mutable. HashMap [ String , RabbitMQConnection ] = new mutable. HashMap [ String , RabbitMQConnection ]() with mutable. SynchronizedMap [ String , RabbitMQConnection ]
41+ var rmqConnection : ActorRef = null
4342
4443 /**
4544 * Client properties that we need to set (in case of rmq cluster)
4645 */
47- val clientProps = Map ( " x-ha-policy" -> " all"
46+ val clientProps = Map [ String , AnyRef ]( " ha-mode " -> " all " , " x-ha-policy" -> " all" , " x-priority " -> new Integer ( 10 ) )
4847
4948 /**
5049 * Keep a count of connection
@@ -59,23 +58,18 @@ object ConnectionHelper {
5958 * @return A RabbitMQ connection
6059 */
6160 private def createConnection () (implicit system : ActorSystem ) = {
62- val name = " flipcast-%s"
63- val reconnectDelay = Duration (Flipcast .rmqConfig.reconnectDelay, MILLISECONDS )
64- val executor = Option (executorService)
65- val addresses = Flipcast .rmqConfig.hosts.map( h => {
66- val tokens = h.split(" :"
67- new Address (tokens(0 ), tokens(1 ).toInt)
68- }).toArray
69- val conn = new RabbitMQConnection (name = name,
70- vhost = Flipcast .rmqConfig.vhost,
71- user = Flipcast .rmqConfig.user,
72- password = Flipcast .rmqConfig.pass,
73- reconnectionDelay = reconnectDelay,
74- executor = executor,
75- addresses = Option (addresses))
76- conn.waitForConnection.await()
77- connections.put(name, conn)
78- conn
61+ if (rmqConnection == null ) {
62+ val connectionFactory = new ConnectionFactory ()
63+ connectionFactory.setClientProperties(clientProps.asJava)
64+ connectionFactory.setAutomaticRecoveryEnabled(true )
65+ connectionFactory.setNetworkRecoveryInterval(10 )
66+ val addresses = Flipcast .rmqConfig.hosts.map( h => {
67+ val tokens = h.split(" :"
68+ new Address (tokens(0 ), tokens(1 ).toInt)
69+ }).toArray
70+ rmqConnection = system.actorOf(ConnectionOwner .props(connectionFactory, 1 second, addresses = Option (addresses), executor = Option (executorService)))
71+ }
72+ rmqConnection
7973 }
8074
8175 /**
@@ -87,14 +81,13 @@ object ConnectionHelper {
8781 * @return ActorRef of consumer
8882 */
8983 def createConsumer (queueName : String , exchange : String , listener : ActorRef , qos : Int ) (implicit system : ActorSystem ) = {
90- val channelParameters = Option (ChannelParameters (qos))
9184 val exchangeParams = ExchangeParameters (name = exchange, passive = false ,
9285 exchangeType = " direct" = true , autodelete = false , clientProps)
86+ val cParams = Option (ChannelParameters (qos))
87+ val queueParams = QueueParameters (queueName, passive = false , durable = true , exclusive = false , autodelete = false , clientProps)
9388 val connection = createConnection()
94- val queueParams = QueueParameters (queueName, passive = false , durable = true , exclusive = false ,
95- autodelete = false , clientProps)
96- val consumer = connection.createConsumer(exchangeParams, queueParams, queueName, listener,
97- channelParams = channelParameters, autoack = false )
89+ val consumer = ConnectionOwner .createChildActor(connection, Consumer .props(listener, exchangeParams,queueParams, queueName,
90+ cParams, autoack = false ))
9891 Amqp .waitForConnection(system, consumer).await()
9992 consumer ! DeclareExchange (exchangeParams)
10093 consumer ! DeclareQueue (queueParams)
@@ -114,10 +107,10 @@ object ConnectionHelper {
114107 val channelParameters = Option (ChannelParameters (1 ))
115108 val exchangeParams = ExchangeParameters (name = exchange, passive = false ,
116109 exchangeType = " direct" = true , autodelete = false , clientProps)
117- val connection = createConnection()
118110 val queueParams = QueueParameters (queueName, passive = false , durable = true , exclusive = false , autodelete = false ,
119111 clientProps)
120- val producer = connection.createChannelOwner(channelParameters)
112+ val connection = createConnection()
113+ val producer = ConnectionOwner .createChildActor(connection, ChannelOwner .props(channelParams = channelParameters))
121114 Amqp .waitForConnection(system, producer).await()
122115 producer ! DeclareExchange (exchangeParams)
123116 producer ! DeclareQueue (queueParams)
@@ -129,10 +122,9 @@ object ConnectionHelper {
129122 * Close all connections
130123 */
131124 def stop () {
132- connections.foreach( c => {
133- log.info(" Closing connection: " + c._1)
134- c._2.stop
135- })
125+ if (rmqConnection != null ) {
126+ rmqConnection ! PoisonPill
127+ }
136128 }
137129
138130}
0 commit comments