From cbd7acfc294c4bf2ae0014bcf04646382bed469d Mon Sep 17 00:00:00 2001 From: Flavio Percoco Date: Fri, 6 Dec 2013 11:06:09 +0100 Subject: [PATCH] Sync rpc fix from oslo-incubator Sync the following fixes from oslo-incubator: ef406a2 Create a shared queue for QPID topic consumers e227c0e Properly reconnect subscribing clients when QPID broker restarts Change-Id: I934a6ea6e9eb510513640870750ea37e6f076df9 Closes-bug: #1251757 Closes-bug: #1257293 --- cinder/openstack/common/rpc/impl_qpid.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/cinder/openstack/common/rpc/impl_qpid.py b/cinder/openstack/common/rpc/impl_qpid.py index 1137278d8..f6e38dd11 100644 --- a/cinder/openstack/common/rpc/impl_qpid.py +++ b/cinder/openstack/common/rpc/impl_qpid.py @@ -18,7 +18,6 @@ import functools import itertools import time -import uuid import eventlet import greenlet @@ -120,7 +119,6 @@ class ConsumerBase(object): }, }, "link": { - "name": link_name, "durable": True, "x-declare": { "durable": False, @@ -135,6 +133,7 @@ class ConsumerBase(object): "link": { "x-declare": { "auto-delete": True, + "exclusive": False, }, }, } @@ -142,6 +141,8 @@ class ConsumerBase(object): raise_invalid_topology_version() addr_opts["link"]["x-declare"].update(link_opts) + if link_name: + addr_opts["link"]["name"] = link_name self.address = "%s ; %s" % (node_name, jsonutils.dumps(addr_opts)) @@ -188,14 +189,16 @@ class DirectConsumer(ConsumerBase): if conf.qpid_topology_version == 1: node_name = "%s/%s" % (msg_id, msg_id) node_opts = {"type": "direct"} + link_name = msg_id elif conf.qpid_topology_version == 2: node_name = "amq.direct/%s" % msg_id node_opts = {} + link_name = None else: raise_invalid_topology_version() super(DirectConsumer, self).__init__(conf, session, callback, - node_name, node_opts, msg_id, + node_name, node_opts, link_name, link_opts) @@ -246,16 +249,14 @@ class FanoutConsumer(ConsumerBase): if conf.qpid_topology_version == 1: node_name = "%s_fanout" % topic node_opts = {"durable": False, "type": "fanout"} - link_name = "%s_fanout_%s" % (topic, uuid.uuid4().hex) elif conf.qpid_topology_version == 2: node_name = "amq.topic/fanout/%s" % topic node_opts = {} - link_name = "" else: raise_invalid_topology_version() super(FanoutConsumer, self).__init__(conf, session, callback, - node_name, node_opts, link_name, + node_name, node_opts, None, link_opts) -- 2.45.2