diff options
Diffstat (limited to 'cpp/src')
| -rw-r--r-- | cpp/src/qpid/broker/SessionAdapter.cpp | 12 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/SessionAdapter.h | 2 | ||||
| -rwxr-xr-x | cpp/src/tests/federation.py | 2 |
3 files changed, 11 insertions, 5 deletions
diff --git a/cpp/src/qpid/broker/SessionAdapter.cpp b/cpp/src/qpid/broker/SessionAdapter.cpp index 15a47de45d..4450d56efb 100644 --- a/cpp/src/qpid/broker/SessionAdapter.cpp +++ b/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/cpp/src/qpid/broker/SessionAdapter.h b/cpp/src/qpid/broker/SessionAdapter.h index 4eaaf13f8d..60a5a0f285 100644 --- a/cpp/src/qpid/broker/SessionAdapter.h +++ b/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/cpp/src/tests/federation.py b/cpp/src/tests/federation.py index 92162fd98d..ad82964007 100755 --- a/cpp/src/tests/federation.py +++ b/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)) |