QPID-1526: add checks for exclusive ownership plus tests to verify they are executed

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

4 files changed, 28 insertions, 6 deletions

diff --git a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
index 15a47de45d..4450d56efb 100644
--- a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
+++ b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
@@ -346,7 +346,7 @@ void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string&
         std::pair<Queue::shared_ptr, bool> queue_created =  
             getBroker().getQueues().declare(name, durable,
                                             autoDelete,
-                                            exclusive ? this : 0);
+                                            exclusive ? &session : 0);
         queue = queue_created.first;
         assert(queue);
         if (queue_created.second) { // This is a new queue
@@ -367,7 +367,7 @@ void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string&
                 exclusiveQueues.push_back(queue);
             }
         } else {
-            if (exclusive && queue->setExclusiveOwner(this)) {
+            if (exclusive && queue->setExclusiveOwner(&session)) {
                 exclusiveQueues.push_back(queue);
             }
         }
@@ -379,7 +379,7 @@ void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string&
                                                       queue_created.second ? "created" : "existing"));
     }
 
-    if (exclusive && !queue->isExclusiveOwner(this)) 
+    if (exclusive && !queue->isExclusiveOwner(&session)) 
         throw ResourceLockedException(QPID_MSG("Cannot grant exclusive access to queue "
                                                << queue->getName()));
 } 
@@ -405,6 +405,9 @@ void SessionAdapter::QueueHandlerImpl::delete_(const string& queue, bool ifUnuse
     }
 
     Queue::shared_ptr q = getQueue(queue);
+    if (q->hasExclusiveOwner() && !q->isExclusiveOwner(&session)) 
+        throw ResourceLockedException(QPID_MSG("Cannot delete queue "
+                                               << queue << "; it is exclusive to another session"));
     if(ifEmpty && q->getMessageCount() > 0){
         throw PreconditionFailedException("Queue not empty.");
     }else if(ifUnused && q->getConsumerCount() > 0){
@@ -473,6 +476,9 @@ SessionAdapter::MessageHandlerImpl::subscribe(const string& queueName,
     Queue::shared_ptr queue = getQueue(queueName);
     if(!destination.empty() && state.exists(destination))
         throw NotAllowedException(QPID_MSG("Consumer tags must be unique"));
+    if (queue->hasExclusiveOwner() && !queue->isExclusiveOwner(&session)) 
+        throw ResourceLockedException(QPID_MSG("Cannot subscribe to exclusive queue "
+                                               << queue->getName()));
 
     state.consume(destination, queue, 
                   acceptMode == 0, acquireMode == 0, exclusive, 
diff --git a/qpid/cpp/src/qpid/broker/SessionAdapter.h b/qpid/cpp/src/qpid/broker/SessionAdapter.h
index 4eaaf13f8d..60a5a0f285 100644
--- a/qpid/cpp/src/qpid/broker/SessionAdapter.h
+++ b/qpid/cpp/src/qpid/broker/SessionAdapter.h
@@ -109,7 +109,7 @@ class Queue;
     };
 
     class QueueHandlerImpl : public QueueHandler,
-            public HandlerHelper, public OwnershipToken
+            public HandlerHelper
     {
         Broker& broker;
         std::vector< boost::shared_ptr<Queue> > exclusiveQueues;
diff --git a/qpid/cpp/src/tests/federation.py b/qpid/cpp/src/tests/federation.py
index 92162fd98d..ad82964007 100755
--- a/qpid/cpp/src/tests/federation.py
+++ b/qpid/cpp/src/tests/federation.py
@@ -177,7 +177,7 @@ class FederationTests(TestBase010):
         #setup queue on remote broker and add some messages
         r_conn = self.connect(host=remote_host(), port=remote_port())
         r_session = r_conn.session("test_pull_from_queue")
-        r_session.queue_declare(queue="my-bridge-queue", exclusive=True, auto_delete=True)
+        r_session.queue_declare(queue="my-bridge-queue", auto_delete=True)
         for i in range(1, 6):
             dp = r_session.delivery_properties(routing_key="my-bridge-queue")
             r_session.message_transfer(message=Message(dp, "Message %d" % i))
diff --git a/qpid/python/tests_0-10/queue.py b/qpid/python/tests_0-10/queue.py
index a3b23a1c32..05e18081fa 100644
--- a/qpid/python/tests_0-10/queue.py
+++ b/qpid/python/tests_0-10/queue.py
@@ -88,7 +88,7 @@ class QueueTests(TestBase010):
         # TestBase.setUp has already opened session(1)
         s1 = self.session
         # Here we open a second separate connection:
-        s2 = self.conn.session("other", 2)
+        s2 = self.conn.session("other")
 
         #declare an exclusive queue:
         s1.queue_declare(queue="exclusive-queue", exclusive=True, auto_delete=True)
@@ -98,6 +98,22 @@ 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:
+            s3.message_subscribe(queue="exclusive-queue")
+            self.fail("Expected message_subscribe on an exclusive queue to raise a channel exception")
+        except SessionException, e:
+            self.assertEquals(405, e.args[0].error_code)
+
+        s4 = self.conn.session("deleter")
+        try:
+            #other connection should not be allowed to declare this:
+            s4.queue_delete(queue="exclusive-queue")
+            self.fail("Expected queue_delete on an exclusive queue to raise a channel exception")
+        except SessionException, e:
+            self.assertEquals(405, e.args[0].error_code)
 
 
     def test_declare_passive(self):