summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorJenkins <jenkins@review.openstack.org>2013-06-26 11:41:19 +0000
committerGerrit Code Review <review@openstack.org>2013-06-26 11:41:19 +0000
commitb7b4dbc41c4b361ca1afcc8ded0d47f78b53a131 (patch)
tree662e98338dc97217fe70ffc4f5729af3f1c01924
parent891287451754ac0394f1b8fa31a0505a304dccc2 (diff)
parent719eba4cae7f2bc11753b17c65c7597b8d6ed72d (diff)
downloadoslo-b7b4dbc41c4b361ca1afcc8ded0d47f78b53a131.zip
oslo-b7b4dbc41c4b361ca1afcc8ded0d47f78b53a131.tar.gz
oslo-b7b4dbc41c4b361ca1afcc8ded0d47f78b53a131.tar.xz
Merge "Don't reconnect to exclusive fanout consumers."
-rw-r--r--openstack/common/rpc/impl_qpid.py25
1 files changed, 24 insertions, 1 deletions
diff --git a/openstack/common/rpc/impl_qpid.py b/openstack/common/rpc/impl_qpid.py
index 6fa9be4..c988ae8 100644
--- a/openstack/common/rpc/impl_qpid.py
+++ b/openstack/common/rpc/impl_qpid.py
@@ -119,10 +119,17 @@ class ConsumerBase(object):
self.address = "%s ; %s" % (node_name, jsonutils.dumps(addr_opts))
- self.reconnect(session)
+ self.connect(session)
+
+ def connect(self, session):
+ """Declare the reciever on connect."""
+ self._declare_receiver(session)
def reconnect(self, session):
"""Re-declare the receiver after a qpid reconnect."""
+ self._declare_receiver(session)
+
+ def _declare_receiver(self, session):
self.session = session
self.receiver = session.receiver(self.address)
self.receiver.capacity = 1
@@ -159,6 +166,9 @@ class ConsumerBase(object):
def get_receiver(self):
return self.receiver
+ def get_node_name(self):
+ return self.address.split(';')[0]
+
class DirectConsumer(ConsumerBase):
"""Queue/consumer class for 'direct'."""
@@ -208,6 +218,7 @@ class FanoutConsumer(ConsumerBase):
'topic' is the topic to listen on
'callback' is the callback to call when messages are received
"""
+ self.conf = conf
super(FanoutConsumer, self).__init__(
session, callback,
@@ -216,6 +227,18 @@ class FanoutConsumer(ConsumerBase):
"%s_fanout_%s" % (topic, uuid.uuid4().hex),
{"exclusive": True})
+ def reconnect(self, session):
+ topic = self.get_node_name()
+ params = {
+ 'session': session,
+ 'topic': topic,
+ 'callback': self.callback,
+ }
+
+ self.__init__(conf=self.conf, **params)
+
+ super(FanoutConsumer, self).reconnect(session)
+
class Publisher(object):
"""Base Publisher class."""