Added hooks in the broker for QMFv2 management of the broker.

Now both DirectExchange and TopicExchange have been subclassed so messages can be
redirected to the embedded management agent (in QMFv1, only the topic exchange was
subclassed this way).


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

2 files changed, 23 insertions, 8 deletions

diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index cbccca6eea..d94f228734 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -35,7 +35,8 @@
 #include "qmf/org/apache/qpid/broker/Package.h"
 #include "qmf/org/apache/qpid/broker/ArgsBrokerEcho.h"
 #include "qmf/org/apache/qpid/broker/ArgsBrokerQueueMoveMessages.h"
-#include "qpid/management/ManagementExchange.h"
+#include "qpid/management/ManagementDirectExchange.h"
+#include "qpid/management/ManagementTopicExchange.h"
 #include "qpid/log/Statement.h"
 #include "qpid/framing/AMQFrame.h"
 #include "qpid/framing/ProtocolInitiation.h"
@@ -234,11 +235,22 @@ Broker::Broker(const Broker::Options& conf) :
     declareStandardExchange(amq_match, HeadersExchange::typeName);
 
     if(conf.enableMgmt) {
-        exchanges.declare(qpid_management, ManagementExchange::typeName);
-        Exchange::shared_ptr mExchange = exchanges.get (qpid_management);
-        Exchange::shared_ptr dExchange = exchanges.get (amq_direct);
+        exchanges.declare(qpid_management, ManagementTopicExchange::typeName);
+        Exchange::shared_ptr mExchange = exchanges.get(qpid_management);
+        Exchange::shared_ptr dExchange = exchanges.get(amq_direct);
         managementAgent->setExchange(mExchange, dExchange);
-        boost::dynamic_pointer_cast<ManagementExchange>(mExchange)->setManagmentAgent(managementAgent.get());
+        boost::dynamic_pointer_cast<ManagementTopicExchange>(mExchange)->setManagmentAgent(managementAgent.get(), 1);
+
+        std::string qmfTopic("qmf.default.topic");
+        std::string qmfDirect("qmf.default.direct");
+
+        std::pair<Exchange::shared_ptr, bool> topicPair(exchanges.declare(qmfTopic, ManagementTopicExchange::typeName));
+        std::pair<Exchange::shared_ptr, bool> directPair(exchanges.declare(qmfDirect, ManagementDirectExchange::typeName));
+
+        boost::dynamic_pointer_cast<ManagementDirectExchange>(directPair.first)->setManagmentAgent(managementAgent.get(), 2);
+        boost::dynamic_pointer_cast<ManagementTopicExchange>(topicPair.first)->setManagmentAgent(managementAgent.get(), 2);
+
+        managementAgent->setExchangeV2(topicPair.first, directPair.first);
     }
     else
         QPID_LOG(info, "Management not enabled");
diff --git a/cpp/src/qpid/broker/ExchangeRegistry.cpp b/cpp/src/qpid/broker/ExchangeRegistry.cpp
index f4a860fa1e..20fdc4164a 100644
--- a/cpp/src/qpid/broker/ExchangeRegistry.cpp
+++ b/cpp/src/qpid/broker/ExchangeRegistry.cpp
@@ -24,7 +24,8 @@
 #include "qpid/broker/FanOutExchange.h"
 #include "qpid/broker/HeadersExchange.h"
 #include "qpid/broker/TopicExchange.h"
-#include "qpid/management/ManagementExchange.h"
+#include "qpid/management/ManagementDirectExchange.h"
+#include "qpid/management/ManagementTopicExchange.h"
 #include "qpid/framing/reply_exceptions.h"
 
 using namespace qpid::broker;
@@ -52,8 +53,10 @@ pair<Exchange::shared_ptr, bool> ExchangeRegistry::declare(const string& name, c
             exchange = Exchange::shared_ptr(new FanOutExchange(name, durable, args, parent, broker));
         }else if (type == HeadersExchange::typeName) {
             exchange = Exchange::shared_ptr(new HeadersExchange(name, durable, args, parent, broker));
-        }else if (type == ManagementExchange::typeName) {
-            exchange = Exchange::shared_ptr(new ManagementExchange(name, durable, args, parent, broker));
+        }else if (type == ManagementDirectExchange::typeName) {
+            exchange = Exchange::shared_ptr(new ManagementDirectExchange(name, durable, args, parent, broker));
+        }else if (type == ManagementTopicExchange::typeName) {
+            exchange = Exchange::shared_ptr(new ManagementTopicExchange(name, durable, args, parent, broker));
         }else{
             FunctionMap::iterator i =  factory.find(type);
             if (i == factory.end()) {