QPID-5283: ensure queue has no consumers before granting exclusive ownership

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1537579 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 23 insertions, 4 deletions

diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index 19b18e1b0e..a805e110ad 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/cpp/src/qpid/broker/Queue.cpp
@@ -1323,7 +1323,7 @@ bool Queue::setExclusiveOwner(const OwnershipToken* const o)
         autoDeleteTask->cancel();
     }
     Mutex::ScopedLock locker(messageLock);
-    if (owner) {
+    if (owner  || users.hasConsumers()) {
         return false;
     } else {
         owner = o;
diff --git a/qpid/cpp/src/tests/acl.py b/qpid/cpp/src/tests/acl.py
index fe8254a6fa..66705e6d24 100755
--- a/qpid/cpp/src/tests/acl.py
+++ b/qpid/cpp/src/tests/acl.py
@@ -1570,7 +1570,8 @@ class ACLTests(TestBase010):
             session = self.get_session('bob','bob')
 
         try:
-            session.message_subscribe(queue='q3', destination='myq1')
+            session.message_subscribe(queue='q3', destination='myq3')
+            session.message_cancel(destination='myq3')
         except qpid.session.SessionException, e:
             if (403 == e.args[0].error_code):
                 self.fail("ACL should allow subscription for q3");
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_10/dtx.py b/qpid/tests/src/py/qpid_tests/broker_0_10/dtx.py
index 19a5c6a8d9..f8d6533a78 100644
--- a/qpid/tests/src/py/qpid_tests/broker_0_10/dtx.py
+++ b/qpid/tests/src/py/qpid_tests/broker_0_10/dtx.py
@@ -386,7 +386,6 @@ class DtxTests(TestBase010):
         """
         guard = self.keepQueuesAlive(["tx-queue"])
         session = self.conn.session("alternate", 1)
-        session.queue_declare(queue="tx-queue", exclusive=True, auto_delete=True)
 
         #publish a message under a transaction
         session.dtx_select()
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py b/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
index 7e8218d092..545bc8cc23 100644
--- a/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
+++ b/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
@@ -99,7 +99,7 @@ class QueueTests(TestBase010):
             self.fail("Expected second exclusive queue_declare to raise a channel exception")
         except SessionException, e:
             self.assertEquals(405, e.args[0].error_code)
-            
+
         s3 = self.conn.session("subscriber")
         try:
             #other connection should not be allowed to declare this:
@@ -132,6 +132,25 @@ class QueueTests(TestBase010):
         except SessionException, e:
             self.assertEquals(405, e.args[0].error_code)
 
+    def test_declare_exclusive_alreadyinuse(self):
+        """
+        Test that exclusivity is real if granted
+        """
+        # TestBase.setUp has already opened session(1)
+        s1 = self.session
+        # Here we open a second separate connection:
+        s2 = self.conn.session("other")
+
+        #declare an exclusive queue:
+        s1.queue_declare(queue="a-queue", auto_delete=True)
+        s1.message_subscribe(queue="a-queue")
+        try:
+            #other connection should not be allowed to declare this:
+            s2.queue_declare(queue="a-queue", exclusive=True, auto_delete=True)
+            self.fail("Expected request for exclusivity to fail")
+        except SessionException, e:
+            self.assertEquals(405, e.args[0].error_code)
+
     def test_declare_passive(self):
         """
         Test that the passive field is honoured in queue.declare