You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kg...@apache.org on 2011/08/16 02:37:35 UTC
svn commit: r1158073 [1/2] - in /qpid/branches/qpid-3346/qpid: ./
cpp/docs/api/ cpp/examples/old_api/tradedemo/ cpp/include/qmf/engine/
cpp/src/ cpp/src/qmf/engine/ cpp/src/qpid/broker/ cpp/src/qpid/cluster/
cpp/src/qpid/management/ cpp/src/tests/ dotn...
Author: kgiusti
Date: Tue Aug 16 00:37:33 2011
New Revision: 1158073
URL: http://svn.apache.org/viewvc?rev=1158073&view=rev
Log:
QPID-3346: checkpoint - added mgmt interfaces, fleshed out sub-optimal implementation
Modified:
qpid/branches/qpid-3346/qpid/ (props changed)
qpid/branches/qpid-3346/qpid/cpp/docs/api/ (props changed)
qpid/branches/qpid-3346/qpid/cpp/examples/old_api/tradedemo/tradedemo_topic_publisher.vcproj (props changed)
qpid/branches/qpid-3346/qpid/cpp/include/qmf/engine/Agent.h (props changed)
qpid/branches/qpid-3346/qpid/cpp/include/qmf/engine/Console.h (props changed)
qpid/branches/qpid-3346/qpid/cpp/src/CMakeLists.txt (props changed)
qpid/branches/qpid-3346/qpid/cpp/src/qmf/engine/Agent.cpp (props changed)
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.h
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.cpp
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.h
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.cpp
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.h
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Messages.h
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.cpp
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.h
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.h
qpid/branches/qpid-3346/qpid/cpp/src/qpid/cluster/UpdateClient.cpp (props changed)
qpid/branches/qpid-3346/qpid/cpp/src/qpid/cluster/UpdateClient.h (props changed)
qpid/branches/qpid-3346/qpid/cpp/src/qpid/management/ManagementAgent.cpp (props changed)
qpid/branches/qpid-3346/qpid/cpp/src/qpid/management/ManagementAgent.h (props changed)
qpid/branches/qpid-3346/qpid/cpp/src/tests/QueueTest.cpp
qpid/branches/qpid-3346/qpid/cpp/src/tests/cluster_tests.py (props changed)
qpid/branches/qpid-3346/qpid/dotnet/build-msbuild.bat (props changed)
qpid/branches/qpid-3346/qpid/dotnet/build-nant-release (props changed)
qpid/branches/qpid-3346/qpid/dotnet/build-nant.bat (props changed)
qpid/branches/qpid-3346/qpid/java/ (props changed)
qpid/branches/qpid-3346/qpid/java/broker/ (props changed)
qpid/branches/qpid-3346/qpid/java/broker/bin/ (props changed)
qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/ (props changed)
qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ (props changed)
qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/ (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ConfigurationManagement.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/LoggingManagement.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedExchange.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/UserManagement.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanAttribute.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanConstructor.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanDescription.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperation.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperationParameter.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/ (props changed)
qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ConnectionTypeTabControl.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ExchangeTypeTabControl.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/MBeanTypeTabControl.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/QueueTypeTabControl.java (props changed)
qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/resources/macosx/Contents/MacOS/qpidmc (props changed)
qpid/branches/qpid-3346/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/ (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/CPPExcludes (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/Excludes (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/JavaExcludes (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/JavaPre010Excludes (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/JavaTransientExcludes (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/XAExcludes (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/clean-dir (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/cpp.async.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/cpp.cluster.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/cpp.noprefetch.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/cpp.ssl.excludes (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/cpp.ssl.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/cpp.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/java-dby-spawn.0-9-1.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/java-dby.0-9-1.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/java-mms-spawn.0-10.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/java-mms-spawn.0-9-1.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/java-mms.0-9-1.testprofile (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/log4j-test.xml (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/test-provider.properties (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/test_resources/ (props changed)
qpid/branches/qpid-3346/qpid/java/test-profiles/testprofile.defaults (props changed)
qpid/branches/qpid-3346/qpid/packaging/windows/ (props changed)
qpid/branches/qpid-3346/qpid/python/ (props changed)
qpid/branches/qpid-3346/qpid/python/examples/api/spout (props changed)
qpid/branches/qpid-3346/qpid/python/qpid/concurrency.py (props changed)
qpid/branches/qpid-3346/qpid/ruby/ext/sasl/extconf.rb (props changed)
qpid/branches/qpid-3346/qpid/specs/management-schema.xml
qpid/branches/qpid-3346/qpid/tests/src/py/qpid_tests/broker_0_10/management.py
qpid/branches/qpid-3346/qpid/tests/src/py/qpid_tests/broker_0_9/queue.py (props changed)
Propchange: qpid/branches/qpid-3346/qpid/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/0.6-release-windows-installer/qpid:926803,927233
/qpid/branches/java-network-refactor/qpid:805429-825319
/qpid/branches/qpid-2935/qpid:1061302-1072333
+/qpid/trunk/qpid:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/docs/api/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -4,3 +4,4 @@
/qpid/branches/0.6-release-windows-installer/qpid/cpp/docs/api:926803,927233
/qpid/branches/java-network-refactor/qpid/cpp/docs/api:805429-825319
/qpid/branches/qpid-2935/qpid/cpp/docs/api:1061302-1072333
+/qpid/trunk/qpid/cpp/docs/api:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/examples/old_api/tradedemo/tradedemo_topic_publisher.vcproj
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/0.6-release-windows-installer/qpid/cpp/examples/tradedemo/tradedemo_topic_publisher.vcproj:926803,927218,927233
/qpid/branches/java-network-refactor/qpid/cpp/examples/tradedemo/tradedemo_topic_publisher.vcproj:805429-825319
/qpid/branches/qpid-2935/qpid/cpp/examples/tradedemo/tradedemo_topic_publisher.vcproj:1061302-1072333
+/qpid/trunk/qpid/cpp/examples/old_api/tradedemo/tradedemo_topic_publisher.vcproj:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/include/qmf/engine/Agent.h
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/include/qmf/engine/Agent.h:1061302-1072333
+/qpid/trunk/qpid/cpp/include/qmf/engine/Agent.h:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/include/qmf/engine/Console.h
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/include/qmf/engine/Console.h:1061302-1072333
+/qpid/trunk/qpid/cpp/include/qmf/engine/Console.h:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/src/CMakeLists.txt
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/0.6-release-windows-installer/qpid/cpp/src/CMakeLists.txt:926803,927233,932132
/qpid/branches/java-network-refactor/qpid/cpp/src/CMakeLists.txt:805429-825319
/qpid/branches/qpid-2935/qpid/cpp/src/CMakeLists.txt:1061302-1072333
+/qpid/trunk/qpid/cpp/src/CMakeLists.txt:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/src/qmf/engine/Agent.cpp
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/src/qmf/engine/Agent.cpp:1061302-1072333
+/qpid/trunk/qpid/cpp/src/qmf/engine/Agent.cpp:1156188
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp Tue Aug 16 00:37:33 2011
@@ -37,6 +37,7 @@
#include "qmf/org/apache/qpid/broker/Package.h"
#include "qmf/org/apache/qpid/broker/ArgsBrokerCreate.h"
#include "qmf/org/apache/qpid/broker/ArgsBrokerDelete.h"
+#include "qmf/org/apache/qpid/broker/ArgsBrokerQuery.h"
#include "qmf/org/apache/qpid/broker/ArgsBrokerEcho.h"
#include "qmf/org/apache/qpid/broker/ArgsBrokerGetLogLevel.h"
#include "qmf/org/apache/qpid/broker/ArgsBrokerQueueMoveMessages.h"
@@ -452,7 +453,7 @@ Manageable::status_t Broker::ManagementM
_qmf::ArgsBrokerQueueMoveMessages& moveArgs=
dynamic_cast<_qmf::ArgsBrokerQueueMoveMessages&>(args);
QPID_LOG (debug, "Broker::queueMoveMessages()");
- if (queueMoveMessages(moveArgs.i_srcQueue, moveArgs.i_destQueue, moveArgs.i_qty))
+ if (queueMoveMessages(moveArgs.i_srcQueue, moveArgs.i_destQueue, moveArgs.i_qty, moveArgs.i_filter))
status = Manageable::STATUS_OK;
else
return Manageable::STATUS_PARAMETER_INVALID;
@@ -482,6 +483,13 @@ Manageable::status_t Broker::ManagementM
status = Manageable::STATUS_OK;
break;
}
+ case _qmf::Broker::METHOD_QUERY :
+ {
+ _qmf::ArgsBrokerQuery& a = dynamic_cast<_qmf::ArgsBrokerQuery&>(args);
+ status = queryObject(a.i_type, a.i_name, a.o_results, getManagementExecutionContext());
+ status = Manageable::STATUS_OK;
+ break;
+ }
default:
QPID_LOG (debug, "Broker ManagementMethod not implemented: id=" << methodId << "]");
status = Manageable::STATUS_NOT_IMPLEMENTED;
@@ -654,6 +662,49 @@ void Broker::deleteObject(const std::str
}
+Manageable::status_t Broker::queryObject(const std::string& type,
+ const std::string& name,
+ Variant::Map& results,
+ const ConnectionState* context)
+{
+ std::string userId;
+ std::string connectionId;
+ if (context) {
+ userId = context->getUserId();
+ connectionId = context->getUrl();
+ }
+ QPID_LOG (debug, "Broker::query(" << type << ", " << name << ")");
+
+ if (type == TYPE_QUEUE)
+ return queryQueue( name, userId, connectionId, results );
+
+ if (type == TYPE_EXCHANGE ||
+ type == TYPE_TOPIC ||
+ type == TYPE_BINDING)
+ return Manageable::STATUS_NOT_IMPLEMENTED;
+
+ throw UnknownObjectType(type);
+}
+
+Manageable::status_t Broker::queryQueue( const std::string& name,
+ const std::string& userId,
+ const std::string& /*connectionId*/,
+ Variant::Map& results )
+{
+ (void) results;
+ if (acl) {
+ if (!acl->authorise(userId, acl::ACT_ACCESS, acl::OBJ_QUEUE, name, NULL) )
+ throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue query request from " << userId));
+ }
+
+ boost::shared_ptr<Queue> q(queues.find(name));
+ if (!q) {
+ QPID_LOG(error, "Query failed: queue not found, name=" << name);
+ return Manageable::STATUS_UNKNOWN_OBJECT;
+ }
+ return Manageable::STATUS_OK;;
+}
+
void Broker::setLogLevel(const std::string& level)
{
QPID_LOG(notice, "Changing log level to " << level);
@@ -723,7 +774,8 @@ void Broker::connect(
uint32_t Broker::queueMoveMessages(
const std::string& srcQueue,
const std::string& destQueue,
- uint32_t qty)
+ uint32_t qty,
+ const Variant::Map& filter)
{
Queue::shared_ptr src_queue = queues.find(srcQueue);
if (!src_queue)
@@ -732,7 +784,7 @@ uint32_t Broker::queueMoveMessages(
if (!dest_queue)
return 0;
- return src_queue->move(dest_queue, qty);
+ return src_queue->move(dest_queue, qty, &filter);
}
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h Tue Aug 16 00:37:33 2011
@@ -157,7 +157,12 @@ public:
const qpid::types::Variant::Map& properties, bool strict, const ConnectionState* context);
void deleteObject(const std::string& type, const std::string& name,
const qpid::types::Variant::Map& options, const ConnectionState* context);
-
+ Manageable::status_t queryObject(const std::string& type, const std::string& name,
+ qpid::types::Variant::Map& results, const ConnectionState* context);
+ Manageable::status_t queryQueue( const std::string& name,
+ const std::string& userId,
+ const std::string& connectionId,
+ qpid::types::Variant::Map& results);
boost::shared_ptr<sys::Poller> poller;
sys::Timer timer;
std::auto_ptr<sys::Timer> clusterTimer;
@@ -258,7 +263,8 @@ public:
*/
uint32_t queueMoveMessages( const std::string& srcQueue,
const std::string& destQueue,
- uint32_t qty);
+ uint32_t qty,
+ const qpid::types::Variant::Map& filter);
boost::shared_ptr<sys::ProtocolFactory> getProtocolFactory(const std::string& name = TCP_TRANSPORT) const;
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp Tue Aug 16 00:37:33 2011
@@ -44,6 +44,11 @@ bool LegacyLVQ::remove(const framing::Se
}
}
+bool LegacyLVQ::next(const QueuedMessage& message, QueuedMessage& next)
+{
+ return this->next(message.position, next);
+}
+
bool LegacyLVQ::next(const framing::SequenceNumber& position, QueuedMessage& message)
{
if (MessageMap::next(position, message)) {
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.h?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/LegacyLVQ.h Tue Aug 16 00:37:33 2011
@@ -42,6 +42,7 @@ class LegacyLVQ : public MessageMap
LegacyLVQ(const std::string& key, bool noBrowse = false, Broker* broker = 0);
bool remove(const framing::SequenceNumber&, QueuedMessage&);
bool next(const framing::SequenceNumber&, QueuedMessage&);
+ bool next(const QueuedMessage&, QueuedMessage&);
bool push(const QueuedMessage& added, QueuedMessage& removed);
void removeIf(Predicate);
void setNoBrowse(bool);
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.cpp?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.cpp Tue Aug 16 00:37:33 2011
@@ -74,6 +74,11 @@ bool MessageDeque::find(const framing::S
return find(position, message, false);
}
+bool MessageDeque::next(const QueuedMessage& message, QueuedMessage& next)
+{
+ return this->next(message.position, next);
+}
+
bool MessageDeque::next(const framing::SequenceNumber& position, QueuedMessage& message)
{
if (messages.empty()) {
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.h?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageDeque.h Tue Aug 16 00:37:33 2011
@@ -41,6 +41,7 @@ class MessageDeque : public Messages
bool remove(const framing::SequenceNumber&, QueuedMessage&);
bool find(const framing::SequenceNumber&, QueuedMessage&);
bool next(const framing::SequenceNumber&, QueuedMessage&);
+ bool next(const QueuedMessage&, QueuedMessage&);
QueuedMessage& front();
void pop();
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.cpp?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.cpp Tue Aug 16 00:37:33 2011
@@ -77,6 +77,11 @@ bool MessageMap::find(const framing::Seq
}
}
+bool MessageMap::next(const QueuedMessage& message, QueuedMessage& next)
+{
+ return this->next(message.position, next);
+}
+
bool MessageMap::next(const framing::SequenceNumber& position, QueuedMessage& message)
{
if (!messages.empty() && position < front().position) {
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.h?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageMap.h Tue Aug 16 00:37:33 2011
@@ -47,6 +47,7 @@ class MessageMap : public Messages
virtual bool remove(const framing::SequenceNumber&, QueuedMessage&);
bool find(const framing::SequenceNumber&, QueuedMessage&);
virtual bool next(const framing::SequenceNumber&, QueuedMessage&);
+ virtual bool next(const QueuedMessage&, QueuedMessage&);
QueuedMessage& front();
void pop();
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Messages.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Messages.h?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Messages.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Messages.h Tue Aug 16 00:37:33 2011
@@ -78,6 +78,14 @@ class Messages
virtual bool next(const framing::SequenceNumber&, QueuedMessage&) = 0;
/**
+ * Return the next message based on the supplied queued message.
+ * The next messages is passed back via the second parameter.
+ * @todo replace with queue iterator
+ * @return true if there is another message, false otherwise.
+ */
+ virtual bool next(const QueuedMessage&, QueuedMessage&) = 0;
+
+ /**
* Note: Caller is responsible for ensuring that there is a front
* (e.g. empty() returns false)
*
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.cpp?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.cpp Tue Aug 16 00:37:33 2011
@@ -88,6 +88,25 @@ bool PriorityQueue::find(const framing::
return find(position, message, false);
}
+bool PriorityQueue::next(const QueuedMessage& message, QueuedMessage& next)
+{
+ uint p = getPriorityLevel(message);
+ QueuedMessage match;
+ match.position = message.position+1;
+ Deque::iterator m = lower_bound(messages[p].begin(), messages[p].end(), match);
+ if (m != messages[p].end()) {
+ next = *m;
+ return true;
+ }
+ while (p-- > 0) {
+ if (!messages[p].empty()) {
+ next = messages[p].front();
+ return true;
+ }
+ }
+ return false;
+}
+
bool PriorityQueue::next(const framing::SequenceNumber& position, QueuedMessage& message)
{
QueuedMessage match;
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.h?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/PriorityQueue.h Tue Aug 16 00:37:33 2011
@@ -46,6 +46,7 @@ class PriorityQueue : public Messages
bool remove(const framing::SequenceNumber&, QueuedMessage&);
bool find(const framing::SequenceNumber&, QueuedMessage&);
bool next(const framing::SequenceNumber&, QueuedMessage&);
+ bool next(const QueuedMessage&, QueuedMessage&);
QueuedMessage& front();
void pop();
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.cpp?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.cpp Tue Aug 16 00:37:33 2011
@@ -42,6 +42,7 @@
#include "qpid/sys/ClusterSafe.h"
#include "qpid/sys/Monitor.h"
#include "qpid/sys/Time.h"
+#include "qpid/types/Variant.h"
#include "qmf/org/apache/qpid/broker/ArgsQueuePurge.h"
#include "qmf/org/apache/qpid/broker/ArgsQueueReroute.h"
@@ -111,8 +112,70 @@ class MessageAllocator
*/
virtual bool canAcquire( const std::string& consumer, const QueuedMessage& qm,
const Mutex::ScopedLock&);
+
+ /** hook to add any interesting management state to the status map (lock held) */
+ virtual void query(qpid::types::Variant::Map&, const Mutex::ScopedLock&) const {};
+
+ /** for move, purge, reroute - check if message matches against a filter,
+ * return true if message matches.
+ */
+ virtual bool match(const qpid::types::Variant::Map* filter,
+ const QueuedMessage& message) const;
+};
+
+
+
+class MessageGroupManager : public QueueObserver, public MessageAllocator
+{
+ const std::string groupIdHeader; // msg header holding group identifier
+ const unsigned int timestamp; // mark messages with timestamp if set
+
+ struct GroupState {
+ //const std::string group; // group identifier
+ //Consumer::shared_ptr owner; // consumer with outstanding acquired messages
+ std::string owner; // consumer with outstanding acquired messages
+ uint32_t acquired; // count of outstanding acquired messages
+ uint32_t total; // count of enqueued messages in this group
+ GroupState() : acquired(0), total(0) {}
+ };
+ typedef std::map<std::string, struct GroupState> GroupMap;
+ typedef std::set<std::string> Consumers;
+
+ GroupMap messageGroups;
+ Consumers consumers;
+
+ static const std::string qpidMessageGroupKey;
+ static const std::string qpidMessageGroupTimestamp;
+ static const std::string qpidMessageGroupDefault;
+ static const std::string qpidMessageGroupFilter; // key for move/purge filter map
+
+ const std::string getGroupId( const QueuedMessage& qm ) const;
+
+ public:
+
+ static boost::shared_ptr<MessageGroupManager> create( Queue *q, const qpid::framing::FieldTable& settings );
+
+ MessageGroupManager(const std::string& header, Queue *q, unsigned int _timestamp=0 )
+ : QueueObserver(), MessageAllocator(q), groupIdHeader( header ), timestamp(_timestamp) {}
+ void enqueued( const QueuedMessage& qm );
+ void acquired( const QueuedMessage& qm );
+ void requeued( const QueuedMessage& qm );
+ void dequeued( const QueuedMessage& qm );
+ void consumerAdded( const Consumer& );
+ void consumerRemoved( const Consumer& );
+ bool nextMessage( Consumer::shared_ptr c, QueuedMessage& next,
+ const Mutex::ScopedLock&);
+ bool canAcquire(const std::string& consumer, const QueuedMessage& msg,
+ const Mutex::ScopedLock&);
+ void query(qpid::types::Variant::Map&, const Mutex::ScopedLock&) const;
+ bool match(const qpid::types::Variant::Map*, const QueuedMessage&) const;
};
+const std::string MessageGroupManager::qpidMessageGroupKey("qpid.group_header_key");
+const std::string MessageGroupManager::qpidMessageGroupTimestamp("qpid.group_timestamp");
+const std::string MessageGroupManager::qpidMessageGroupDefault("qpid.no_group"); /** @todo KAG: make configurable in Broker options */
+const std::string MessageGroupManager::qpidMessageGroupFilter("qpid.group_id");
+
}}
Queue::Queue(const string& _name, bool _autodelete,
@@ -140,7 +203,8 @@ Queue::Queue(const string& _name, bool _
deleted(false),
barrier(*this),
autoDeleteTimeout(0),
- allocator(new MessageAllocator( this )) // KAG TODO: FIX!!
+ allocator(new MessageAllocator( this )),
+ type(FIFO)
{
if (parent != 0 && broker != 0) {
ManagementAgent* agent = broker->getManagementAgent();
@@ -275,7 +339,7 @@ bool Queue::acquireMessageAt(const Seque
}
}
-bool Queue::acquire(const QueuedMessage& msg, const std::string& consumer)
+bool Queue::acquire(const QueuedMessage& msg, const std::string& consumer)
{
Mutex::ScopedLock locker(messageLock);
assertClusterSafe();
@@ -535,6 +599,29 @@ void Queue::purgeExpired(qpid::sys::Dura
}
}
+
+namespace {
+ // for use with purge/move below - collect messages that match a given filter
+ struct Collector {
+ const uint32_t maxMatches;
+ const qpid::types::Variant::Map *filter;
+ std::deque<QueuedMessage> matches;
+ boost::shared_ptr<MessageAllocator> allocator;
+ Collector(boost::shared_ptr<MessageAllocator> a, uint32_t m,
+ const qpid::types::Variant::Map *f)
+ : maxMatches(m), filter(f), allocator(a) {}
+ void operator() (QueuedMessage& qm)
+ {
+ if (maxMatches == 0 || matches.size() < maxMatches) {
+ if (allocator->match( filter, qm )) {
+ matches.push_back(qm);
+ }
+ }
+ }
+ };
+}
+
+
/**
* purge - for purging all or some messages on a queue
* depending on the purge_request
@@ -546,52 +633,61 @@ void Queue::purgeExpired(qpid::sys::Dura
* The dest exchange may be supplied to re-route messages through the exchange.
* It is safe to re-route messages such that they arrive back on the same queue,
* even if the queue is ordered by priority.
+ *
+ * An optional filter can be supplied that will be applied against each message. The
+ * message is purged only if the filter matches. See MessageAllocator for more detail.
*/
-uint32_t Queue::purge(const uint32_t purge_request, boost::shared_ptr<Exchange> dest)
+uint32_t Queue::purge(const uint32_t purge_request, boost::shared_ptr<Exchange> dest,
+ const qpid::types::Variant::Map *filter)
{
+ Collector c(allocator, purge_request, filter);
+
Mutex::ScopedLock locker(messageLock);
- uint32_t purge_count = purge_request; // only comes into play if >0
- std::deque<DeliverableMessage> rerouteQueue;
+ messages->foreach( boost::bind<void>(boost::ref(c), _1) );
- uint32_t count = 0;
- // Either purge them all or just the some (purge_count) while the queue isn't empty.
- while((!purge_request || purge_count--) && !messages->empty()) {
- if (dest.get()) {
- //
- // If there is a destination exchange, stage the messages onto a reroute queue
- // so they don't wind up getting purged more than once.
- //
- DeliverableMessage msg(messages->front().payload);
- rerouteQueue.push_back(msg);
- }
- popAndDequeue();
- count++;
- }
+ uint32_t count = c.matches.size();
- //
- // Re-route purged messages into the destination exchange. Note that there's no need
- // to test dest.get() here because if it is NULL, the rerouteQueue will be empty.
- //
- while (!rerouteQueue.empty()) {
- DeliverableMessage msg(rerouteQueue.front());
- rerouteQueue.pop_front();
- dest->routeWithAlternate(msg);
+ // first remove all matches
+ for (std::deque<QueuedMessage>::iterator qmsg = c.matches.begin();
+ qmsg != c.matches.end(); qmsg++) {
+ /** @todo KAG: need a direct remove method here */
+ bool ok = acquire(qmsg->position, *qmsg);
+ (void) ok; assert(ok);
+ dequeue(0, *qmsg);
+ }
+
+ // now reroute if necessary
+ if (dest.get()) {
+ while (!c.matches.empty()) {
+ QueuedMessage msg = c.matches.front();
+ c.matches.pop_front();
+ assert(msg.payload);
+ DeliverableMessage dmsg(msg.payload);
+ dest->routeWithAlternate(dmsg);
+ }
}
-
return count;
}
-uint32_t Queue::move(const Queue::shared_ptr destq, uint32_t qty) {
+uint32_t Queue::move(const Queue::shared_ptr destq, uint32_t qty,
+ const qpid::types::Variant::Map *filter)
+{
+ Collector c(allocator, qty, filter);
+
Mutex::ScopedLock locker(messageLock);
- uint32_t move_count = qty; // only comes into play if qty >0
- uint32_t count = 0; // count how many were moved for returning
+ messages->foreach( boost::bind<void>(boost::ref(c), _1) );
- while((!qty || move_count--) && !messages->empty()) {
- QueuedMessage qmsg = messages->front();
- boost::intrusive_ptr<Message> msg = qmsg.payload;
- destq->deliver(msg); // deliver message to the destination queue
- popAndDequeue();
- count++;
+ uint32_t count = c.matches.size();
+
+ while (!c.matches.empty()) {
+ QueuedMessage qmsg = c.matches.front();
+ c.matches.pop_front();
+ /** @todo KAG: need a direct remove method here */
+ bool ok = acquire(qmsg.position, qmsg);
+ (void) ok; assert(ok);
+ dequeue(0, qmsg);
+ assert(qmsg.payload);
+ destq->deliver(qmsg.payload);
}
return count;
}
@@ -614,6 +710,7 @@ bool Queue::acquire(const qpid::framing:
{
if (messages->remove(position, msg)) {
acquired( msg );
+ ++dequeueSincePurge;
return true;
}
return false;
@@ -912,17 +1009,29 @@ void Queue::configureImpl(const FieldTab
if (lvqKey.size()) {
QPID_LOG(debug, "Configured queue " << getName() << " as Last Value Queue with key " << lvqKey);
messages = std::auto_ptr<Messages>(new MessageMap(lvqKey));
+ type = LVQ;
} else if (_settings.get(qpidLastValueQueueNoBrowse)) {
QPID_LOG(debug, "Configured queue " << getName() << " as Legacy Last Value Queue with 'no-browse' on");
messages = LegacyLVQ::updateOrReplace(messages, qpidVQMatchProperty, true, broker);
+ type = LVQ;
} else if (_settings.get(qpidLastValueQueue)) {
QPID_LOG(debug, "Configured queue " << getName() << " as Legacy Last Value Queue");
messages = LegacyLVQ::updateOrReplace(messages, qpidVQMatchProperty, false, broker);
+ type = LVQ;
} else {
std::auto_ptr<Messages> m = Fairshare::create(_settings);
if (m.get()) {
messages = m;
QPID_LOG(debug, "Configured queue " << getName() << " as priority queue.");
+ type = PRIORITY;
+ }
+ }
+
+ { // override default message allocator if message groups configured.
+ boost::shared_ptr<MessageAllocator> ma = boost::static_pointer_cast<MessageAllocator>(MessageGroupManager::create( this, _settings ));
+ if (ma) {
+ allocator = ma;
+ type = GROUP;
}
}
@@ -1181,7 +1290,7 @@ Manageable::status_t Queue::ManagementMe
case _qmf::Queue::METHOD_PURGE :
{
_qmf::ArgsQueuePurge& purgeArgs = (_qmf::ArgsQueuePurge&) args;
- purge(purgeArgs.i_request);
+ purge(purgeArgs.i_request, boost::shared_ptr<Exchange>(), &purgeArgs.i_filter);
status = Manageable::STATUS_OK;
}
break;
@@ -1202,7 +1311,7 @@ Manageable::status_t Queue::ManagementMe
}
}
- purge(rerouteArgs.i_request, dest);
+ purge(rerouteArgs.i_request, dest, &rerouteArgs.i_filter);
status = Manageable::STATUS_OK;
}
break;
@@ -1211,6 +1320,14 @@ Manageable::status_t Queue::ManagementMe
return status;
}
+
+void Queue::query(qpid::types::Variant::Map& results) const
+{
+ Mutex::ScopedLock locker(messageLock);
+ /** @todo add any interesting queue state into results */
+ if (allocator) allocator->query( results, messageLock );
+}
+
void Queue::setPosition(SequenceNumber n) {
Mutex::ScopedLock locker(messageLock);
sequence = n;
@@ -1357,145 +1474,258 @@ void Queue::UsageBarrier::destroy()
// KAG TBD: flesh out...
-class MessageGroupManager : public QueueObserver, public MessageAllocator
-{
- const std::string groupIdHeader; // msg header holding group identifier
- struct GroupState {
- const std::string group; // group identifier
- //Consumer::shared_ptr owner; // consumer with outstanding acquired messages
- std::string owner; // consumer with outstanding acquired messages
- uint32_t acquired; // count of outstanding acquired messages
- uint32_t total; // count of enqueued messages in this group
- GroupState() : acquired(0), total(0) {}
- };
- std::map<std::string, struct GroupState> messageGroups;
- std::set<std::string> consumers;
-
- public:
-
- MessageGroupManager(const std::string& header, Queue *q )
- : QueueObserver(), MessageAllocator(q), groupIdHeader( header ) {}
- void enqueued( const QueuedMessage& qm );
- void removed( const QueuedMessage& qm );
- void requeued( const QueuedMessage& qm );
- void dequeued( const QueuedMessage& qm );
- void consumerAdded( const Consumer& );
- void consumerRemoved( const Consumer& );
- bool nextMessage( Consumer::shared_ptr c, QueuedMessage& next,
- const Mutex::ScopedLock&);
- bool canAcquire(const std::string& consumer, const QueuedMessage& msg,
- const Mutex::ScopedLock&);
-};
-namespace {
- const std::string NO_GROUP("");
- const std::string getGroupId( const QueuedMessage& qm, const std::string& key )
- {
- const qpid::framing::FieldTable* headers = qm.payload->getApplicationHeaders();
- if (!headers) return NO_GROUP;
- return headers->getAsString(key);
- }
+const std::string MessageGroupManager::getGroupId( const QueuedMessage& qm ) const
+{
+ const qpid::framing::FieldTable* headers = qm.payload->getApplicationHeaders();
+ if (!headers) return qpidMessageGroupDefault;
+ FieldTable::ValuePtr id = headers->get( groupIdHeader );
+ if (!id || !id->convertsTo<std::string>()) return qpidMessageGroupDefault;
+ return id->get<std::string>();
}
void MessageGroupManager::enqueued( const QueuedMessage& qm )
{
- std::string group( getGroupId(qm, groupIdHeader) );
- messageGroups[group].total++;
+ std::string group( getGroupId(qm) );
+ uint32_t total = ++messageGroups[group].total;
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": added message to group id=" << group << " total=" << total );
}
-void MessageGroupManager::removed( const QueuedMessage& qm )
+void MessageGroupManager::acquired( const QueuedMessage& qm )
{
- std::string group( getGroupId(qm, groupIdHeader) );
- std::map<std::string, struct GroupState>::iterator gs = messageGroups.find( group );
+ std::string group( getGroupId(qm) );
+ GroupMap::iterator gs = messageGroups.find( group );
assert( gs != messageGroups.end() );
- gs->second.acquired += 1;
+ GroupState& state( gs->second );
+ state.acquired += 1;
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": acquired message in group id=" << group << " acquired=" << state.acquired );
}
void MessageGroupManager::requeued( const QueuedMessage& qm )
{
- std::string group( getGroupId(qm, groupIdHeader) );
- std::map<std::string, struct GroupState>::iterator gs = messageGroups.find( group );
+ std::string group( getGroupId(qm) );
+ GroupMap::iterator gs = messageGroups.find( group );
assert( gs != messageGroups.end() );
GroupState& state( gs->second );
assert( state.acquired != 0 );
state.acquired -= 1;
if (state.acquired == 0 && !state.owner.empty()) {
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": consumer name=" << state.owner << " released group id=" << gs->first);
state.owner.clear(); // KAG TODO: need to invalidate consumer's positions?
}
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": requeued message to group id=" << group << " acquired=" << state.acquired );
}
void MessageGroupManager::dequeued( const QueuedMessage& qm )
{
- std::string group( getGroupId(qm, groupIdHeader) );
- std::map<std::string, struct GroupState>::iterator gs = messageGroups.find( group );
+ std::string group( getGroupId(qm) );
+ GroupMap::iterator gs = messageGroups.find( group );
assert( gs != messageGroups.end() );
GroupState& state( gs->second );
assert( state.total != 0 );
- state.total -= 1;
+ uint32_t total = state.total -= 1;
assert( state.acquired != 0 );
state.acquired -= 1;
- if (state.total == 0) messageGroups.erase( gs );
- else if (state.acquired == 0 && !state.owner.empty()) {
- state.owner.clear(); // KAG TODO: need to invalidate consumer's positions?
+ if (state.total == 0) {
+ QPID_LOG( trace, "group queue " << queue->getName() << ": deleting group id=" << gs->first);
+ messageGroups.erase( gs );
+ } else {
+ if (state.acquired == 0 && !state.owner.empty()) {
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": consumer name=" << state.owner << " released group id=" << gs->first);
+ state.owner.clear(); // KAG TODO: need to invalidate consumer's positions?
+ }
}
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": dequeued message from group id=" << group << " total=" << total );
}
void MessageGroupManager::consumerAdded( const Consumer& c )
{
- bool unique = consumers.insert( c.getName() ).second;
+ const std::string& name(c.getName());
+ bool unique = consumers.insert( name ).second;
(void) unique; assert( unique );
+ QPID_LOG( trace, "group queue " << queue->getName() << ": added consumer name=" << name );
}
void MessageGroupManager::consumerRemoved( const Consumer& c )
{
- size_t count = consumers.erase( c.getName() );
+ const std::string& name(c.getName());
+ size_t count = consumers.erase( name );
(void) count; assert( count == 1 );
bool needReset = false;
- for (std::map<std::string, struct GroupState>::iterator gs = messageGroups.begin();
+ for (GroupMap::iterator gs = messageGroups.begin();
gs != messageGroups.end(); ++gs) {
GroupState& state( gs->second );
- if (state.owner == c.getName()) {
+ if (state.owner == name) {
state.owner.clear();
needReset = true;
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": consumer name=" << name << " released group id=" << gs->first);
}
}
if (needReset) {
// KAG TODO: How do I invalidate all consumers that need invalidating????
}
+ QPID_LOG( trace, "group queue " << queue->getName() << ": removed consumer name=" << name );
}
bool MessageGroupManager::nextMessage( Consumer::shared_ptr c, QueuedMessage& next,
- const Mutex::ScopedLock& l)
+ const Mutex::ScopedLock& )
{
- // KAG TODO: FIX!!!
- return MessageAllocator::nextMessage( c, next, l );
+ Messages& messages(queue->getMessages());
+
+ if (messages.empty())
+ return false;
+
+ if (c->preAcquires()) { // not browsing
+ next = messages.front();
+ QueuedMessage current;
+ do {
+ current = next;
+ /** @todo KAG: horrifingly suboptimal - optimize */
+ std::string group( getGroupId( current ) );
+ GroupMap::iterator gs = messageGroups.find( group ); /** @todo need to cache this somehow */
+ assert( gs != messageGroups.end() );
+ GroupState& state( gs->second );
+ if (state.owner.empty()) {
+ state.owner = c->getName();
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": consumer name=" << c->getName() << " has acquired group id=" << group);
+ return true;
+ }
+ if (state.owner == c->getName()) {
+ return true;
+ }
+ } while (messages.next( current, next )); /** @todo: .next() is a linear search from front - optimize */
+ return false;
+ } else if (messages.next(c->position, next))
+ return true;
+ return false;
}
bool MessageGroupManager::canAcquire(const std::string& consumer, const QueuedMessage& qm,
const Mutex::ScopedLock&)
{
- std::string group( getGroupId(qm, groupIdHeader) );
- std::map<std::string, struct GroupState>::iterator gs = messageGroups.find( group );
+ std::string group( getGroupId(qm) );
+ GroupMap::iterator gs = messageGroups.find( group );
assert( gs != messageGroups.end() );
GroupState& state( gs->second );
if (state.owner.empty()) {
state.owner = consumer;
+ QPID_LOG( trace, "group queue " << queue->getName() <<
+ ": consumer name=" << consumer << " has acquired group id=" << gs->first);
return true;
}
return state.owner == consumer;
}
+namespace {
+ const std::string GroupQueryKey("qpid.message_group_queue");
+ const std::string GroupHeaderKey("group_header_key");
+ const std::string GroupStateKey("group_state");
+ const std::string GroupIdKey("group_id");
+ const std::string GroupMsgCount("msg_count");
+ const std::string GroupTimestamp("timestamp");
+ const std::string GroupConsumer("consumer");
+}
+
+void MessageGroupManager::query(qpid::types::Variant::Map& status,
+ const Mutex::ScopedLock&) const
+{
+ /** Add a description of the current state of the message groups for this queue.
+ FORMAT:
+ { "qpid.message_group_queue":
+ { "group_header_key" : "<KEY>",
+ "group_state" :
+ [ { "group_id" : "<name>",
+ "msg_count" : <int>,
+ "timestamp" : <absTime>,
+ "consumer" : <consumer name> },
+ {...} // one for each known group
+ ]
+ }
+ }
+ **/
+
+ assert(status.find(GroupQueryKey) == status.end());
+ qpid::types::Variant::Map state;
+ qpid::types::Variant::List groups;
+
+ state[GroupHeaderKey] = groupIdHeader;
+ for (GroupMap::const_iterator g = messageGroups.begin();
+ g != messageGroups.end(); ++g) {
+ qpid::types::Variant::Map info;
+ info[GroupIdKey] = g->first;
+ info[GroupMsgCount] = g->second.total;
+ info[GroupTimestamp] = 0; /** @todo KAG - NEED HEAD MSG TIMESTAMP */
+ info[GroupConsumer] = g->second.owner;
+ groups.push_back(info);
+ }
+ state[GroupStateKey] = groups;
+ status[GroupQueryKey] = state;
+}
+
+bool MessageGroupManager::match(const qpid::types::Variant::Map* filter,
+ const QueuedMessage& message) const
+{
+ if (!filter) return true;
+ qpid::types::Variant::Map::const_iterator i = filter->find( qpidMessageGroupFilter );
+ if (i == filter->end()) return true;
+ if (i->second.asString() == getGroupId(message)) return true;
+ return false;
+}
+
+boost::shared_ptr<MessageGroupManager> MessageGroupManager::create( Queue *q,
+ const qpid::framing::FieldTable& settings )
+{
+ boost::shared_ptr<MessageGroupManager> empty;
+
+ if (settings.isSet(qpidMessageGroupKey)) {
+
+ Queue::Disposition qt = q->getDisposition();
+
+ if (qt == Queue::LVQ) {
+ QPID_LOG( error, "Message Groups cannot be enabled on LVQ Queues, queue=" << q->getName());
+ return empty;
+ }
+ if (qt == Queue::PRIORITY) {
+ QPID_LOG( error, "Message Groups cannot be enabled for Priority Queues, queue=" << q->getName());
+ return empty;
+ }
+ std::string headerKey = settings.getAsString(qpidMessageGroupKey);
+ if (headerKey.empty()) {
+ QPID_LOG( error, "A Message Group header key must be configured, queue=" << q->getName());
+ return empty;
+ }
+ unsigned int timestamp = settings.getAsInt(qpidMessageGroupTimestamp);
+
+ boost::shared_ptr<MessageGroupManager> manager( new MessageGroupManager( headerKey, q, timestamp ) );
+
+ q->addObserver( boost::static_pointer_cast<QueueObserver>(manager) );
+
+ QPID_LOG( debug, "Configured Queue '" << q->getName() <<
+ "' for message grouping using header key '" << headerKey << "'" <<
+ " (timestamp=" << timestamp << ")");
+ return manager;
+ }
+ return empty;
+}
@@ -1526,5 +1756,13 @@ bool MessageAllocator::canAcquire(const
}
+// default match - ignore filter and always match.
+bool MessageAllocator::match(const qpid::types::Variant::Map*,
+ const QueuedMessage&) const
+{
+ return true;
+}
+
+
Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.h?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Queue.h Tue Aug 16 00:37:33 2011
@@ -60,7 +60,7 @@ class QueueEvents;
class QueueRegistry;
class TransactionContext;
class MessageAllocator;
-
+
/**
* The brokers representation of an amqp queue. Messages are
* delivered to a queue from where they can be dispatched to
@@ -70,6 +70,10 @@ class MessageAllocator;
class Queue : public boost::enable_shared_from_this<Queue>,
public PersistableQueue, public management::Manageable {
+ public:
+ enum Disposition {FIFO, LVQ, PRIORITY, GROUP};
+
+ private:
struct UsageBarrier
{
Queue& parent;
@@ -129,7 +133,9 @@ class Queue : public boost::enable_share
UsageBarrier barrier;
int autoDeleteTimeout;
boost::intrusive_ptr<qpid::sys::TimerTask> autoDeleteTask;
- std::auto_ptr<MessageAllocator> allocator;
+ boost::shared_ptr<MessageAllocator> allocator;
+ Disposition type;
+
void push(boost::intrusive_ptr<Message>& msg, bool isRecovery=false);
void setPolicy(std::auto_ptr<QueuePolicy> policy);
@@ -262,11 +268,14 @@ class Queue : public boost::enable_share
bool exclusive = false);
QPID_BROKER_EXTERN void cancel(Consumer::shared_ptr c);
- uint32_t purge(const uint32_t purge_request=0, boost::shared_ptr<Exchange> dest=boost::shared_ptr<Exchange>()); //defaults to all messages
+ uint32_t purge(const uint32_t purge_request=0, //defaults to all messages
+ boost::shared_ptr<Exchange> dest=boost::shared_ptr<Exchange>(),
+ const ::qpid::types::Variant::Map *filter=0);
QPID_BROKER_EXTERN void purgeExpired(sys::Duration);
//move qty # of messages to destination Queue destq
- uint32_t move(const Queue::shared_ptr destq, uint32_t qty);
+ uint32_t move(const Queue::shared_ptr destq, uint32_t qty,
+ const qpid::types::Variant::Map *filter=0);
QPID_BROKER_EXTERN uint32_t getMessageCount() const;
QPID_BROKER_EXTERN uint32_t getEnqueueCompleteMessageCount() const;
@@ -353,6 +362,7 @@ class Queue : public boost::enable_share
management::ManagementObject* GetManagementObject (void) const;
management::Manageable::status_t
ManagementMethod (uint32_t methodId, management::Args& args, std::string& text);
+ void query(::qpid::types::Variant::Map&) const;
/** Apply f to each Message on the queue. */
template <class F> void eachMessage(F f) {
@@ -402,6 +412,8 @@ class Queue : public boost::enable_share
uint32_t getDequeueSincePurge() { return dequeueSincePurge.get(); }
void setDequeueSincePurge(uint32_t value);
+
+ Disposition getDisposition() const { return type; }
};
}
}
Propchange: qpid/branches/qpid-3346/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/src/qpid/cluster/UpdateClient.cpp:1061302-1072333
+/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/src/qpid/cluster/UpdateClient.h
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/src/qpid/cluster/UpdateClient.h:1061302-1072333
+/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/src/qpid/management/ManagementAgent.cpp
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/src/qpid/management/ManagementAgent.cpp:1061302-1072333
+/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp:1156188
Propchange: qpid/branches/qpid-3346/qpid/cpp/src/qpid/management/ManagementAgent.h
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/src/qpid/management/ManagementAgent.h:1061302-1072333
+/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h:1156188
Modified: qpid/branches/qpid-3346/qpid/cpp/src/tests/QueueTest.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/tests/QueueTest.cpp?rev=1158073&r1=1158072&r2=1158073&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/tests/QueueTest.cpp Tue Aug 16 00:37:33 2011
@@ -56,12 +56,12 @@ class TestConsumer : public virtual Cons
public:
typedef boost::shared_ptr<TestConsumer> shared_ptr;
- intrusive_ptr<Message> last;
+ QueuedMessage last;
bool received;
- TestConsumer(bool acquire = true):Consumer("test", acquire), received(false) {};
+ TestConsumer(std::string name="test", bool acquire = true):Consumer(name, acquire), received(false) {};
virtual bool deliver(QueuedMessage& msg){
- last = msg.payload;
+ last = msg;
received = true;
return true;
};
@@ -148,16 +148,16 @@ QPID_AUTO_TEST_CASE(testConsumers){
queue->deliver(msg1);
BOOST_CHECK(queue->dispatch(c1));
- BOOST_CHECK_EQUAL(msg1.get(), c1->last.get());
+ BOOST_CHECK_EQUAL(msg1.get(), c1->last.payload.get());
queue->deliver(msg2);
BOOST_CHECK(queue->dispatch(c2));
- BOOST_CHECK_EQUAL(msg2.get(), c2->last.get());
+ BOOST_CHECK_EQUAL(msg2.get(), c2->last.payload.get());
c1->received = false;
queue->deliver(msg3);
BOOST_CHECK(queue->dispatch(c1));
- BOOST_CHECK_EQUAL(msg3.get(), c1->last.get());
+ BOOST_CHECK_EQUAL(msg3.get(), c1->last.payload.get());
//Test cancellation:
queue->cancel(c1);
@@ -213,7 +213,7 @@ QPID_AUTO_TEST_CASE(testDequeue){
if (!consumer->received)
sleep(2);
- BOOST_CHECK_EQUAL(msg3.get(), consumer->last.get());
+ BOOST_CHECK_EQUAL(msg3.get(), consumer->last.payload.get());
BOOST_CHECK_EQUAL(uint32_t(0), queue->getMessageCount());
received = queue->get().payload;
@@ -297,14 +297,14 @@ QPID_AUTO_TEST_CASE(testSeek){
queue->deliver(msg2);
queue->deliver(msg3);
- TestConsumer::shared_ptr consumer(new TestConsumer(false));
+ TestConsumer::shared_ptr consumer(new TestConsumer("test", false));
SequenceNumber seq(2);
consumer->position = seq;
QueuedMessage qm;
queue->dispatch(consumer);
- BOOST_CHECK_EQUAL(msg3.get(), consumer->last.get());
+ BOOST_CHECK_EQUAL(msg3.get(), consumer->last.payload.get());
queue->dispatch(consumer);
queue->dispatch(consumer); // make sure over-run is safe
@@ -571,7 +571,7 @@ QPID_AUTO_TEST_CASE(testLVQAcquire){
// set mode to no browse and check
args.setOrdering(client::LVQ_NO_BROWSE);
queue->configure(args);
- TestConsumer::shared_ptr c1(new TestConsumer(false));
+ TestConsumer::shared_ptr c1(new TestConsumer("test", false));
queue->dispatch(c1);
queue->dispatch(c1);
@@ -705,6 +705,67 @@ QPID_AUTO_TEST_CASE(testQueueCleaner) {
BOOST_CHECK_EQUAL(queue->getMessageCount(), 0u);
}
+QPID_AUTO_TEST_CASE(testGroupsMultiConsumer) {
+ FieldTable args;
+ Queue::shared_ptr queue(new Queue("my_queue", true));
+ args.setString("qpid.group_header_key", "GROUP-ID");
+ queue->configure(args);
+
+ for (int i = 0; i < 3; ++i) {
+ intrusive_ptr<Message> msg1 = create_message("e", "A");
+ msg1->getProperties<MessageProperties>()->getApplicationHeaders().setString("GROUP-ID","a");
+ queue->deliver(msg1);
+
+ intrusive_ptr<Message> msg2 = create_message("e", "A");
+ msg2->getProperties<MessageProperties>()->getApplicationHeaders().setString("GROUP-ID","b");
+ queue->deliver(msg2);
+ }
+
+ BOOST_CHECK_EQUAL(uint32_t(6), queue->getMessageCount());
+
+ TestConsumer::shared_ptr c1(new TestConsumer("C1"));
+ TestConsumer::shared_ptr c2(new TestConsumer("C2"));
+
+ queue->consume(c1);
+ queue->consume(c2);
+
+ std::deque<QueuedMessage> dequeMe;
+
+ queue->dispatch(c1); // now owns group "a"
+ dequeMe.push_back(c1->last);
+ queue->dispatch(c2); // now owns group "b"
+ dequeMe.push_back(c2->last);
+
+ queue->dispatch(c2); // should skip next "a", get last "b"
+ std::string group = c2->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsString("GROUP-ID");
+ dequeMe.push_back(c2->last);
+ BOOST_CHECK_EQUAL( group, std::string("b") );
+
+ queue->dispatch(c1); // should get last "a"
+ group = c1->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsString("GROUP-ID");
+ dequeMe.push_back(c1->last);
+ BOOST_CHECK_EQUAL( group, std::string("a") );
+
+ // now "free up" the groups
+ while (!dequeMe.empty()) {
+ queue->dequeue( 0, dequeMe.front() );
+ dequeMe.pop_front();
+ }
+
+ // now c2 should be able to acquire group "a", and c1 group "b"
+
+ queue->dispatch(c2);
+ group = c2->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsString("GROUP-ID");
+ BOOST_CHECK_EQUAL( group, std::string("a") );
+
+ queue->dispatch(c1);
+ group = c1->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsString("GROUP-ID");
+ BOOST_CHECK_EQUAL( group, std::string("b") );
+
+ queue->cancel(c1);
+ queue->cancel(c2);
+}
+
QPID_AUTO_TEST_CASE(testMultiQueueLastNode){
TestMessageStoreOC testStore;
Propchange: qpid/branches/qpid-3346/qpid/cpp/src/tests/cluster_tests.py
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/src/tests/cluster_tests.py:1061302-1072333
+/qpid/trunk/qpid/cpp/src/tests/cluster_tests.py:1156188
Propchange: qpid/branches/qpid-3346/qpid/dotnet/build-msbuild.bat
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/dotnet/build-msbuild.bat:1061302-1072333
+/qpid/trunk/qpid/dotnet/build-msbuild.bat:1156188
Propchange: qpid/branches/qpid-3346/qpid/dotnet/build-nant-release
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/dotnet/build-nant-release:1061302-1072333
+/qpid/trunk/qpid/dotnet/build-nant-release:1156188
Propchange: qpid/branches/qpid-3346/qpid/dotnet/build-nant.bat
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1 +1,2 @@
/qpid/branches/qpid-2935/qpid/dotnet/build-nant.bat:1061302-1072333
+/qpid/trunk/qpid/dotnet/build-nant.bat:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -4,3 +4,4 @@
/qpid/branches/java-network-refactor/qpid/java:805429-821809
/qpid/branches/qpid-2935/qpid/java:1061302-1072333
/qpid/trunk/qpid:796646-796653
+/qpid/trunk/qpid/java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/broker/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,4 +3,4 @@
/qpid/branches/java-network-refactor/qpid/java/broker:805429-821809
/qpid/branches/jmx_mc_gsoc09/qpid/java/broker:787599
/qpid/branches/qpid-2935/qpid/java/broker:1061302-1072333
-/qpid/trunk/qpid/java/broker:742626,743015,743028-743029,743304,743306,743311,743357,744113,747363,747367,747369-747370,747376,747783,747868-747870,747875,748561,748591,748641,748680,748686,749149,749282,749285,749315,749340,749572,753219-753220,753253,754934,754958,755256,757258,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/broker:742626,743015,743028-743029,743304,743306,743311,743357,744113,747363,747367,747369-747370,747376,747783,747868-747870,747875,748561,748591,748641,748680,748686,749149,749282,749285,749315,749340,749572,753219-753220,753253,754934,754958,755256,757258,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790,1156188
Propchange: qpid/branches/qpid-3346/qpid/java/broker/bin/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,4 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/broker/bin:795950-829653
/qpid/branches/java-network-refactor/qpid/java/broker/bin:805429-821809
/qpid/branches/qpid-2935/qpid/java/broker/bin:1061302-1072333
-/qpid/trunk/qpid/java/broker/bin:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/broker/bin:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790,1156188
Propchange: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -4,4 +4,4 @@
/qpid/branches/java-network-refactor/qpid/java/broker/src/main/java/org/apache/qpid/server/management:805429-821809
/qpid/branches/jmx_mc_gsoc09/qpid/java/broker/src/main/java/org/apache/qpid/server/management:787599
/qpid/branches/qpid-2935/qpid/java/broker/src/main/java/org/apache/qpid/server/management:1061302-1072333
-/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790,1156188
Propchange: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -4,4 +4,4 @@
/qpid/branches/java-network-refactor/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:805429-821809
/qpid/branches/jmx_mc_gsoc09/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:787599
/qpid/branches/qpid-2935/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:1061302-1072333
-/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:753219-753220,753253,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:753219-753220,753253,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790,1156188
Propchange: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1,2 +1,3 @@
/qpid/branches/0.5.x-dev/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:930288
/qpid/branches/qpid-2935/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:1061302-1072333
+/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ConfigurationManagement.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ConfigurationManagement.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ConfigurationManagement.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/management/ConfigurationManagement.java:742626,743015,743028-743029,743304,743306,743311,743357,744113,747363,747367,747369-747370,747376,747783,747868-747870,747875,748591,748641,748680,748686,749149,749282,749285,749315,749340,749572,753219-753220,753253,754934,754958,755256,757258,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ConfigurationManagement.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/LoggingManagement.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/LoggingManagement.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/LoggingManagement.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/management/LoggingManagement.java:742626,743015,743028-743029,743304,743306,743311,743357,744113,747363,747367,747369-747370,747376,747783,747868-747869,747875,748561,748591,748641,748680,748686,749149,749282,749285,749315,749340,749572,753219-753220,753253,754934,754958,755256,757258,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/LoggingManagement.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management/ManagedBroker.java:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ManagedConnection.java:742626,743015,743028-743029,743304,743306,743311,743357,744113,747363,747367,747369-747370,747376,747783,747868-747870,747875,748561,748591,748641,748680,748686,749149,749282,749285,749315,749340,749572,753219-753220,753253,754934,754958,755256,757258,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedExchange.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedExchange.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedExchange.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/ManagedExchange.java:742626,743015,743028-743029,743304,743306,743311,743357,744113,747363,747367,747369-747370,747376,747783,747868-747870,747875,748561,748591,748641,748680,748686,749149,749282,749285,749315,749340,749572,753219-753220,753253,754934,754958,755256,757258,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedExchange.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ManagedQueue.java:753219-753220,753253,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/UserManagement.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/UserManagement.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/UserManagement.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/management/UserManagement.java:742626,743015,743028-743029,743304,743306,743311,743357,744113,747363,747367,747369-747370,747376,747783,747868-747870,747875,748561,748591,748641,748680,748686,749149,749282,749285,749315,749340,749572,753219-753220,753253,754934,754958,755256,757258,757270,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/UserManagement.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanAttribute.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanAttribute.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanAttribute.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanAttribute.java:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanAttribute.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanConstructor.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanConstructor.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanConstructor.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanConstructor.java:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanConstructor.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanDescription.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanDescription.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanDescription.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanDescription.java:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanDescription.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperation.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperation.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperation.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanOperation.java:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperation.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperationParameter.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,3 +3,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperationParameter.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperationParameter.java:1061302-1072333
/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanOperationParameter.java:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790
+/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperationParameter.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-network-refactor/qpid/java/management/eclipse-plugin/src:805429-821809
/qpid/branches/jmx_mc_gsoc09/qpid/java/management/eclipse-plugin/src:788755
/qpid/branches/qpid-2935/qpid/java/management/eclipse-plugin/src:1061302-1072333
+/qpid/trunk/qpid/java/management/eclipse-plugin/src:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ConnectionTypeTabControl.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ConnectionTypeTabControl.java:795950-829653
/qpid/branches/java-network-refactor/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ConnectionTypeTabControl.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ConnectionTypeTabControl.java:1061302-1072333
+/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ConnectionTypeTabControl.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ExchangeTypeTabControl.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ExchangeTypeTabControl.java:795950-829653
/qpid/branches/java-network-refactor/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ExchangeTypeTabControl.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ExchangeTypeTabControl.java:1061302-1072333
+/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/ExchangeTypeTabControl.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/MBeanTypeTabControl.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/MBeanTypeTabControl.java:795950-829653
/qpid/branches/java-network-refactor/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/MBeanTypeTabControl.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/MBeanTypeTabControl.java:1061302-1072333
+/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/MBeanTypeTabControl.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/QueueTypeTabControl.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/QueueTypeTabControl.java:795950-829653
/qpid/branches/java-network-refactor/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/QueueTypeTabControl.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/QueueTypeTabControl.java:1061302-1072333
+/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/type/QueueTypeTabControl.java:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/management/eclipse-plugin/src/main/resources/macosx/Contents/MacOS/qpidmc
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/management/eclipse-plugin/src/main/resources/macosx/Contents/MacOS/qpidmc:795950-829653
/qpid/branches/java-network-refactor/qpid/java/management/eclipse-plugin/src/main/resources/macosx/Contents/MacOS/qpidmc:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/eclipse-plugin/src/main/resources/macosx/Contents/MacOS/qpidmc:1061302-1072333
+/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/resources/macosx/Contents/MacOS/qpidmc:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -3,4 +3,5 @@
/qpid/branches/java-broker-0-10/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java:795950-829653
/qpid/branches/java-network-refactor/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java:805429-821809
/qpid/branches/qpid-2935/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java:1061302-1072333
+/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java:1156188
/qpid/trunk/qpid/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java:796646-796653
Propchange: qpid/branches/qpid-3346/qpid/java/test-profiles/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -1,2 +1,3 @@
/qpid/branches/0.5.x-dev/qpid/java/test-profiles:931179
/qpid/branches/qpid-2935/qpid/java/test-profiles:1061302-1072333
+/qpid/trunk/qpid/java/test-profiles:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/test-profiles/CPPExcludes
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/test-profiles/010Excludes:795950-829653
/qpid/branches/java-network-refactor/qpid/java/test-profiles/010Excludes:805429-821809
/qpid/branches/qpid-2935/qpid/java/test-profiles/CPPExcludes:1061302-1072333
+/qpid/trunk/qpid/java/test-profiles/CPPExcludes:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/test-profiles/Excludes
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/test-profiles/Excludes:795950-829653
/qpid/branches/java-network-refactor/qpid/java/test-profiles/Excludes:805429-821809
/qpid/branches/qpid-2935/qpid/java/test-profiles/Excludes:1061302-1072333
+/qpid/trunk/qpid/java/test-profiles/Excludes:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/test-profiles/JavaExcludes
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/test-profiles/08Excludes:795950-829653
/qpid/branches/java-network-refactor/qpid/java/test-profiles/08Excludes:805429-821809
/qpid/branches/qpid-2935/qpid/java/test-profiles/JavaExcludes:1061302-1072333
+/qpid/trunk/qpid/java/test-profiles/JavaExcludes:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/test-profiles/JavaPre010Excludes
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/test-profiles/08StandaloneExcludes:795950-829653
/qpid/branches/java-network-refactor/qpid/java/test-profiles/08StandaloneExcludes:805429-821809
/qpid/branches/qpid-2935/qpid/java/test-profiles/08StandaloneExcludes:1061302-1072333
+/qpid/trunk/qpid/java/test-profiles/JavaPre010Excludes:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/test-profiles/JavaTransientExcludes
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/test-profiles/08TransientExcludes:795950-829653
/qpid/branches/java-network-refactor/qpid/java/test-profiles/08TransientExcludes:805429-821809
/qpid/branches/qpid-2935/qpid/java/test-profiles/JavaTransientExcludes:1061302-1072333
+/qpid/trunk/qpid/java/test-profiles/JavaTransientExcludes:1156188
Propchange: qpid/branches/qpid-3346/qpid/java/test-profiles/XAExcludes
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Aug 16 00:37:33 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/test-profiles/XAExcludes:795950-829653
/qpid/branches/java-network-refactor/qpid/java/test-profiles/XAExcludes:805429-821809
/qpid/branches/qpid-2935/qpid/java/test-profiles/XAExcludes:1061302-1072333
+/qpid/trunk/qpid/java/test-profiles/XAExcludes:1156188
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org