From 0745cd0d738fbec7bd48529d7024535fc3301931 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Mon, 10 Dec 2007 20:22:23 +0000 Subject: Patches from Ted Ross QPID-697 Fixed access-rights constants for management schema. Added mutex to fix problems associated with concurrent invocation of accessors for queue statistics. Removed queue schema content that is not relevant to QPID. QPID-698 This patch creates a new subdirectory in python called "mgmt-cli". python/mgmt-cli/main.py can be executed from the shell. If no arguments are supplied, it attempts to connect to the broker at localhost:5672. The first argument is the hostname for the target broker and the second (optional) argument is the TCP port (defaults to 5672). It is assumed that the AMQP spec file is in the following location: /usr/share/amqp/amqp.0-10-preview.xml It is also required that the qpid/python directory be in the PYTHONPATH environment variable. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@603034 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/src/qpid/broker/Queue.cpp | 7 +++++++ cpp/src/qpid/management/ManagementObject.h | 4 ++-- 2 files changed, 9 insertions(+), 2 deletions(-) (limited to 'cpp/src') diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp index e2fd998cc0..a5384014d8 100644 --- a/cpp/src/qpid/broker/Queue.cpp +++ b/cpp/src/qpid/broker/Queue.cpp @@ -97,6 +97,7 @@ void Queue::deliver(intrusive_ptr& msg){ push(msg); msg->enqueueComplete(); if (mgmtObject.get() != 0) { + Mutex::ScopedLock alock(mgmtObject->accessorLock); mgmtObject->inc_msgTotalEnqueues (); mgmtObject->inc_byteTotalEnqueues (msg->contentSize ()); mgmtObject->inc_msgDepth (); @@ -104,6 +105,7 @@ void Queue::deliver(intrusive_ptr& msg){ } }else { if (mgmtObject.get() != 0) { + Mutex::ScopedLock alock(mgmtObject->accessorLock); mgmtObject->inc_msgTotalEnqueues (); mgmtObject->inc_byteTotalEnqueues (msg->contentSize ()); mgmtObject->inc_msgDepth (); @@ -122,6 +124,7 @@ void Queue::recover(intrusive_ptr& msg){ push(msg); msg->enqueueComplete(); // mark the message as enqueued if (mgmtObject.get() != 0) { + Mutex::ScopedLock alock(mgmtObject->accessorLock); mgmtObject->inc_msgTotalEnqueues (); mgmtObject->inc_byteTotalEnqueues (msg->contentSize ()); mgmtObject->inc_msgPersistEnqueues (); @@ -140,6 +143,7 @@ void Queue::recover(intrusive_ptr& msg){ void Queue::process(intrusive_ptr& msg){ push(msg); if (mgmtObject.get() != 0) { + Mutex::ScopedLock alock(mgmtObject->accessorLock); mgmtObject->inc_msgTotalEnqueues (); mgmtObject->inc_byteTotalEnqueues (msg->contentSize ()); mgmtObject->inc_msgTxnEnqueues (); @@ -323,6 +327,7 @@ void Queue::consume(Consumer&, bool requestExclusive){ consumerCount++; if (mgmtObject.get() != 0){ + Mutex::ScopedLock alock(mgmtObject->accessorLock); mgmtObject->inc_consumers (); } } @@ -333,6 +338,7 @@ void Queue::cancel(Consumer& c){ consumerCount--; if(exclusive) exclusive = false; if (mgmtObject.get() != 0){ + Mutex::ScopedLock alock(mgmtObject->accessorLock); mgmtObject->dec_consumers (); } } @@ -363,6 +369,7 @@ void Queue::pop(){ if (policy.get()) policy->dequeued(msg.payload->contentSize()); if (mgmtObject.get() != 0){ + Mutex::ScopedLock alock(mgmtObject->accessorLock); mgmtObject->inc_msgTotalDequeues (); mgmtObject->inc_byteTotalDequeues (msg.payload->contentSize()); mgmtObject->dec_msgDepth (); diff --git a/cpp/src/qpid/management/ManagementObject.h b/cpp/src/qpid/management/ManagementObject.h index a8ba231419..ff136c397d 100644 --- a/cpp/src/qpid/management/ManagementObject.h +++ b/cpp/src/qpid/management/ManagementObject.h @@ -54,8 +54,8 @@ class ManagementObject static const uint8_t TYPE_LSTR = 7; static const uint8_t ACCESS_RC = 1; - static const uint8_t ACCESS_RW = 1; - static const uint8_t ACCESS_RO = 1; + static const uint8_t ACCESS_RW = 2; + static const uint8_t ACCESS_RO = 3; static const uint8_t DIR_I = 1; static const uint8_t DIR_O = 2; -- cgit v1.2.1