QPID-1476 - routing keys used for updates can't be used to discriminate by agent

- Fixed routing keys in agents and binding keys in consoles
- Added some additional debug output for ManagementAgentImpl
- Minor cleanup in the connection close path for ManagementAgentImpl


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

2 files changed, 24 insertions, 13 deletions

diff --git a/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp b/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp
index 08aefa59d9..173785a671 100644
--- a/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp
+++ b/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp
@@ -192,6 +192,10 @@ void ManagementAgentImpl::raiseEvent(const ManagementEvent& event, severity_t se
     Buffer outBuffer(eventBuffer, MA_BUFFER_SIZE);
     uint32_t outLen;
     uint8_t sev = (severity == SEV_DEFAULT) ? event.getSeverity() : (uint8_t) severity;
+    stringstream key;
+
+    key << "console.event." << assignedBrokerBank << "." << assignedAgentBank << "." <<
+        event.getPackageName() << "." << event.getEventName();
 
     encodeHeader(outBuffer, 'e');
     outBuffer.putShortString(event.getPackageName());
@@ -202,8 +206,7 @@ void ManagementAgentImpl::raiseEvent(const ManagementEvent& event, severity_t se
     event.encode(outBuffer);
     outLen = MA_BUFFER_SIZE - outBuffer.available();
     outBuffer.reset();
-    connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management",
-                              "console.event." + event.getPackageName() + "." + event.getEventName());
+    connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", key.str());
 }
 
 uint32_t ManagementAgentImpl::pollCallbacks(uint32_t callLimit)
@@ -684,6 +687,10 @@ void ManagementAgentImpl::periodicProcessing()
 
     moveNewObjectsLH();
 
+    if (debugLevel >= DEBUG_PUBLISH) {
+        cout << "Objects managed: " << managementObjects.size() << endl;
+    }
+
     if (clientWasAdded) {
         clientWasAdded = false;
         for (ManagementObjectMap::iterator iter = managementObjects.begin();
@@ -752,12 +759,16 @@ void ManagementAgentImpl::periodicProcessing()
         if (contentSize > 0) {
             msgBuffer.reset();
             stringstream key;
-            key << "console.obj." << baseObject->getPackageName() << "." << baseObject->getClassName() << "." <<
-                assignedBrokerBank << "." << assignedAgentBank;
+            key << "console.obj." << assignedBrokerBank << "." << assignedAgentBank << "." <<
+                baseObject->getPackageName() << "." << baseObject->getClassName();
             connThreadBody.sendBuffer(msgBuffer, contentSize, "qpid.management", key.str());
         }
     }
 
+    if (debugLevel >= DEBUG_PUBLISH && !deleteList.empty()) {
+        cout << "Deleting " << deleteList.size() << " objects" << endl;
+    }
+
     // Delete flagged objects
     for (list<pair<ObjectId, ManagementObject*> >::reverse_iterator iter = deleteList.rbegin();
          iter != deleteList.rend();
@@ -798,6 +809,8 @@ void ManagementAgentImpl::ConnectionThread::run()
                     cout << "    Connection established" << endl;
                 {
                     Mutex::ScopedLock _lock(connLock);
+                    if (shutdown)
+                        return;
                     operational = true;
                     agent.startProtocol();
                     try {
@@ -809,12 +822,12 @@ void ManagementAgentImpl::ConnectionThread::run()
                         cout << "QMF Agent connection has been lost" << endl;
 
                     operational = false;
+                    agent.connected = false;
                 }
                 delay = delayMin;
+                connection.close();
                 delete subscriptions;
                 subscriptions = 0;
-                session.close();
-                connection.close();
             }
         } catch (exception &e) {
             if (delay < delayMax)
diff --git a/qpid/cpp/src/qpid/management/ManagementBroker.cpp b/qpid/cpp/src/qpid/management/ManagementBroker.cpp
index 23ef8d9e6a..48b73546b3 100644
--- a/qpid/cpp/src/qpid/management/ManagementBroker.cpp
+++ b/qpid/cpp/src/qpid/management/ManagementBroker.cpp
@@ -237,7 +237,7 @@ void ManagementBroker::raiseEvent(const ManagementEvent& event, severity_t sever
     outLen = MA_BUFFER_SIZE - outBuffer.available();
     outBuffer.reset();
     sendBuffer(outBuffer, outLen, mExchange,
-               "console.event." + event.getPackageName() + "." + event.getEventName());
+               "console.event.1.0." + event.getPackageName() + "." + event.getEventName());
 }
 
 ManagementBroker::Periodic::Periodic (ManagementBroker& _broker, uint32_t _seconds)
@@ -357,13 +357,11 @@ void ManagementBroker::periodicProcessing (void)
 
     moveNewObjectsLH();
 
-    if (clientWasAdded)
-    {
+    if (clientWasAdded) {
         clientWasAdded = false;
         for (ManagementObjectMap::iterator iter = managementObjects.begin ();
              iter != managementObjects.end ();
-             iter++)
-        {
+             iter++) {
             ManagementObject* object = iter->second;
             object->setAllChanged ();
         }
@@ -386,7 +384,7 @@ void ManagementBroker::periodicProcessing (void)
 
             contentSize = BUFSIZE - msgBuffer.available ();
             msgBuffer.reset ();
-            routingKey = "console.obj." + object->getPackageName() + "." + object->getClassName() + "1.0";
+            routingKey = "console.obj.1.0." + object->getPackageName() + "." + object->getClassName();
             sendBuffer (msgBuffer, contentSize, mExchange, routingKey);
         }
         
@@ -398,7 +396,7 @@ void ManagementBroker::periodicProcessing (void)
 
             contentSize = BUFSIZE - msgBuffer.available ();
             msgBuffer.reset ();
-            routingKey = "console.obj." + object->getPackageName() + "." + object->getClassName() + "1.0";
+            routingKey = "console.obj.1.0." + object->getPackageName() + "." + object->getClassName();
             sendBuffer (msgBuffer, contentSize, mExchange, routingKey);
         }