QPID-1461 - Added qmf event support for connections.

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

7 files changed, 84 insertions, 14 deletions

diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index 4dbfb153f2..f0b9980861 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -25,8 +25,9 @@
 #include "qpid/log/Statement.h"
 #include "qpid/ptr_map.h"
 #include "qpid/framing/AMQP_ClientProxy.h"
-#include "qpid/agent/ManagementAgent.h"
 #include "qpid/framing/enum.h"
+#include "qmf/org/apache/qpid/broker/EventClientConnect.h"
+#include "qmf/org/apache/qpid/broker/EventClientDisconnect.h"
 
 #include <boost/bind.hpp>
 #include <boost/ptr_container/ptr_vector.hpp>
@@ -55,7 +56,8 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
     mgmtClosing(false),
     mgmtId(mgmtId_),
     mgmtObject(0),
-    links(broker_.getLinks())
+    links(broker_.getLinks()),
+    agent(0)
 {
     Manageable* parent = broker.GetVhostObject();
 
@@ -64,7 +66,7 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
 
     if (parent != 0)
     {
-        ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+        agent = ManagementAgent::Singleton::getInstance();
 		
 		
         // TODO set last bool true if system connection
@@ -83,8 +85,11 @@ void Connection::requestIOProcessing(boost::function0<void> callback)
 
 Connection::~Connection()
 {
-    if (mgmtObject != 0)
+    if (mgmtObject != 0) {
         mgmtObject->resourceDestroy();
+        if (!isLink)
+            agent->raiseEvent(_qmf::EventClientDisconnect(mgmtId, ConnectionState::getUserId()));
+    }
     if (isLink)
         links.notifyClosed(mgmtId);
 }
@@ -153,8 +158,10 @@ void Connection::notifyConnectionForced(const string& text)
 void Connection::setUserId(const string& userId)
 {
     ConnectionState::setUserId(userId);
-    if (mgmtObject != 0)
+    if (mgmtObject != 0) {
         mgmtObject->set_authIdentity(userId);
+        agent->raiseEvent(_qmf::EventClientConnect(mgmtId, userId));
+    }
 }
 
 void Connection::setFederationLink(bool b)
diff --git a/cpp/src/qpid/broker/Connection.h b/cpp/src/qpid/broker/Connection.h
index 8ab58c1067..350ed2c07f 100644
--- a/cpp/src/qpid/broker/Connection.h
+++ b/cpp/src/qpid/broker/Connection.h
@@ -39,6 +39,7 @@
 #include "qpid/framing/AMQP_ClientProxy.h"
 #include "qpid/framing/AMQP_ServerOperations.h"
 #include "qpid/framing/ProtocolVersion.h"
+#include "qpid/agent/ManagementAgent.h"
 #include "qpid/management/Manageable.h"
 #include "qpid/ptr_map.h"
 #include "qpid/sys/AggregateOutput.h"
@@ -93,6 +94,9 @@ class Connection : public sys::ConnectionInputHandler,
     std::string getAuthCredentials();
     void notifyConnectionForced(const std::string& text);
     void setUserId(const string& uid);
+    const std::string& getUserId() const { return ConnectionState::getUserId(); }
+    const std::string& getMgmtId() const { return mgmtId; }
+    management::ManagementAgent* getAgent() const { return agent; }
     void setFederationLink(bool b);
 
     template <class F> void eachSessionHandler(F f) {
@@ -115,6 +119,7 @@ class Connection : public sys::ConnectionInputHandler,
     boost::function0<void> ioCallback;
     qmf::org::apache::qpid::broker::Connection* mgmtObject;
     LinkRegistry& links;
+    management::ManagementAgent* agent;
 };
 
 }}
diff --git a/cpp/src/qpid/broker/ConnectionHandler.cpp b/cpp/src/qpid/broker/ConnectionHandler.cpp
index 8639b7949e..e8e2cc3e3b 100644
--- a/cpp/src/qpid/broker/ConnectionHandler.cpp
+++ b/cpp/src/qpid/broker/ConnectionHandler.cpp
@@ -28,11 +28,12 @@
 #include "qpid/log/Statement.h"
 #include "qpid/Url.h"
 #include "AclModule.h"
