intercept mgmt msgs based on qmf version

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qmf-devel0.7a@928880 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 69 insertions, 32 deletions

diff --git a/qpid/cpp/src/qpid/management/ManagementAgent.cpp b/qpid/cpp/src/qpid/management/ManagementAgent.cpp
index c8efae9f99..86fb765759 100644
--- a/qpid/cpp/src/qpid/management/ManagementAgent.cpp
+++ b/qpid/cpp/src/qpid/management/ManagementAgent.cpp
@@ -888,35 +888,68 @@ void ManagementAgent::sendCommandComplete (string replyToKey, uint32_t sequence,
 
 bool ManagementAgent::dispatchCommand (Deliverable&      deliverable,
                                        const string&     routingKey,
-                                       const FieldTable* /*args*/)
+                                       const FieldTable* /*args*/,
+                                       const bool topic)
 {
     Mutex::ScopedLock lock (userLock);
     Message&  msg = ((DeliverableMessage&) deliverable).getMessage ();
 
-    // Parse the routing key.  This management broker should act as though it
-    // is bound to the exchange to match the following keys:
-    //
-    //    agent.1.0.#
-    //    broker
-    //    schema.#
+    if (qmf1Support && topic) {
 
-    if (routingKey == "broker") {
-        dispatchAgentCommandLH(msg);
-        return false;
-    }
+        // qmf1 is bound only to the topic management exchange.
+        // Parse the routing key.  This management broker should act as though it
+        // is bound to the exchange to match the following keys:
+        //
+        //    agent.1.0.#
+        //    broker
+        //    schema.#
 
-    else if (routingKey.compare(0, 9, "agent.1.0") == 0) {
-        dispatchAgentCommandLH(msg);
-        return false;
-    }
+        if (routingKey == "broker") {
+            dispatchAgentCommandLH(msg);
+            return false;
+        }
+
+        if (routingKey.length() > 6) {
+
+            if (routingKey.compare(0, 9, "agent.1.0") == 0) {
+                dispatchAgentCommandLH(msg);
+                return false;
+            }
+
+            if (routingKey.compare(0, 8, "agent.1.") == 0) {
+                return authorizeAgentMessageLH(msg);
+            }
 
-    else if (routingKey.compare(0, 8, "agent.1.") == 0) {
-        return authorizeAgentMessageLH(msg);
+            if (routingKey.compare(0, 7, "schema.") == 0) {
+                dispatchAgentCommandLH(msg);
+                return true;
+            }
+        }
     }
 
-    else if (routingKey.compare(0, 7, "schema.") == 0) {
-        dispatchAgentCommandLH(msg);
-        return true;
+    if (qmf2Support) {
+
+        if (topic) {
+
+            // Intercept messages bound to:
+            //  "console.ind.locate.# - process these messages, and also allow them to be forwarded.
+
+            if (routingKey.compare(0, 18, "console.ind.locate") == 0) {
+                dispatchAgentCommandLH(msg);
+                return true;
+            }
+
+        } else { // direct exchange
+
+            // Intercept messages bound to:
+            //  "broker" - generic alias for the local broker
+            //  "<name_address>" - the broker agent's proper name
+            // and do not forward them futher
+            if (routingKey == "broker" || routingKey == name_address) {
+                dispatchAgentCommandLH(msg);
+                return false;
+            }
+        }
     }
 
     return true;
diff --git a/qpid/cpp/src/qpid/management/ManagementAgent.h b/qpid/cpp/src/qpid/management/ManagementAgent.h
index 2f83a66476..913aa555a5 100644
--- a/qpid/cpp/src/qpid/management/ManagementAgent.h
+++ b/qpid/cpp/src/qpid/management/ManagementAgent.h
@@ -107,7 +107,8 @@ public:
 
     bool dispatchCommand (qpid::broker::Deliverable&       msg,
                           const std::string&         routingKey,
-                          const framing::FieldTable* args);
+                          const framing::FieldTable* args,
+                          const bool topic);
 
     const framing::Uuid& getUuid() const { return uuid; }
 
diff --git a/qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp b/qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp
index 0813e30891..6dc41ef073 100644
--- a/qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp
+++ b/qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp
@@ -29,13 +29,16 @@ using namespace qpid::framing;
 using namespace qpid::sys;
 
 ManagementDirectExchange::ManagementDirectExchange(const string& _name, Manageable* _parent, Broker* b) :
-    Exchange (_name, _parent, b), DirectExchange(_name, _parent, b) {}
+    Exchange (_name, _parent, b),
+    DirectExchange(_name, _parent, b),
+    managementAgent(0) {}
 ManagementDirectExchange::ManagementDirectExchange(const std::string& _name,
                                                    bool               _durable,
                                                    const FieldTable&  _args,
                                                    Manageable*        _parent, Broker* b) :
     Exchange (_name, _durable, _args, _parent, b), 
-    DirectExchange(_name, _durable, _args, _parent, b) {}
+    DirectExchange(_name, _durable, _args, _parent, b),
+    managementAgent(0) {}
 
 void ManagementDirectExchange::route(Deliverable&      msg,
                                      const string&     routingKey,
@@ -43,7 +46,8 @@ void ManagementDirectExchange::route(Deliverable&      msg,
 {
     bool routeIt = true;
 
-    // TODO: Intercept messages directed to the embedded agent and send them to the management agent.
+    if (managementAgent)
+        routeIt = managementAgent->dispatchCommand(msg, routingKey, args, false /*direct*/);
 
     if (routeIt)
         DirectExchange::route(msg, routingKey, args);
diff --git a/qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp b/qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp
index 98650b3adf..7fdce133e5 100644
--- a/qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp
+++ b/qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp
@@ -28,13 +28,16 @@ using namespace qpid::framing;
 using namespace qpid::sys;
 
 ManagementTopicExchange::ManagementTopicExchange(const string& _name, Manageable* _parent, Broker* b) :
-    Exchange (_name, _parent, b), TopicExchange(_name, _parent, b) {}
+    Exchange (_name, _parent, b),
+    TopicExchange(_name, _parent, b),
+    managementAgent(0) {}
 ManagementTopicExchange::ManagementTopicExchange(const std::string& _name,
                                                  bool               _durable,
                                                  const FieldTable&  _args,
                                                  Manageable*        _parent, Broker* b) :
     Exchange (_name, _durable, _args, _parent, b), 
-    TopicExchange(_name, _durable, _args, _parent, b) {}
+    TopicExchange(_name, _durable, _args, _parent, b),
+    managementAgent(0) {}
 
 void ManagementTopicExchange::route(Deliverable&      msg,
                                     const string&     routingKey,
@@ -43,12 +46,8 @@ void ManagementTopicExchange::route(Deliverable&      msg,
     bool routeIt = true;
 
     // Intercept management agent commands
-    if (qmfVersion == 1) {
-        if ((routingKey.length() > 6 &&
-             routingKey.substr(0, 6).compare("agent.") == 0) ||
-            (routingKey == "broker"))
-            routeIt = managementAgent->dispatchCommand(msg, routingKey, args);
-    }
+    if (managementAgent)
+        routeIt = managementAgent->dispatchCommand(msg, routingKey, args, true /* topic */);
 
     if (routeIt)
         TopicExchange::route(msg, routingKey, args);