Sync rpc fix from oslo-incubator

Sync the following fixes from oslo-incubator:

ef406a2 Create a shared queue for QPID topic consumers

Closes-bug: #1257293
Change-Id: Id0e1d9704ef1cb506ae68c3848f8615e7e364484

1 files changed, 6 insertions, 3 deletions

diff --git a/heat/openstack/common/rpc/impl_qpid.py b/heat/openstack/common/rpc/impl_qpid.py
index ad0a4b5a4..cc44b48f2 100644
--- a/heat/openstack/common/rpc/impl_qpid.py
+++ b/heat/openstack/common/rpc/impl_qpid.py
@@ -131,14 +131,13 @@ class ConsumerBase(object):
                     },
                 },
             }
-            if link_name:
-                addr_opts["link"]["name"] = link_name
             addr_opts["node"]["x-declare"].update(node_opts)
         elif conf.qpid_topology_version == 2:
             addr_opts = {
                 "link": {
                     "x-declare": {
                         "auto-delete": True,
+                        "exclusive": False,
                     },
                 },
             }
@@ -146,6 +145,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))
 
@@ -220,14 +221,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)