+#include "qmf/org/apache/qpid/broker/EventClientConnectFail.h"
 
 using namespace qpid;
 using namespace qpid::broker;
 using namespace qpid::framing;
-
+namespace _qmf = qmf::org::apache::qpid::broker;
 
 namespace
 {
@@ -103,7 +104,19 @@ void ConnectionHandler::Handler::startOk(const framing::FieldTable& clientProper
                                          const string& response,
                                          const string& /*locale*/)
 {
-    authenticator->start(mechanism, response);
+    try {
+        authenticator->start(mechanism, response);
+    } catch (std::exception& e) {
+        management::ManagementAgent* agent = connection.getAgent();
+        if (agent) {
+            string error;
+            string uid;
+            authenticator->getError(error);
+            authenticator->getUid(uid);
+            agent->raiseEvent(_qmf::EventClientConnectFail(connection.getMgmtId(), uid, error));
+        }
+        throw;
+    }
     connection.setFederationLink(clientProperties.get(QPID_FED_LINK));
     connection.setFederationPeerTag(clientProperties.getAsString(QPID_FED_TAG));
     if (connection.isFederationLink()) {
@@ -117,7 +130,19 @@ void ConnectionHandler::Handler::startOk(const framing::FieldTable& clientProper
 
 void ConnectionHandler::Handler::secureOk(const string& response)
 {
-    authenticator->step(response);
+    try {
+        authenticator->step(response);
+    } catch (std::exception& e) {
+        management::ManagementAgent* agent = connection.getAgent();
+        if (agent) {
+            string error;
+            string uid;
+            authenticator->getError(error);
+            authenticator->getUid(uid);
+            agent->raiseEvent(_qmf::EventClientConnectFail(connection.getMgmtId(), uid, error));
+        }
+        throw;
+    }
 }
 
 void ConnectionHandler::Handler::tuneOk(uint16_t /*channelmax*/,
diff --git a/cpp/src/qpid/broker/Link.cpp b/cpp/src/qpid/broker/Link.cpp
index b026c89b75..bda9c80f0b 100644
--- a/cpp/src/qpid/broker/Link.cpp
+++ b/cpp/src/qpid/broker/Link.cpp
@@ -23,7 +23,8 @@
 #include "LinkRegistry.h"
 #include "Broker.h"
 #include "Connection.h"
-#include "qpid/agent/ManagementAgent.h"
+#include "qmf/org/apache/qpid/broker/EventBrokerLinkUp.h"
+#include "qmf/org/apache/qpid/broker/EventBrokerLinkDown.h"
 #include "boost/bind.hpp"
 #include "qpid/log/Statement.h"
 #include "qpid/framing/enum.h"
@@ -40,6 +41,7 @@ using qpid::management::ManagementObject;
 using qpid::management::Manageable;
 using qpid::management::Args;
 using qpid::sys::Mutex;
+using std::stringstream;
 namespace _qmf = qmf::org::apache::qpid::broker;
 
 Link::Link(LinkRegistry*  _links,
@@ -62,11 +64,12 @@ Link::Link(LinkRegistry*  _links,
       currentInterval(1),
       closing(false),
       channelCounter(1),
-      connection(0)
+      connection(0),
+      agent(0)
 {
     if (parent != 0)
     {
-        ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+        agent = ManagementAgent::Singleton::getInstance();
         if (agent != 0)
         {
             mgmtObject = new _qmf::Link(agent, this, parent, _host, _port, _transport, _durable);
@@ -123,8 +126,11 @@ void Link::startConnectionLH ()
 void Link::established ()
 {
     Mutex::ScopedLock mutex(lock);
+    stringstream addr;
+    addr << host << ":" << port;
 
-    QPID_LOG (info, "Inter-broker link established to " << host << ":" << port);
+    QPID_LOG (info, "Inter-broker link established to " << addr.str());
+    agent->raiseEvent(_qmf::EventBrokerLinkUp(addr.str()));
     setStateLH(STATE_OPERATIONAL);
     currentInterval = 1;
     visitCount      = 0;
@@ -138,8 +144,12 @@ void Link::closed (int, std::string text)
 
     connection = 0;
 
-    if (state == STATE_OPERATIONAL)
-        QPID_LOG (warning, "Inter-broker link disconnected from " << host << ":" << port);
+    if (state == STATE_OPERATIONAL) {
+        stringstream addr;
+        addr << host << ":" << port;
+        QPID_LOG (warning, "Inter-broker link disconnected from " << addr.str());
+        agent->raiseEvent(_qmf::EventBrokerLinkDown(addr.str()));
+    }
 
     for (Bridges::iterator i = active.begin(); i != active.end(); i++) {
         (*i)->cancel();
diff --git a/cpp/src/qpid/broker/Link.h b/cpp/src/qpid/broker/Link.h
index 69461472b1..a8dd528071 100644
--- a/cpp/src/qpid/broker/Link.h
+++ b/cpp/src/qpid/broker/Link.h
@@ -29,6 +29,7 @@
 #include "qpid/sys/Mutex.h"
 #include "qpid/framing/FieldTable.h"
 #include "qpid/management/Manageable.h"
+#include "qpid/agent/ManagementAgent.h"
 #include "qmf/org/apache/qpid/broker/Link.h"
 #include <boost/ptr_container/ptr_vector.hpp>
 
@@ -65,6 +66,7 @@ namespace qpid {
             Bridges active;    // Bridges active
             uint channelCounter;
             Connection* connection;
+            management::ManagementAgent* agent;
 
             static const int STATE_WAITING     = 1;
             static const int STATE_CONNECTING  = 2;
diff --git a/cpp/src/qpid/broker/SaslAuthenticator.cpp b/cpp/src/qpid/broker/SaslAuthenticator.cpp
index c8c447e97b..6c6ddd63cb 100644
--- a/cpp/src/qpid/broker/SaslAuthenticator.cpp
+++ b/cpp/src/qpid/broker/SaslAuthenticator.cpp
@@ -70,6 +70,8 @@ public:
     void getMechanisms(framing::Array& mechanisms);
     void start(const std::string& mechanism, const std::string& response);
     void step(const std::string& response);
+    void getUid(std::string& uid);
+    void getError(std::string& error);
 };
 
 bool SaslAuthenticator::available(void)
@@ -202,6 +204,23 @@ CyrusAuthenticator::~CyrusAuthenticator()
     }
 }
 
+void CyrusAuthenticator::getError(string& error)
+{
+    error = string(sasl_errdetail(sasl_conn));
+}
+
+void CyrusAuthenticator::getUid(string& uid)
+{
+    int code;
+    const void* ptr;
+
+    code = sasl_getprop(sasl_conn, SASL_USERNAME, &ptr);
+    if (SASL_OK != code)
+        return;
+
+    uid = string(const_cast<char*>(static_cast<const char*>(ptr)));
+}
+
 void CyrusAuthenticator::getMechanisms(Array& mechanisms)
 {
     const char *separator = " ";
diff --git a/cpp/src/qpid/broker/SaslAuthenticator.h b/cpp/src/qpid/broker/SaslAuthenticator.h
index 3e277f0527..2598b6d177 100644
--- a/cpp/src/qpid/broker/SaslAuthenticator.h
+++ b/cpp/src/qpid/broker/SaslAuthenticator.h
@@ -38,6 +38,8 @@ public:
     virtual void getMechanisms(framing::Array& mechanisms) = 0;
     virtual void start(const std::string& mechanism, const std::string& response) = 0;
     virtual void step(const std::string& response) = 0;
+    virtual void getUid(std::string&) {}
+    virtual void getError(std::string&) {}
 
     static bool available(void);