summaryrefslogtreecommitdiff
path: root/cpp/src
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src')
-rw-r--r--cpp/src/qpid/acl/Acl.cpp29
-rw-r--r--cpp/src/qpid/acl/Acl.h3
-rw-r--r--cpp/src/qpid/acl/management-schema.xml42
-rw-r--r--cpp/src/qpid/agent/ManagementAgent.h37
-rw-r--r--cpp/src/qpid/agent/ManagementAgentImpl.cpp186
-rw-r--r--cpp/src/qpid/agent/ManagementAgentImpl.h70
-rw-r--r--cpp/src/qpid/broker/Connection.cpp1
-rw-r--r--cpp/src/qpid/broker/ConnectionState.h4
-rw-r--r--cpp/src/qpid/broker/SessionAdapter.cpp268
-rw-r--r--cpp/src/qpid/framing/Buffer.h4
-rw-r--r--cpp/src/qpid/management/Manageable.h1
-rw-r--r--cpp/src/qpid/management/ManagementBroker.cpp328
-rw-r--r--cpp/src/qpid/management/ManagementBroker.h60
-rw-r--r--cpp/src/qpid/management/ManagementEvent.h48
-rw-r--r--cpp/src/qpid/management/ManagementObject.cpp30
-rw-r--r--cpp/src/qpid/management/ManagementObject.h58
16 files changed, 683 insertions, 486 deletions
diff --git a/cpp/src/qpid/acl/Acl.cpp b/cpp/src/qpid/acl/Acl.cpp
index 0a793c88e0..bc932d836c 100644
--- a/cpp/src/qpid/acl/Acl.cpp
+++ b/cpp/src/qpid/acl/Acl.cpp
@@ -25,6 +25,10 @@
#include "qpid/shared_ptr.h"
#include "qpid/log/Logger.h"
#include "qmf/org/apache/qpid/acl/Package.h"
+#include "qmf/org/apache/qpid/acl/EventAllow.h"
+#include "qmf/org/apache/qpid/acl/EventDeny.h"
+#include "qmf/org/apache/qpid/acl/EventFileLoaded.h"
+#include "qmf/org/apache/qpid/acl/EventFileLoadFailed.h"
#include <map>
@@ -41,7 +45,7 @@ namespace _qmf = qmf::org::apache::qpid::acl;
Acl::Acl (AclValues& av, broker::Broker& b): aclValues(av), broker(&b), transferAcl(false)
{
- ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ agent = ManagementAgent::Singleton::getInstance();
if (agent != 0){
_qmf::Package packageInit(agent);
@@ -86,7 +90,11 @@ Acl::Acl (AclValues& av, broker::Broker& b): aclValues(av), broker(&b), transfer
switch (aclreslt)
{
case ALLOWLOG:
- QPID_LOG(info, "ACL Allow id:" << id <<" action:" << AclHelper::getActionStr(action) << " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name );
+ QPID_LOG(info, "ACL Allow id:" << id <<" action:" << AclHelper::getActionStr(action) <<
+ " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name );
+ agent->raiseEvent(_qmf::EventAllow(id, AclHelper::getActionStr(action),
+ AclHelper::getObjectTypeStr(objType),
+ name, framing::FieldTable()));
case ALLOW:
return true;
case DENY:
@@ -94,13 +102,12 @@ Acl::Acl (AclValues& av, broker::Broker& b): aclValues(av), broker(&b), transfer
return false;
case DENYLOG:
if (mgmtObject!=0) mgmtObject->inc_aclDenyCount();
- default:
- QPID_LOG(info, "ACL Deny id:" << id << " action:" << AclHelper::getActionStr(action) << " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name);
- if (mgmtObject!=0){
- framing::FieldTable _params;
- mgmtObject->event_aclEvent(1, id, AclHelper::getActionStr(action),AclHelper::getObjectTypeStr(objType),name, _params);
- }
- return false;
+ default:
+ QPID_LOG(info, "ACL Deny id:" << id << " action:" << AclHelper::getActionStr(action) << " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name);
+ agent->raiseEvent(_qmf::EventDeny(id, AclHelper::getActionStr(action),
+ AclHelper::getObjectTypeStr(objType),
+ name, framing::FieldTable()));
+ return false;
}
return false;
}
@@ -115,7 +122,7 @@ Acl::Acl (AclValues& av, broker::Broker& b): aclValues(av), broker(&b), transfer
boost::shared_ptr<AclData> d(new AclData);
AclReader ar;
if (ar.read(aclFile, d)){
- mgmtObject->event_fileNotLoaded("","See log for file load reason failure");
+ agent->raiseEvent(_qmf::EventFileLoadFailed("", "See log for file load reason failure"));
return false;
}
@@ -127,7 +134,7 @@ Acl::Acl (AclValues& av, broker::Broker& b): aclValues(av), broker(&b), transfer
sys::AbsTime now = sys::AbsTime::now();
int64_t ns = sys::Duration(now);
mgmtObject->set_lastAclLoad(ns);
- mgmtObject->event_fileLoaded("");
+ agent->raiseEvent(_qmf::EventFileLoaded(""));
}
return true;
}
diff --git a/cpp/src/qpid/acl/Acl.h b/cpp/src/qpid/acl/Acl.h
index fe1c1500bb..8a3825f683 100644
--- a/cpp/src/qpid/acl/Acl.h
+++ b/cpp/src/qpid/acl/Acl.h
@@ -27,6 +27,7 @@
#include "qpid/RefCounted.h"
#include "qpid/broker/AclModule.h"
#include "qpid/management/Manageable.h"
+#include "qpid/agent/ManagementAgent.h"
#include "qmf/org/apache/qpid/acl/Acl.h"
#include <map>
@@ -57,7 +58,7 @@ private:
bool transferAcl;
boost::shared_ptr<AclData> data;
qmf::org::apache::qpid::acl::Acl* mgmtObject; // mgnt owns lifecycle
-
+ qpid::management::ManagementAgent* agent;
public:
Acl (AclValues& av, broker::Broker& b);
diff --git a/cpp/src/qpid/acl/management-schema.xml b/cpp/src/qpid/acl/management-schema.xml
index 7d20353755..f362561356 100644
--- a/cpp/src/qpid/acl/management-schema.xml
+++ b/cpp/src/qpid/acl/management-schema.xml
@@ -17,32 +17,28 @@
-->
<class name="acl">
- <property name="brokerRef" type="objId" references="qpid.Broker" access="RO" index="y" parentRef="y"/>
- <property name="policyFile" type="sstr" access="RO" desc="Name of the policy file"/>
- <property name="enforcingAcl" type="bool" access="RO" desc="Currently Enforcing ACL"/>
- <property name="transferAcl" type="bool" access="RO" desc="Any transfer ACL rules in force"/>
- <property name="lastAclLoad" type="absTime" access="RO" desc="Timestamp of last successful load of ACL"/>
- <statistic name="aclDenyCount" type="count64" unit="record" desc="Number of ACL requests denied"/>
+ <property name="brokerRef" type="objId" references="org.apache.qpid.broker:Broker" access="RO" index="y" parentRef="y"/>
+ <property name="policyFile" type="sstr" access="RO" desc="Name of the policy file"/>
+ <property name="enforcingAcl" type="bool" access="RO" desc="Currently Enforcing ACL"/>
+ <property name="transferAcl" type="bool" access="RO" desc="Any transfer ACL rules in force"/>
+ <property name="lastAclLoad" type="absTime" access="RO" desc="Timestamp of last successful load of ACL"/>
+ <statistic name="aclDenyCount" type="count64" unit="request" desc="Number of ACL requests denied"/>
<method name="reloadACLFile" desc="Reload the ACL file"/>
+ </class>
- <event name="aclEvent" defaultSeverity="info" desc="Event generated by the ACL policy">
- <arg name="denied" type="bool"/>
- <arg name="authId" type="sstr"/>
- <arg name="action" type="sstr"/>
- <arg name="objType" type="sstr"/>
- <arg name="name" type="sstr"/>
- <arg name="params" type="map"/>
- </event>
-
- <event name="fileLoaded" defaultSeverity="warning" desc="ACL file successfully loaded - New policy in effect">
- <arg name="authId" type="sstr" desc="Name of user who initiated the file load"/>
- </event>
+ <eventArguments>
+ <arg name="action" type="sstr"/>
+ <arg name="arguments" type="map"/>
+ <arg name="objectName" type="sstr"/>
+ <arg name="objectType" type="sstr"/>
+ <arg name="reason" type="sstr"/>
+ <arg name="userId" type="sstr"/>
+ </eventArguments>
- <event name="fileNotLoaded" defaultSeverity="error" desc="Replacement ACL file could not be loaded">
- <arg name="authId" type="sstr" desc="Name of user who initiated the file load"/>
- <arg name="reason" type="sstr" desc="Reason for failure"/>
- </event>
- </class>
+ <event name="allow" args="userId, action, objectType, objectName, arguments"/>
+ <event name="deny" args="userId, action, objectType, objectName, arguments"/>
+ <event name="fileLoaded" args="userId"/>
+ <event name="fileLoadFailed" args="userId, reason"/>
</schema>
diff --git a/cpp/src/qpid/agent/ManagementAgent.h b/cpp/src/qpid/agent/ManagementAgent.h
index 1c219f7463..6af9abc26b 100644
--- a/cpp/src/qpid/agent/ManagementAgent.h
+++ b/cpp/src/qpid/agent/ManagementAgent.h
@@ -21,6 +21,7 @@
//
#include "qpid/management/ManagementObject.h"
+#include "qpid/management/ManagementEvent.h"
#include "qpid/management/Manageable.h"
#include "qpid/sys/Mutex.h"
@@ -43,8 +44,8 @@ class ManagementAgent
static ManagementAgent* agent;
};
- ManagementAgent () {}
- virtual ~ManagementAgent () {}
+ ManagementAgent() {}
+ virtual ~ManagementAgent() {}
virtual int getMaxThreads() = 0;
@@ -78,10 +79,16 @@ class ManagementAgent
// package initializer generated by the management code generator.
//
virtual void
- RegisterClass (std::string packageName,
- std::string className,
- uint8_t* md5Sum,
- management::ManagementObject::writeSchemaCall_t schemaCall) = 0;
+ registerClass(std::string& packageName,
+ std::string& className,
+ uint8_t* md5Sum,
+ management::ManagementObject::writeSchemaCall_t schemaCall) = 0;
+
+ virtual void
+ registerEvent(std::string& packageName,
+ std::string& eventName,
+ uint8_t* md5Sum,
+ management::ManagementEvent::writeSchemaCall_t schemaCall) = 0;
// Add a management object to the agent. Once added, this object shall be visible
// in the greater management context.
@@ -97,8 +104,11 @@ class ManagementAgent
// pointer. This allows the management agent to report the deletion of the object
// in an orderly way.
//
- virtual ObjectId addObject (ManagementObject* objectPtr,
- uint64_t persistId = 0) = 0;
+ virtual ObjectId addObject(ManagementObject* objectPtr, uint64_t persistId = 0) = 0;
+
+ //
+ //
+ virtual void raiseEvent(const ManagementEvent& event) = 0;
// If "useExternalThread" was set to true in init, this method must
// be called to provide a thread for any pending method calls that have arrived.
@@ -113,7 +123,7 @@ class ManagementAgent
// to pollCallbacks are necessary to clear the backlog. If callLimit is zero,
// the return value will also be zero.
//
- virtual uint32_t pollCallbacks (uint32_t callLimit = 0) = 0;
+ virtual uint32_t pollCallbacks(uint32_t callLimit = 0) = 0;
// If "useExternalThread" was set to true in the constructor, this method provides
// a standard file descriptor that can be used in a select statement to signal that
@@ -121,14 +131,7 @@ class ManagementAgent
// least one method call). When this fd is ready-for-read, pollCallbacks may be
// invoked. Calling pollCallbacks shall reset the ready-to-read state of the fd.
//
- virtual int getSignalFd (void) = 0;
-
-protected:
- friend class ManagementObject;
- virtual sys::Mutex& getMutex() = 0;
- virtual framing::Buffer* startEventLH() = 0;
- virtual void finishEventLH(framing::Buffer* buf) = 0;
-
+ virtual int getSignalFd() = 0;
};
}}
diff --git a/cpp/src/qpid/agent/ManagementAgentImpl.cpp b/cpp/src/qpid/agent/ManagementAgentImpl.cpp
index 6a1542b1f2..b178d0fc28 100644
--- a/cpp/src/qpid/agent/ManagementAgentImpl.cpp
+++ b/cpp/src/qpid/agent/ManagementAgentImpl.cpp
@@ -75,12 +75,13 @@ ManagementAgent* ManagementAgent::Singleton::getInstance()
return agent;
}
-const string ManagementAgentImpl::storeMagicNumber("MA01");
+const string ManagementAgentImpl::storeMagicNumber("MA02");
ManagementAgentImpl::ManagementAgentImpl() :
extThread(false), writeFd(-1), readFd(-1),
- clientWasAdded(true), requestedBank(0),
- assignedBank(0), brokerBank(0), bootSequence(0),
+ connected(false), lastFailure("never connected"),
+ clientWasAdded(true), requestedBrokerBank(0), requestedAgentBank(0),
+ assignedBrokerBank(0), assignedAgentBank(0), bootSequence(0),
connThreadBody(*this), connThread(connThreadBody),
pubThreadBody(*this), pubThread(pubThreadBody)
{
@@ -122,18 +123,24 @@ void ManagementAgentImpl::init(string brokerHost,
storeData(true);
}
-ManagementAgentImpl::~ManagementAgentImpl()
-{
+void ManagementAgentImpl::registerClass(std::string& packageName,
+ std::string& className,
+ uint8_t* md5Sum,
+ management::ManagementObject::writeSchemaCall_t schemaCall)
+{
+ Mutex::ScopedLock lock(agentLock);
+ PackageMap::iterator pIter = findOrAddPackage(packageName);
+ addClassLocal(ManagementItem::CLASS_KIND_TABLE, pIter, className, md5Sum, schemaCall);
}
-void ManagementAgentImpl::RegisterClass(std::string packageName,
- std::string className,
- uint8_t* md5Sum,
+void ManagementAgentImpl::registerEvent(std::string& packageName,
+ std::string& eventName,
+ uint8_t* md5Sum,
management::ManagementObject::writeSchemaCall_t schemaCall)
{
Mutex::ScopedLock lock(agentLock);
- PackageMap::iterator pIter = FindOrAddPackage(packageName);
- AddClassLocal(pIter, className, md5Sum, schemaCall);
+ PackageMap::iterator pIter = findOrAddPackage(packageName);
+ addClassLocal(ManagementItem::CLASS_KIND_EVENT, pIter, eventName, md5Sum, schemaCall);
}
ObjectId ManagementAgentImpl::addObject(ManagementObject* object,
@@ -151,6 +158,23 @@ ObjectId ManagementAgentImpl::addObject(ManagementObject* object,
return objectId;
}
+void ManagementAgentImpl::raiseEvent(const ManagementEvent& event)
+{
+ Mutex::ScopedLock lock(agentLock);
+ Buffer outBuffer(eventBuffer, MA_BUFFER_SIZE);
+ uint32_t outLen;
+
+ encodeHeader(outBuffer, 'e');
+ outBuffer.putShortString(event.getPackageName());
+ outBuffer.putShortString(event.getEventName());
+ outBuffer.putBin128(event.getMd5Sum());
+ outBuffer.putLongLong(uint64_t(Duration(now())));
+ event.encode(outBuffer);
+ outLen = MA_BUFFER_SIZE - outBuffer.available();
+ outBuffer.reset();
+ connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "mgmt.event");
+}
+
uint32_t ManagementAgentImpl::pollCallbacks(uint32_t callLimit)
{
Mutex::ScopedLock lock(agentLock);
@@ -184,10 +208,12 @@ void ManagementAgentImpl::startProtocol()
char rawbuffer[512];
Buffer buffer(rawbuffer, 512);
- EncodeHeader(buffer, 'A');
+ connected = true;
+ encodeHeader(buffer, 'A');
buffer.putShortString("RemoteAgent [C++]");
systemId.encode (buffer);
- buffer.putLong(requestedBank);
+ buffer.putLong(requestedBrokerBank);
+ buffer.putLong(requestedAgentBank);
uint32_t length = 512 - buffer.available();
buffer.reset();
connThreadBody.sendBuffer(buffer, length, "qpid.management", "broker");
@@ -197,10 +223,12 @@ void ManagementAgentImpl::storeData(bool requested)
{
if (!storeFile.empty()) {
ofstream outFile(storeFile.c_str());
- uint32_t bankToWrite = requested ? requestedBank : assignedBank;
+ uint32_t brokerBankToWrite = requested ? requestedBrokerBank : assignedBrokerBank;
+ uint32_t agentBankToWrite = requested ? requestedAgentBank : assignedAgentBank;
if (outFile.good()) {
- outFile << storeMagicNumber << " " << bankToWrite << " " << bootSequence << endl;
+ outFile << storeMagicNumber << " " << brokerBankToWrite << " " <<
+ agentBankToWrite << " " << bootSequence << endl;
outFile.close();
}
}
@@ -215,7 +243,8 @@ void ManagementAgentImpl::retrieveData()
if (inFile.good()) {
inFile >> mn;
if (mn == storeMagicNumber) {
- inFile >> requestedBank;
+ inFile >> requestedBrokerBank;
+ inFile >> requestedAgentBank;
inFile >> bootSequence;
}
inFile.close();
@@ -229,7 +258,7 @@ void ManagementAgentImpl::sendCommandComplete(string replyToKey, uint32_t sequen
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'z', sequence);
+ encodeHeader(outBuffer, 'z', sequence);
outBuffer.putLong(code);
outBuffer.putShortString(text);
outLen = MA_BUFFER_SIZE - outBuffer.available();
@@ -241,20 +270,23 @@ void ManagementAgentImpl::handleAttachResponse(Buffer& inBuffer)
{
Mutex::ScopedLock lock(agentLock);
- brokerBank = inBuffer.getLong();
- assignedBank = inBuffer.getLong();
- if (assignedBank != requestedBank) {
- if (requestedBank == 0)
- cout << "Initial object-id bank assigned: " << assignedBank << endl;
+ assignedBrokerBank = inBuffer.getLong();
+ assignedAgentBank = inBuffer.getLong();
+ if ((assignedBrokerBank != requestedBrokerBank) ||
+ (assignedAgentBank != requestedAgentBank)) {
+ if (requestedAgentBank == 0)
+ cout << "Initial object-id bank assigned: " << assignedBrokerBank << "." <<
+ assignedAgentBank << endl;
else
- cout << "Collision in object-id! New bank assigned: " << assignedBank << endl;
+ cout << "Collision in object-id! New bank assigned: " << assignedBrokerBank <<
+ "." << assignedAgentBank << endl;
storeData();
}
- attachment.setBanks(brokerBank, assignedBank);
+ attachment.setBanks(assignedBrokerBank, assignedAgentBank);
// Bind to qpid.management to receive commands
- connThreadBody.bindToBank(assignedBank);
+ connThreadBody.bindToBank(assignedBrokerBank, assignedAgentBank);
// Send package indications for all local packages
for (PackageMap::iterator pIter = packages.begin();
@@ -263,8 +295,8 @@ void ManagementAgentImpl::handleAttachResponse(Buffer& inBuffer)
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'p');
- EncodePackageIndication(outBuffer, pIter);
+ encodeHeader(outBuffer, 'p');
+ encodePackageIndication(outBuffer, pIter);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "broker");
@@ -273,8 +305,8 @@ void ManagementAgentImpl::handleAttachResponse(Buffer& inBuffer)
ClassMap cMap = pIter->second;
for (ClassMap::iterator cIter = cMap.begin(); cIter != cMap.end(); cIter++) {
outBuffer.reset();
- EncodeHeader(outBuffer, 'q');
- EncodeClassIndication(outBuffer, pIter, cIter);
+ encodeHeader(outBuffer, 'q');
+ encodeClassIndication(outBuffer, pIter, cIter);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "broker");
@@ -294,14 +326,14 @@ void ManagementAgentImpl::handleSchemaRequest(Buffer& inBuffer, uint32_t sequenc
PackageMap::iterator pIter = packages.find(packageName);
if (pIter != packages.end()) {
- ClassMap cMap = pIter->second;
+ ClassMap& cMap = pIter->second;
ClassMap::iterator cIter = cMap.find(key);
if (cIter != cMap.end()) {
- SchemaClass schema = cIter->second;
+ SchemaClass& schema = cIter->second;
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 's', sequence);
+ encodeHeader(outBuffer, 's', sequence);
schema.writeSchemaCall(outBuffer);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
@@ -331,7 +363,7 @@ void ManagementAgentImpl::invokeMethodRequest(Buffer& inBuffer, uint32_t sequenc
inBuffer.getBin128(hash);
inBuffer.getShortString(methodName);
- EncodeHeader(outBuffer, 'm', sequence);
+ encodeHeader(outBuffer, 'm', sequence);
ManagementObjectMap::iterator iter = managementObjects.find(objId);
if (iter == managementObjects.end() || iter->second->isDeleted()) {
@@ -344,7 +376,14 @@ void ManagementAgentImpl::invokeMethodRequest(Buffer& inBuffer, uint32_t sequenc
outBuffer.putShortString (Manageable::StatusText (Manageable::STATUS_INVALID_PARAMETER));
}
else
- iter->second->doMethod(methodName, inBuffer, outBuffer);
+ try {
+ outBuffer.record();
+ iter->second->doMethod(methodName, inBuffer, outBuffer);
+ } catch(std::exception& e) {
+ outBuffer.restore();
+ outBuffer.putLong(Manageable::STATUS_EXCEPTION);
+ outBuffer.putShortString(e.what());
+ }
}
outLen = MA_BUFFER_SIZE - outBuffer.available();
@@ -379,7 +418,7 @@ void ManagementAgentImpl::handleGetQuery(Buffer& inBuffer, uint32_t sequence, st
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'g', sequence);
+ encodeHeader(outBuffer, 'g', sequence);
object->writeProperties(outBuffer);
object->writeStatistics(outBuffer, true);
outLen = MA_BUFFER_SIZE - outBuffer.available();
@@ -419,7 +458,7 @@ void ManagementAgentImpl::received(Message& msg)
replyToKey = rt.getRoutingKey();
}
- if (CheckHeader(inBuffer, &opcode, &sequence))
+ if (checkHeader(inBuffer, &opcode, &sequence))
{
if (opcode == 'a') handleAttachResponse(inBuffer);
else if (opcode == 'S') handleSchemaRequest(inBuffer, sequence);
@@ -429,16 +468,16 @@ void ManagementAgentImpl::received(Message& msg)
}
}
-void ManagementAgentImpl::EncodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq)
+void ManagementAgentImpl::encodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq)
{
buf.putOctet('A');
buf.putOctet('M');
- buf.putOctet('1');
+ buf.putOctet('2');
buf.putOctet(opcode);
buf.putLong (seq);
}
-bool ManagementAgentImpl::CheckHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq)
+bool ManagementAgentImpl::checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq)
{
if (buf.getSize() < 8)
return false;
@@ -450,10 +489,10 @@ bool ManagementAgentImpl::CheckHeader(Buffer& buf, uint8_t *opcode, uint32_t *se
*opcode = buf.getOctet();
*seq = buf.getLong();
- return h1 == 'A' && h2 == 'M' && h3 == '1';
+ return h1 == 'A' && h2 == 'M' && h3 == '2';
}
-ManagementAgentImpl::PackageMap::iterator ManagementAgentImpl::FindOrAddPackage(std::string name)
+ManagementAgentImpl::PackageMap::iterator ManagementAgentImpl::findOrAddPackage(const string& name)
{
PackageMap::iterator pIter = packages.find(name);
if (pIter != packages.end())
@@ -467,8 +506,8 @@ ManagementAgentImpl::PackageMap::iterator ManagementAgentImpl::FindOrAddPackage(
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'p');
- EncodePackageIndication(outBuffer, result.first);
+ encodeHeader(outBuffer, 'p');
+ encodePackageIndication(outBuffer, result.first);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "mgmt.schema.package");
@@ -486,8 +525,9 @@ void ManagementAgentImpl::moveNewObjectsLH()
newManagementObjects.clear();
}
-void ManagementAgentImpl::AddClassLocal(PackageMap::iterator pIter,
- string className,
+void ManagementAgentImpl::addClassLocal(uint8_t classKind,
+ PackageMap::iterator pIter,
+ const string& className,
uint8_t* md5Sum,
management::ManagementObject::writeSchemaCall_t schemaCall)
{
@@ -502,32 +542,28 @@ void ManagementAgentImpl::AddClassLocal(PackageMap::iterator pIter,
return;
// No such class found, create a new class with local information.
- SchemaClass classInfo;
-
- classInfo.writeSchemaCall = schemaCall;
- cMap[key] = classInfo;
-
- // TODO: Publish a class-indication message
+ cMap.insert(std::pair<SchemaClassKey, SchemaClass>(key, SchemaClass(schemaCall, classKind)));
}
-void ManagementAgentImpl::EncodePackageIndication(Buffer& buf,
+void ManagementAgentImpl::encodePackageIndication(Buffer& buf,
PackageMap::iterator pIter)
{
buf.putShortString((*pIter).first);
}
-void ManagementAgentImpl::EncodeClassIndication(Buffer& buf,
+void ManagementAgentImpl::encodeClassIndication(Buffer& buf,
PackageMap::iterator pIter,
ClassMap::iterator cIter)
{
SchemaClassKey key = (*cIter).first;
+ buf.putOctet((*cIter).second.kind);
buf.putShortString((*pIter).first);
buf.putShortString(key.name);
- buf.putBin128 (key.hash);
+ buf.putBin128(key.hash);
}
-void ManagementAgentImpl::PeriodicProcessing()
+void ManagementAgentImpl::periodicProcessing()
{
#define BUFSIZE 65536
Mutex::ScopedLock lock(agentLock);
@@ -536,9 +572,12 @@ void ManagementAgentImpl::PeriodicProcessing()
string routingKey;
std::list<ObjectId> deleteList;
+ if (!connected)
+ return;
+
{
Buffer msgBuffer(msgChars, BUFSIZE);
- EncodeHeader(msgBuffer, 'h');
+ encodeHeader(msgBuffer, 'h');
msgBuffer.putLongLong(uint64_t(Duration(now())));
contentSize = BUFSIZE - msgBuffer.available();
@@ -573,7 +612,7 @@ void ManagementAgentImpl::PeriodicProcessing()
if (object->getConfigChanged() || object->isDeleted())
{
Buffer msgBuffer(msgChars, BUFSIZE);
- EncodeHeader(msgBuffer, 'c');
+ encodeHeader(msgBuffer, 'c');
object->writeProperties(msgBuffer);
contentSize = BUFSIZE - msgBuffer.available();
@@ -585,7 +624,7 @@ void ManagementAgentImpl::PeriodicProcessing()
if (object->getInstChanged())
{
Buffer msgBuffer(msgChars, BUFSIZE);
- EncodeHeader(msgBuffer, 'i');
+ encodeHeader(msgBuffer, 'i');
object->writeStatistics(msgBuffer);
contentSize = BUFSIZE - msgBuffer.available();
@@ -664,8 +703,8 @@ ManagementAgentImpl::ConnectionThread::~ConnectionThread()
void ManagementAgentImpl::ConnectionThread::sendBuffer(Buffer& buf,
uint32_t length,
- string exchange,
- string routingKey)
+ const string& exchange,
+ const string& routingKey)
{
{
Mutex::ScopedLock _lock(connLock);
@@ -683,10 +722,10 @@ void ManagementAgentImpl::ConnectionThread::sendBuffer(Buffer& buf,
session.messageTransfer(arg::content=msg, arg::destination=exchange);
}
-void ManagementAgentImpl::ConnectionThread::bindToBank(uint32_t agentBank)
+void ManagementAgentImpl::ConnectionThread::bindToBank(uint32_t brokerBank, uint32_t agentBank)
{
stringstream key;
- key << "agent." << agentBank;
+ key << "agent." << brokerBank << "." << agentBank;
session.exchangeBind(arg::exchange="qpid.management", arg::queue=queueName.str(),
arg::bindingKey=key.str());
}
@@ -695,28 +734,7 @@ void ManagementAgentImpl::ConnectionThread::bindToBank(uint32_t agentBank)
void ManagementAgentImpl::PublishThread::run()
{
while (true) {
- ::sleep(5);
- agent.PeriodicProcessing();
+ ::sleep(agent.getInterval());
+ agent.periodicProcessing();
}
}
-
-Mutex& ManagementAgentImpl::getMutex()
-{
- return agentLock;
-}
-
-Buffer* ManagementAgentImpl::startEventLH()
-{
- Buffer* outBuffer(new Buffer(eventBuffer, MA_BUFFER_SIZE));
- EncodeHeader(*outBuffer, 'e');
- outBuffer->putLongLong(uint64_t(Duration(now())));
- return outBuffer;
-}
-
-void ManagementAgentImpl::finishEventLH(Buffer* outBuffer)
-{
- uint32_t outLen = MA_BUFFER_SIZE - outBuffer->available();
- outBuffer->reset();
- connThreadBody.sendBuffer(*outBuffer, outLen, "qpid.management", "mgmt.event");
- delete outBuffer;
-}
diff --git a/cpp/src/qpid/agent/ManagementAgentImpl.h b/cpp/src/qpid/agent/ManagementAgentImpl.h
index 7d9be6daf9..a964694690 100644
--- a/cpp/src/qpid/agent/ManagementAgentImpl.h
+++ b/cpp/src/qpid/agent/ManagementAgentImpl.h
@@ -43,24 +43,34 @@ class ManagementAgentImpl : public ManagementAgent, public client::MessageListen
public:
ManagementAgentImpl();
- virtual ~ManagementAgentImpl();
+ virtual ~ManagementAgentImpl() {};
+ //
+ // Methods from ManagementAgent
+ //
int getMaxThreads() { return 1; }
void init(std::string brokerHost = "localhost",
uint16_t brokerPort = 5672,
uint16_t intervalSeconds = 10,
bool useExternalThread = false,
std::string storeFile = "");
- void RegisterClass(std::string packageName,
- std::string className,
- uint8_t* md5Sum,
+ bool isConnected() { return connected; }
+ std::string& getLastFailure() { return lastFailure; }
+ void registerClass(std::string& packageName,
+ std::string& className,
+ uint8_t* md5Sum,
management::ManagementObject::writeSchemaCall_t schemaCall);
- ObjectId addObject (management::ManagementObject* objectPtr,
- uint64_t persistId = 0);
- uint32_t pollCallbacks (uint32_t callLimit = 0);
- int getSignalFd (void);
+ void registerEvent(std::string& packageName,
+ std::string& eventName,
+ uint8_t* md5Sum,
+ management::ManagementObject::writeSchemaCall_t schemaCall);
+ ObjectId addObject(management::ManagementObject* objectPtr, uint64_t persistId = 0);
+ void raiseEvent(const management::ManagementEvent& event);
+ uint32_t pollCallbacks(uint32_t callLimit = 0);
+ int getSignalFd();
- void PeriodicProcessing();
+ uint16_t getInterval() { return interval; }
+ void periodicProcessing();
private:
@@ -84,8 +94,10 @@ class ManagementAgentImpl : public ManagementAgent, public client::MessageListen
struct SchemaClass {
management::ManagementObject::writeSchemaCall_t writeSchemaCall;
+ uint8_t kind;
- SchemaClass () : writeSchemaCall(0) {}
+ SchemaClass(const management::ManagementObject::writeSchemaCall_t call,
+ const uint8_t _kind) : writeSchemaCall(call), kind(_kind) {}
};
struct QueuedMethod {
@@ -120,12 +132,16 @@ class ManagementAgentImpl : public ManagementAgent, public client::MessageListen
framing::Uuid systemId;
std::string host;
uint16_t port;
+ bool connected;
+ std::string lastFailure;
+
+ bool clientWasAdded;
+ uint32_t requestedBrokerBank;
+ uint32_t requestedAgentBank;
+ uint32_t assignedBrokerBank;
+ uint32_t assignedAgentBank;
+ uint16_t bootSequence;
- bool clientWasAdded;
- uint32_t requestedBank;
- uint32_t assignedBank;
- uint32_t brokerBank;
- uint16_t bootSequence;
# define MA_BUFFER_SIZE 65536
char outputBuffer[MA_BUFFER_SIZE];
char eventBuffer[MA_BUFFER_SIZE];
@@ -148,9 +164,9 @@ class ManagementAgentImpl : public ManagementAgent, public client::MessageListen
~ConnectionThread();
void sendBuffer(qpid::framing::Buffer& buf,
uint32_t length,
- std::string exchange,
- std::string routingKey);
- void bindToBank(uint32_t agentBank);
+ const std::string& exchange,
+ const std::string& routingKey);
+ void bindToBank(uint32_t brokerBank, uint32_t agentBank);
};
class PublishThread : public sys::Runnable
@@ -171,19 +187,20 @@ class ManagementAgentImpl : public ManagementAgent, public client::MessageListen
void startProtocol();
void storeData(bool requested=false);
void retrieveData();
- PackageMap::iterator FindOrAddPackage (std::string name);
+ PackageMap::iterator findOrAddPackage(const std::string& name);
void moveNewObjectsLH();
- void AddClassLocal (PackageMap::iterator pIter,
- std::string className,
+ void addClassLocal (uint8_t classKind,
+ PackageMap::iterator pIter,
+ const std::string& className,
uint8_t* md5Sum,
management::ManagementObject::writeSchemaCall_t schemaCall);
- void EncodePackageIndication (qpid::framing::Buffer& buf,
+ void encodePackageIndication (framing::Buffer& buf,
PackageMap::iterator pIter);
- void EncodeClassIndication (qpid::framing::Buffer& buf,
+ void encodeClassIndication (framing::Buffer& buf,
PackageMap::iterator pIter,
ClassMap::iterator cIter);
- void EncodeHeader (qpid::framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0);
- bool CheckHeader (qpid::framing::Buffer& buf, uint8_t *opcode, uint32_t *seq);
+ void encodeHeader (framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0);
+ bool checkHeader (framing::Buffer& buf, uint8_t *opcode, uint32_t *seq);
void sendCommandComplete (std::string replyToKey, uint32_t sequence,
uint32_t code = 0, std::string text = std::string("OK"));
void handleAttachResponse (qpid::framing::Buffer& inBuffer);
@@ -194,9 +211,6 @@ class ManagementAgentImpl : public ManagementAgent, public client::MessageListen
void handleGetQuery (qpid::framing::Buffer& inBuffer, uint32_t sequence, std::string replyTo);
void handleMethodRequest (qpid::framing::Buffer& inBuffer, uint32_t sequence, std::string replyTo);
void handleConsoleAddedIndication();
- sys::Mutex& getMutex();
- framing::Buffer* startEventLH();
- void finishEventLH(framing::Buffer* outBuffer);
};
}}
diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index 38affad51f..3ed41a3f2d 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -72,6 +72,7 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
if (agent != 0)
mgmtObject = new _qmf::Connection(agent, this, parent, mgmtId, !isLink, false);
agent->addObject(mgmtObject);
+ ConnectionState::setUrl(mgmtId);
}
}
diff --git a/cpp/src/qpid/broker/ConnectionState.h b/cpp/src/qpid/broker/ConnectionState.h
index aac31bbf96..c04bd46f72 100644
--- a/cpp/src/qpid/broker/ConnectionState.h
+++ b/cpp/src/qpid/broker/ConnectionState.h
@@ -63,6 +63,9 @@ class ConnectionState : public ConnectionToken, public management::Manageable
virtual void setUserId(const string& uid) { userId = uid; }
const string& getUserId() const { return userId; }
+ void setUrl(const string& _url) { url = _url; }
+ const string& getUrl() const { return url; }
+
void setFederationLink(bool b) { federationLink = b; }
bool isFederationLink() const { return federationLink; }
@@ -85,6 +88,7 @@ class ConnectionState : public ConnectionToken, public management::Manageable
uint16_t heartbeat;
uint64_t stagingThreshold;
string userId;
+ string url;
bool federationLink;
};
diff --git a/cpp/src/qpid/broker/SessionAdapter.cpp b/cpp/src/qpid/broker/SessionAdapter.cpp
index 64b0b80446..9c23abeba9 100644
--- a/cpp/src/qpid/broker/SessionAdapter.cpp
+++ b/cpp/src/qpid/broker/SessionAdapter.cpp
@@ -26,6 +26,15 @@
#include "qpid/log/Statement.h"
#include "qpid/amqp_0_10/exceptions.h"
#include "qpid/framing/SequenceSet.h"
+#include "qpid/agent/ManagementAgent.h"
+#include "qmf/org/apache/qpid/broker/EventExchangeDeclare.h"
+#include "qmf/org/apache/qpid/broker/EventExchangeDelete.h"
+#include "qmf/org/apache/qpid/broker/EventQueueDeclare.h"
+#include "qmf/org/apache/qpid/broker/EventQueueDelete.h"
+#include "qmf/org/apache/qpid/broker/EventBind.h"
+#include "qmf/org/apache/qpid/broker/EventUnbind.h"
+#include "qmf/org/apache/qpid/broker/EventSubscribe.h"
+#include "qmf/org/apache/qpid/broker/EventUnsubscribe.h"
#include <boost/format.hpp>
#include <boost/cast.hpp>
#include <boost/bind.hpp>
@@ -36,6 +45,8 @@ namespace broker {
using namespace qpid;
using namespace qpid::framing;
using namespace qpid::framing::dtx;
+using namespace qpid::management;
+namespace _qmf = qmf::org::apache::qpid::broker;
typedef std::vector<Queue::shared_ptr> QueueVector;
@@ -54,18 +65,17 @@ void SessionAdapter::ExchangeHandlerImpl::declare(const string& exchange, const
const string& alternateExchange,
bool passive, bool durable, bool /*autoDelete*/, const FieldTable& args){
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
std::map<acl::Property, std::string> params;
- params.insert(make_pair(acl::TYPE, type));
- params.insert(make_pair(acl::ALTERNATE, alternateExchange));
- params.insert(make_pair(acl::PASSIVE, std::string(passive ? "true" : "false") ));
- params.insert(make_pair(acl::DURABLE, std::string(durable ? "true" : "false")));
- if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::EXCHANGE,exchange,&params) )
- throw NotAllowedException("ACL denied exhange declare request");
- }
-
+ params.insert(make_pair(acl::TYPE, type));
+ params.insert(make_pair(acl::ALTERNATE, alternateExchange));
+ params.insert(make_pair(acl::PASSIVE, std::string(passive ? "true" : "false") ));
+ params.insert(make_pair(acl::DURABLE, std::string(durable ? "true" : "false")));
+ if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::EXCHANGE,exchange,&params) )
+ throw NotAllowedException("ACL denied exhange declare request");
+ }
+
//TODO: implement autoDelete
Exchange::shared_ptr alternate;
if (!alternateExchange.empty()) {
@@ -90,6 +100,13 @@ void SessionAdapter::ExchangeHandlerImpl::declare(const string& exchange, const
checkType(response.first, type);
checkAlternate(response.first, alternate);
}
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventExchangeDeclare(getConnection().getUrl(), getConnection().getUserId(), exchange, type,
+ alternateExchange, durable, false, args,
+ response.second ? "created" : "existing"));
+
}catch(UnknownExchangeTypeException& e){
throw CommandInvalidException(QPID_MSG("Exchange type not implemented: " << type));
}
@@ -106,38 +123,37 @@ void SessionAdapter::ExchangeHandlerImpl::checkType(Exchange::shared_ptr exchang
void SessionAdapter::ExchangeHandlerImpl::checkAlternate(Exchange::shared_ptr exchange, Exchange::shared_ptr alternate)
{
if (alternate && alternate != exchange->getAlternate())
- throw NotAllowedException(
- QPID_MSG("Exchange declared with alternate-exchange "
- << exchange->getAlternate()->getName() << ", requested "
- << alternate->getName()));
+ throw NotAllowedException(QPID_MSG("Exchange declared with alternate-exchange "
+ << exchange->getAlternate()->getName() << ", requested "
+ << alternate->getName()));
}
-void SessionAdapter::ExchangeHandlerImpl::delete_(const string& name, bool /*ifUnused*/){
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::DELETE,acl::EXCHANGE,name,NULL) )
- throw NotAllowedException("ACL denied exhange delete request");
+void SessionAdapter::ExchangeHandlerImpl::delete_(const string& name, bool /*ifUnused*/)
+{
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
+ if (!acl->authorise(getConnection().getUserId(),acl::DELETE,acl::EXCHANGE,name,NULL) )
+ throw NotAllowedException("ACL denied exhange delete request");
}
-
//TODO: implement unused
Exchange::shared_ptr exchange(getBroker().getExchanges().get(name));
if (exchange->inUseAsAlternate()) throw NotAllowedException(QPID_MSG("Exchange in use as alternate-exchange."));
if (exchange->isDurable()) getBroker().getStore().destroy(*exchange);
if (exchange->getAlternate()) exchange->getAlternate()->decAlternateUsers();
getBroker().getExchanges().destroy(name);
-}
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventExchangeDelete(getConnection().getUrl(), getConnection().getUserId(), name));
+}
ExchangeQueryResult SessionAdapter::ExchangeHandlerImpl::query(const string& name)
{
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::ACCESS,acl::EXCHANGE,name,NULL) )
- throw NotAllowedException("ACL denied exhange query request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
+ if (!acl->authorise(getConnection().getUserId(),acl::ACCESS,acl::EXCHANGE,name,NULL) )
+ throw NotAllowedException("ACL denied exhange query request");
}
try {
@@ -147,15 +163,15 @@ ExchangeQueryResult SessionAdapter::ExchangeHandlerImpl::query(const string& nam
return ExchangeQueryResult("", false, true, FieldTable());
}
}
+
void SessionAdapter::ExchangeHandlerImpl::bind(const string& queueName,
const string& exchangeName, const string& routingKey,
- const FieldTable& arguments){
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::BIND,acl::EXCHANGE,exchangeName,routingKey) )
- throw NotAllowedException("ACL denied exhange bind request");
+ const FieldTable& arguments)
+{
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
+ if (!acl->authorise(getConnection().getUserId(),acl::BIND,acl::EXCHANGE,exchangeName,routingKey) )
+ throw NotAllowedException("ACL denied exhange bind request");
}
Queue::shared_ptr queue = getQueue(queueName);
@@ -167,30 +183,29 @@ void SessionAdapter::ExchangeHandlerImpl::bind(const string& queueName,
if (exchange->isDurable() && queue->isDurable()) {
getBroker().getStore().bind(*exchange, *queue, routingKey, arguments);
}
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventBind(getConnection().getUrl(), getConnection().getUserId(), exchangeName, queueName, exchangeRoutingKey, arguments));
}
}else{
- throw NotFoundException(
- "Bind failed. No such exchange: " + exchangeName);
+ throw NotFoundException("Bind failed. No such exchange: " + exchangeName);
}
}
-void
-SessionAdapter::ExchangeHandlerImpl::unbind(const string& queueName,
- const string& exchangeName,
- const string& routingKey)
+void SessionAdapter::ExchangeHandlerImpl::unbind(const string& queueName,
+ const string& exchangeName,
+ const string& routingKey)
{
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
std::map<acl::Property, std::string> params;
- params.insert(make_pair(acl::QUEUENAME, queueName));
- params.insert(make_pair(acl::ROUTINGKEY, routingKey));
- if (!acl->authorise(getConnection().getUserId(),acl::UNBIND,acl::EXCHANGE,exchangeName,&params) )
- throw NotAllowedException("ACL denied exchange unbind request");
+ params.insert(make_pair(acl::QUEUENAME, queueName));
+ params.insert(make_pair(acl::ROUTINGKEY, routingKey));
+ if (!acl->authorise(getConnection().getUserId(),acl::UNBIND,acl::EXCHANGE,exchangeName,&params) )
+ throw NotAllowedException("ACL denied exchange unbind request");
}
-
Queue::shared_ptr queue = getQueue(queueName);
if (!queue.get()) throw NotFoundException("Unbind failed. No such exchange: " + exchangeName);
@@ -198,10 +213,14 @@ SessionAdapter::ExchangeHandlerImpl::unbind(const string& queueName,
if (!exchange.get()) throw NotFoundException("Unbind failed. No such exchange: " + exchangeName);
//TODO: revise unbind to rely solely on binding key (not args)
- if (exchange->unbind(queue, routingKey, 0) && exchange->isDurable() && queue->isDurable()) {
- getBroker().getStore().unbind(*exchange, *queue, routingKey, FieldTable());
- }
+ if (exchange->unbind(queue, routingKey, 0)) {
+ if (exchange->isDurable() && queue->isDurable())
+ getBroker().getStore().unbind(*exchange, *queue, routingKey, FieldTable());
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventUnbind(getConnection().getUrl(), getConnection().getUserId(), exchangeName, queueName, routingKey));
+ }
}
ExchangeBoundResult SessionAdapter::ExchangeHandlerImpl::bound(const std::string& exchangeName,
@@ -209,16 +228,15 @@ ExchangeBoundResult SessionAdapter::ExchangeHandlerImpl::bound(const std::string
const std::string& key,
const framing::FieldTable& args)
{
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
std::map<acl::Property, std::string> params;
- params.insert(make_pair(acl::QUEUENAME, queueName));
- params.insert(make_pair(acl::ROUTINGKEY, key));
- if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::EXCHANGE,exchangeName,&params) )
- throw NotAllowedException("ACL denied exhange bound request");
+ params.insert(make_pair(acl::QUEUENAME, queueName));
+ params.insert(make_pair(acl::ROUTINGKEY, key));
+ if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::EXCHANGE,exchangeName,&params) )
+ throw NotAllowedException("ACL denied exhange bound request");
}
-
+
Exchange::shared_ptr exchange;
try {
exchange = getBroker().getExchanges().get(exchangeName);
@@ -279,13 +297,12 @@ bool SessionAdapter::QueueHandlerImpl::isLocal(const ConnectionToken* t) const
QueueQueryResult SessionAdapter::QueueHandlerImpl::query(const string& name)
{
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::ACCESS,acl::QUEUE,name,NULL) )
- throw NotAllowedException("ACL denied queue query request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
+ if (!acl->authorise(getConnection().getUserId(),acl::ACCESS,acl::QUEUE,name,NULL) )
+ throw NotAllowedException("ACL denied queue query request");
}
-
+
Queue::shared_ptr queue = session.getBroker().getQueues().find(name);
if (queue) {
@@ -305,20 +322,19 @@ QueueQueryResult SessionAdapter::QueueHandlerImpl::query(const string& name)
}
void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string& alternateExchange,
- bool passive, bool durable, bool exclusive,
- bool autoDelete, const qpid::framing::FieldTable& arguments){
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
+ bool passive, bool durable, bool exclusive,
+ bool autoDelete, const qpid::framing::FieldTable& arguments)
+{
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
std::map<acl::Property, std::string> params;
- params.insert(make_pair(acl::ALTERNATE, alternateExchange));
- params.insert(make_pair(acl::PASSIVE, std::string(passive ? "true" : "false") ));
- params.insert(make_pair(acl::DURABLE, std::string(durable ? "true" : "false")));
- params.insert(make_pair(acl::EXCLUSIVE, std::string(exclusive ? "true" : "false")));
- params.insert(make_pair(acl::AUTODELETE, std::string(autoDelete ? "true" : "false")));
- if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::QUEUE,name,&params) )
- throw NotAllowedException("ACL denied queue create request");
+ params.insert(make_pair(acl::ALTERNATE, alternateExchange));
+ params.insert(make_pair(acl::PASSIVE, std::string(passive ? "true" : "false") ));
+ params.insert(make_pair(acl::DURABLE, std::string(durable ? "true" : "false")));
+ params.insert(make_pair(acl::EXCLUSIVE, std::string(exclusive ? "true" : "false")));
+ params.insert(make_pair(acl::AUTODELETE, std::string(autoDelete ? "true" : "false")));
+ if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::QUEUE,name,&params) )
+ throw NotAllowedException("ACL denied queue create request");
}
Exchange::shared_ptr alternate;
@@ -327,17 +343,16 @@ void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string&
}
Queue::shared_ptr queue;
if (passive && !name.empty()) {
- queue = getQueue(name);
+ queue = getQueue(name);
//TODO: check alternate-exchange is as expected
} else {
- std::pair<Queue::shared_ptr, bool> queue_created =
- getBroker().getQueues().declare(
- name, durable,
- autoDelete,
- exclusive ? this : 0);
- queue = queue_created.first;
- assert(queue);
- if (queue_created.second) { // This is a new queue
+ std::pair<Queue::shared_ptr, bool> queue_created =
+ getBroker().getQueues().declare(name, durable,
+ autoDelete,
+ exclusive ? this : 0);
+ queue = queue_created.first;
+ assert(queue);
+ if (queue_created.second) { // This is a new queue
if (alternate) {
queue->setAlternateExchange(alternate);
alternate->incAlternateUsers();
@@ -346,44 +361,50 @@ void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string&
//apply settings & create persistent record if required
queue_created.first->create(arguments);
- //add default binding:
- getBroker().getExchanges().getDefault()->bind(queue, name, 0);
+ //add default binding:
+ getBroker().getExchanges().getDefault()->bind(queue, name, 0);
queue->bound(getBroker().getExchanges().getDefault()->getName(), name, arguments);
//handle automatic cleanup:
- if (exclusive) {
- exclusiveQueues.push_back(queue);
- }
- } else {
+ if (exclusive) {
+ exclusiveQueues.push_back(queue);
+ }
+ } else {
if (exclusive && queue->setExclusiveOwner(this)) {
- exclusiveQueues.push_back(queue);
+ exclusiveQueues.push_back(queue);
}
}
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventQueueDeclare(getConnection().getUrl(), getConnection().getUserId(),
+ name, durable, exclusive, autoDelete, arguments,
+ queue_created.second ? "created" : "existing"));
}
+
if (exclusive && !queue->isExclusiveOwner(this))
- throw ResourceLockedException(
- QPID_MSG("Cannot grant exclusive access to queue "
- << queue->getName()));
+ throw ResourceLockedException(QPID_MSG("Cannot grant exclusive access to queue "
+ << queue->getName()));
}
void SessionAdapter::QueueHandlerImpl::purge(const string& queue){
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::PURGE,acl::QUEUE,queue,NULL) )
- throw NotAllowedException("ACL denied queue purge request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl)
+ {
+ if (!acl->authorise(getConnection().getUserId(),acl::PURGE,acl::QUEUE,queue,NULL) )
+ throw NotAllowedException("ACL denied queue purge request");
}
getQueue(queue)->purge();
}
void SessionAdapter::QueueHandlerImpl::delete_(const string& queue, bool ifUnused, bool ifEmpty){
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::DELETE,acl::QUEUE,queue,NULL) )
- throw NotAllowedException("ACL denied queue delete request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl)
+ {
+ if (!acl->authorise(getConnection().getUserId(),acl::DELETE,acl::QUEUE,queue,NULL) )
+ throw NotAllowedException("ACL denied queue delete request");
}
ChannelException error(0, "");
@@ -401,6 +422,10 @@ void SessionAdapter::QueueHandlerImpl::delete_(const string& queue, bool ifUnuse
q->destroy();
getBroker().getQueues().destroy(queue);
q->unbind(getBroker().getExchanges(), q);
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventQueueDelete(getConnection().getUrl(), getConnection().getUserId(), queue));
}
}
@@ -441,12 +466,12 @@ SessionAdapter::MessageHandlerImpl::subscribe(const string& queueName,
const FieldTable& arguments)
{
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- // add flags as needed
- if (!acl->authorise(getConnection().getUserId(),acl::CONSUME,acl::QUEUE,queueName,NULL) )
- throw NotAllowedException("ACL denied Queue subscribe request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl)
+ {
+ // add flags as needed
+ if (!acl->authorise(getConnection().getUserId(),acl::CONSUME,acl::QUEUE,queueName,NULL) )
+ throw NotAllowedException("ACL denied Queue subscribe request");
}
Queue::shared_ptr queue = getQueue(queueName);
@@ -457,12 +482,21 @@ SessionAdapter::MessageHandlerImpl::subscribe(const string& queueName,
state.consume(MessageDelivery::getMessageDeliveryToken(destination, acceptMode, acquireMode),
tag, queue, false, //TODO get rid of no-local
acceptMode == 0, acquireMode == 0, exclusive, &arguments);
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventSubscribe(getConnection().getUrl(), getConnection().getUserId(),
+ queueName, destination, exclusive, arguments));
}
void
SessionAdapter::MessageHandlerImpl::cancel(const string& destination )
{
state.cancel(destination);
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventUnsubscribe(getConnection().getUrl(), getConnection().getUserId(), destination));
}
void
diff --git a/cpp/src/qpid/framing/Buffer.h b/cpp/src/qpid/framing/Buffer.h
index 4f482dc206..828e6e963a 100644
--- a/cpp/src/qpid/framing/Buffer.h
+++ b/cpp/src/qpid/framing/Buffer.h
@@ -28,7 +28,9 @@
namespace qpid {
namespace framing {
-struct OutOfBounds : qpid::Exception {};
+struct OutOfBounds : qpid::Exception {
+ OutOfBounds() : qpid::Exception(std::string("Out of Bounds")) {}
+};
class Content;
class FieldTable;
diff --git a/cpp/src/qpid/management/Manageable.h b/cpp/src/qpid/management/Manageable.h
index afa5c968f7..b4d80d8fad 100644
--- a/cpp/src/qpid/management/Manageable.h
+++ b/cpp/src/qpid/management/Manageable.h
@@ -45,6 +45,7 @@ class Manageable
static const status_t STATUS_INVALID_PARAMETER = 4;
static const status_t STATUS_FEATURE_NOT_IMPLEMENTED = 5;
static const status_t STATUS_FORBIDDEN = 6;
+ static const status_t STATUS_EXCEPTION = 7;
static const status_t STATUS_USER = 0x00010000;
// Every "Manageable" object must hold a reference to exactly one
diff --git a/cpp/src/qpid/management/ManagementBroker.cpp b/cpp/src/qpid/management/ManagementBroker.cpp
index 3ae98e8264..0e046bb813 100644
--- a/cpp/src/qpid/management/ManagementBroker.cpp
+++ b/cpp/src/qpid/management/ManagementBroker.cpp
@@ -179,14 +179,24 @@ void ManagementBroker::setExchange (qpid::broker::Exchange::shared_ptr _mexchang
dExchange = _dexchange;
}
-void ManagementBroker::RegisterClass (string packageName,
- string className,
+void ManagementBroker::registerClass (string& packageName,
+ string& className,
uint8_t* md5Sum,
ManagementObject::writeSchemaCall_t schemaCall)
{
Mutex::ScopedLock lock(userLock);
- PackageMap::iterator pIter = FindOrAddPackageLH(packageName);
- AddClass(pIter, className, md5Sum, schemaCall);
+ PackageMap::iterator pIter = findOrAddPackageLH(packageName);
+ addClassLH(ManagementItem::CLASS_KIND_TABLE, pIter, className, md5Sum, schemaCall);
+}
+
+void ManagementBroker::registerEvent (string& packageName,
+ string& eventName,
+ uint8_t* md5Sum,
+ ManagementObject::writeSchemaCall_t schemaCall)
+{
+ Mutex::ScopedLock lock(userLock);
+ PackageMap::iterator pIter = findOrAddPackageLH(packageName);
+ addClassLH(ManagementItem::CLASS_KIND_EVENT, pIter, eventName, md5Sum, schemaCall);
}
ObjectId ManagementBroker::addObject (ManagementObject* object,
@@ -211,6 +221,23 @@ ObjectId ManagementBroker::addObject (ManagementObject* object,
return objId;
}
+void ManagementBroker::raiseEvent(const ManagementEvent& event)
+{
+ Mutex::ScopedLock lock (userLock);
+ Buffer outBuffer(eventBuffer, MA_BUFFER_SIZE);
+ uint32_t outLen;
+
+ encodeHeader(outBuffer, 'e');
+ outBuffer.putShortString(event.getPackageName());
+ outBuffer.putShortString(event.getEventName());
+ outBuffer.putBin128(event.getMd5Sum());
+ outBuffer.putLongLong(uint64_t(Duration(now())));
+ event.encode(outBuffer);
+ outLen = MA_BUFFER_SIZE - outBuffer.available();
+ outBuffer.reset();
+ sendBuffer(outBuffer, outLen, mExchange, "mgmt.event");
+}
+
ManagementBroker::Periodic::Periodic (ManagementBroker& _broker, uint32_t _seconds)
: TimerTask (qpid::sys::Duration ((_seconds ? _seconds : 1) * qpid::sys::TIME_SEC)), broker(_broker) {}
@@ -219,7 +246,7 @@ ManagementBroker::Periodic::~Periodic () {}
void ManagementBroker::Periodic::fire ()
{
broker.timer.add (intrusive_ptr<TimerTask> (new Periodic (broker, broker.interval)));
- broker.PeriodicProcessing ();
+ broker.periodicProcessing ();
}
void ManagementBroker::clientAdded (void)
@@ -233,35 +260,35 @@ void ManagementBroker::clientAdded (void)
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader (outBuffer, 'x');
+ encodeHeader (outBuffer, 'x');
outLen = MA_BUFFER_SIZE - outBuffer.available ();
outBuffer.reset ();
- SendBuffer (outBuffer, outLen, dExchange, aIter->second->routingKey);
+ sendBuffer (outBuffer, outLen, dExchange, aIter->second->routingKey);
}
}
-void ManagementBroker::EncodeHeader (Buffer& buf, uint8_t opcode, uint32_t seq)
+void ManagementBroker::encodeHeader (Buffer& buf, uint8_t opcode, uint32_t seq)
{
buf.putOctet ('A');
buf.putOctet ('M');
- buf.putOctet ('1');
+ buf.putOctet ('2');
buf.putOctet (opcode);
buf.putLong (seq);
}
-bool ManagementBroker::CheckHeader (Buffer& buf, uint8_t *opcode, uint32_t *seq)
+bool ManagementBroker::checkHeader (Buffer& buf, uint8_t *opcode, uint32_t *seq)
{
- uint8_t h1 = buf.getOctet ();
- uint8_t h2 = buf.getOctet ();
- uint8_t h3 = buf.getOctet ();
+ uint8_t h1 = buf.getOctet();
+ uint8_t h2 = buf.getOctet();
+ uint8_t h3 = buf.getOctet();
- *opcode = buf.getOctet ();
- *seq = buf.getLong ();
+ *opcode = buf.getOctet();
+ *seq = buf.getLong();
- return h1 == 'A' && h2 == 'M' && h3 == '1';
+ return h1 == 'A' && h2 == 'M' && h3 == '2';
}
-void ManagementBroker::SendBuffer (Buffer& buf,
+void ManagementBroker::sendBuffer (Buffer& buf,
uint32_t length,
qpid::broker::Exchange::shared_ptr exchange,
string routingKey)
@@ -304,7 +331,7 @@ void ManagementBroker::moveNewObjectsLH()
newManagementObjects.clear();
}
-void ManagementBroker::PeriodicProcessing (void)
+void ManagementBroker::periodicProcessing (void)
{
#define BUFSIZE 65536
Mutex::ScopedLock lock (userLock);
@@ -315,13 +342,13 @@ void ManagementBroker::PeriodicProcessing (void)
{
Buffer msgBuffer(msgChars, BUFSIZE);
- EncodeHeader(msgBuffer, 'h');
+ encodeHeader(msgBuffer, 'h');
msgBuffer.putLongLong(uint64_t(Duration(now())));
contentSize = BUFSIZE - msgBuffer.available ();
msgBuffer.reset ();
routingKey = "mgmt." + uuid.str() + ".heartbeat";
- SendBuffer (msgBuffer, contentSize, mExchange, routingKey);
+ sendBuffer (msgBuffer, contentSize, mExchange, routingKey);
}
moveNewObjectsLH();
@@ -350,25 +377,25 @@ void ManagementBroker::PeriodicProcessing (void)
if (object->getConfigChanged () || object->isDeleted ())
{
Buffer msgBuffer (msgChars, BUFSIZE);
- EncodeHeader (msgBuffer, 'c');
+ encodeHeader (msgBuffer, 'c');
object->writeProperties(msgBuffer);
contentSize = BUFSIZE - msgBuffer.available ();
msgBuffer.reset ();
routingKey = "mgmt." + uuid.str() + ".prop." + object->getClassName ();
- SendBuffer (msgBuffer, contentSize, mExchange, routingKey);
+ sendBuffer (msgBuffer, contentSize, mExchange, routingKey);
}
if (object->getInstChanged ())
{
Buffer msgBuffer (msgChars, BUFSIZE);
- EncodeHeader (msgBuffer, 'i');
+ encodeHeader (msgBuffer, 'i');
object->writeStatistics(msgBuffer);
contentSize = BUFSIZE - msgBuffer.available ();
msgBuffer.reset ();
routingKey = "mgmt." + uuid.str () + ".stat." + object->getClassName ();
- SendBuffer (msgBuffer, contentSize, mExchange, routingKey);
+ sendBuffer (msgBuffer, contentSize, mExchange, routingKey);
}
if (object->isDeleted ())
@@ -393,12 +420,12 @@ void ManagementBroker::sendCommandComplete (string replyToKey, uint32_t sequence
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader (outBuffer, 'z', sequence);
+ encodeHeader (outBuffer, 'z', sequence);
outBuffer.putLong (code);
outBuffer.putShortString (text);
outLen = MA_BUFFER_SIZE - outBuffer.available ();
outBuffer.reset ();
- SendBuffer (outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer (outBuffer, outLen, dExchange, replyToKey);
}
bool ManagementBroker::dispatchCommand (Deliverable& deliverable,
@@ -411,7 +438,7 @@ bool ManagementBroker::dispatchCommand (Deliverable& deliverable,
// Parse the routing key. This management broker should act as though it
// is bound to the exchange to match the following keys:
//
- // agent.0.#
+ // agent.1.0.#
// broker
if (routingKey == "broker") {
@@ -419,12 +446,12 @@ bool ManagementBroker::dispatchCommand (Deliverable& deliverable,
return false;
}
- else if (routingKey.compare(0, 7, "agent.0") == 0) {
+ else if (routingKey.compare(0, 9, "agent.1.0") == 0) {
dispatchAgentCommandLH(msg);
return false;
}
- else if (routingKey.compare(0, 6, "agent.") == 0) {
+ else if (routingKey.compare(0, 8, "agent.1.") == 0) {
return authorizeAgentMessageLH(msg);
}
@@ -447,7 +474,7 @@ void ManagementBroker::handleMethodRequestLH (Buffer& inBuffer, string replyToKe
inBuffer.getShortString(className);
inBuffer.getBin128(hash);
inBuffer.getShortString(methodName);
- EncodeHeader(outBuffer, 'm', sequence);
+ encodeHeader(outBuffer, 'm', sequence);
if (acl != 0) {
string userId = ((const qpid::broker::ConnectionState*) connToken)->getUserId();
@@ -460,7 +487,7 @@ void ManagementBroker::handleMethodRequestLH (Buffer& inBuffer, string replyToKe
outBuffer.putShortString(Manageable::StatusText(Manageable::STATUS_FORBIDDEN));
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
- SendBuffer(outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer(outBuffer, outLen, dExchange, replyToKey);
return;
}
}
@@ -476,12 +503,19 @@ void ManagementBroker::handleMethodRequestLH (Buffer& inBuffer, string replyToKe
outBuffer.putShortString (Manageable::StatusText (Manageable::STATUS_INVALID_PARAMETER));
}
else
- iter->second->doMethod(methodName, inBuffer, outBuffer);
+ try {
+ outBuffer.record();
+ iter->second->doMethod(methodName, inBuffer, outBuffer);
+ } catch(std::exception& e) {
+ outBuffer.restore();
+ outBuffer.putLong(Manageable::STATUS_EXCEPTION);
+ outBuffer.putShortString(e.what());
+ }
}
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
- SendBuffer(outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer(outBuffer, outLen, dExchange, replyToKey);
}
void ManagementBroker::handleBrokerRequestLH (Buffer&, string replyToKey, uint32_t sequence)
@@ -489,12 +523,12 @@ void ManagementBroker::handleBrokerRequestLH (Buffer&, string replyToKey, uint32
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader (outBuffer, 'b', sequence);
+ encodeHeader (outBuffer, 'b', sequence);
uuid.encode (outBuffer);
outLen = MA_BUFFER_SIZE - outBuffer.available ();
outBuffer.reset ();
- SendBuffer (outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer (outBuffer, outLen, dExchange, replyToKey);
}
void ManagementBroker::handlePackageQueryLH (Buffer&, string replyToKey, uint32_t sequence)
@@ -506,11 +540,11 @@ void ManagementBroker::handlePackageQueryLH (Buffer&, string replyToKey, uint32_
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader (outBuffer, 'p', sequence);
- EncodePackageIndication (outBuffer, pIter);
+ encodeHeader (outBuffer, 'p', sequence);
+ encodePackageIndication (outBuffer, pIter);
outLen = MA_BUFFER_SIZE - outBuffer.available ();
outBuffer.reset ();
- SendBuffer (outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer (outBuffer, outLen, dExchange, replyToKey);
}
sendCommandComplete (replyToKey, sequence);
@@ -521,7 +555,7 @@ void ManagementBroker::handlePackageIndLH (Buffer& inBuffer, string /*replyToKey
std::string packageName;
inBuffer.getShortString(packageName);
- FindOrAddPackageLH(packageName);
+ findOrAddPackageLH(packageName);
}
void ManagementBroker::handleClassQueryLH(Buffer& inBuffer, string replyToKey, uint32_t sequence)
@@ -542,11 +576,11 @@ void ManagementBroker::handleClassQueryLH(Buffer& inBuffer, string replyToKey, u
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'q', sequence);
- EncodeClassIndication(outBuffer, pIter, cIter);
+ encodeHeader(outBuffer, 'q', sequence);
+ encodeClassIndication(outBuffer, pIter, cIter);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
- SendBuffer(outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer(outBuffer, outLen, dExchange, replyToKey);
}
}
}
@@ -558,26 +592,27 @@ void ManagementBroker::handleClassIndLH (Buffer& inBuffer, string replyToKey, ui
std::string packageName;
SchemaClassKey key;
+ uint8_t kind = inBuffer.getOctet();
inBuffer.getShortString(packageName);
inBuffer.getShortString(key.name);
inBuffer.getBin128(key.hash);
- PackageMap::iterator pIter = FindOrAddPackageLH(packageName);
+ PackageMap::iterator pIter = findOrAddPackageLH(packageName);
ClassMap::iterator cIter = pIter->second.find(key);
if (cIter == pIter->second.end()) {
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
uint32_t sequence = nextRequestSequence++;
- EncodeHeader (outBuffer, 'S', sequence);
+ encodeHeader (outBuffer, 'S', sequence);
outBuffer.putShortString(packageName);
outBuffer.putShortString(key.name);
outBuffer.putBin128(key.hash);
outLen = MA_BUFFER_SIZE - outBuffer.available ();
outBuffer.reset ();
- SendBuffer (outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer (outBuffer, outLen, dExchange, replyToKey);
- pIter->second.insert(pair<SchemaClassKey, SchemaClass>(key, SchemaClass(sequence)));
+ pIter->second.insert(pair<SchemaClassKey, SchemaClass>(key, SchemaClass(kind, sequence)));
}
}
@@ -612,11 +647,11 @@ void ManagementBroker::handleSchemaRequestLH(Buffer& inBuffer, string replyToKey
SchemaClass& classInfo = cIter->second;
if (classInfo.hasSchema()) {
- EncodeHeader(outBuffer, 's', sequence);
+ encodeHeader(outBuffer, 's', sequence);
classInfo.appendSchema(outBuffer);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
- SendBuffer(outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer(outBuffer, outLen, dExchange, replyToKey);
}
else
sendCommandComplete(replyToKey, sequence, 1, "Schema not available");
@@ -634,9 +669,10 @@ void ManagementBroker::handleSchemaResponseLH(Buffer& inBuffer, string /*replyTo
SchemaClassKey key;
inBuffer.record();
- inBuffer.getShortString (packageName);
- inBuffer.getShortString (key.name);
- inBuffer.getBin128 (key.hash);
+ inBuffer.getOctet();
+ inBuffer.getShortString(packageName);
+ inBuffer.getShortString(key.name);
+ inBuffer.getBin128(key.hash);
inBuffer.restore();
PackageMap::iterator pIter = packages.find(packageName);
@@ -644,7 +680,7 @@ void ManagementBroker::handleSchemaResponseLH(Buffer& inBuffer, string /*replyTo
ClassMap& cMap = pIter->second;
ClassMap::iterator cIter = cMap.find(key);
if (cIter != cMap.end() && cIter->second.pendingSequence == sequence) {
- size_t length = ValidateSchema(inBuffer);
+ size_t length = validateSchema(inBuffer, cIter->second.kind);
if (length == 0) {
QPID_LOG(warning, "Management Broker received invalid schema response: " << packageName << "." << key.name);
cMap.erase(key);
@@ -658,11 +694,11 @@ void ManagementBroker::handleSchemaResponseLH(Buffer& inBuffer, string /*replyTo
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'q');
- EncodeClassIndication(outBuffer, pIter, cIter);
+ encodeHeader(outBuffer, 'q');
+ encodeClassIndication(outBuffer, pIter, cIter);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
- SendBuffer(outBuffer, outLen, mExchange, "mgmt." + uuid.str() + ".schema");
+ sendBuffer(outBuffer, outLen, mExchange, "mgmt." + uuid.str() + ".schema");
}
}
}
@@ -727,7 +763,7 @@ void ManagementBroker::deleteOrphanedAgentsLH()
void ManagementBroker::handleAttachRequestLH (Buffer& inBuffer, string replyToKey, uint32_t sequence, const ConnectionToken* connToken)
{
string label;
- uint32_t requestedBank;
+ uint32_t requestedBrokerBank, requestedAgentBank;
uint32_t assignedBank;
ObjectId connectionRef = ((const ConnectionState*) connToken)->GetManagementObject()->getObjectId();
Uuid systemId;
@@ -737,14 +773,15 @@ void ManagementBroker::handleAttachRequestLH (Buffer& inBuffer, string replyToKe
RemoteAgentMap::iterator aIter = remoteAgents.find(connectionRef);
if (aIter != remoteAgents.end()) {
// There already exists an agent on this session. Reject the request.
- sendCommandComplete (replyToKey, sequence, 1, "Connection already has remote agent");
+ sendCommandComplete(replyToKey, sequence, 1, "Connection already has remote agent");
return;
}
- inBuffer.getShortString (label);
- systemId.decode (inBuffer);
- requestedBank = inBuffer.getLong ();
- assignedBank = assignBankLH (requestedBank);
+ inBuffer.getShortString(label);
+ systemId.decode(inBuffer);
+ requestedBrokerBank = inBuffer.getLong();
+ requestedAgentBank = inBuffer.getLong();
+ assignedBank = assignBankLH(requestedAgentBank);
RemoteAgent* agent = new RemoteAgent;
agent->objIdBank = assignedBank;
@@ -755,7 +792,8 @@ void ManagementBroker::handleAttachRequestLH (Buffer& inBuffer, string replyToKe
agent->mgmtObject->set_label (label);
agent->mgmtObject->set_registeredTo (broker->GetManagementObject()->getObjectId());
agent->mgmtObject->set_systemId (systemId);
- agent->mgmtObject->set_objectIdBank (assignedBank);
+ agent->mgmtObject->set_brokerBank (brokerBank);
+ agent->mgmtObject->set_agentBank (assignedBank);
addObject (agent->mgmtObject);
remoteAgents[connectionRef] = agent;
@@ -764,12 +802,12 @@ void ManagementBroker::handleAttachRequestLH (Buffer& inBuffer, string replyToKe
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader (outBuffer, 'a', sequence);
+ encodeHeader (outBuffer, 'a', sequence);
outBuffer.putLong (brokerBank);
outBuffer.putLong (assignedBank);
outLen = MA_BUFFER_SIZE - outBuffer.available ();
outBuffer.reset ();
- SendBuffer (outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer (outBuffer, outLen, dExchange, replyToKey);
}
void ManagementBroker::handleGetQueryLH (Buffer& inBuffer, string replyToKey, uint32_t sequence)
@@ -799,12 +837,12 @@ void ManagementBroker::handleGetQueryLH (Buffer& inBuffer, string replyToKey, ui
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader (outBuffer, 'g', sequence);
+ encodeHeader (outBuffer, 'g', sequence);
object->writeProperties(outBuffer);
object->writeStatistics(outBuffer, true);
outLen = MA_BUFFER_SIZE - outBuffer.available ();
outBuffer.reset ();
- SendBuffer (outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer (outBuffer, outLen, dExchange, replyToKey);
}
}
@@ -824,7 +862,7 @@ bool ManagementBroker::authorizeAgentMessageLH(Message& msg)
msg.encodeContent(inBuffer);
inBuffer.reset();
- if (!CheckHeader(inBuffer, &opcode, &sequence))
+ if (!checkHeader(inBuffer, &opcode, &sequence))
return false;
if (opcode == 'M') {
@@ -861,12 +899,12 @@ bool ManagementBroker::authorizeAgentMessageLH(Message& msg)
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'm', sequence);
+ encodeHeader(outBuffer, 'm', sequence);
outBuffer.putLong(Manageable::STATUS_FORBIDDEN);
outBuffer.putShortString(Manageable::StatusText(Manageable::STATUS_FORBIDDEN));
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
- SendBuffer(outBuffer, outLen, dExchange, replyToKey);
+ sendBuffer(outBuffer, outLen, dExchange, replyToKey);
}
return false;
@@ -900,7 +938,7 @@ void ManagementBroker::dispatchAgentCommandLH(Message& msg)
msg.encodeContent(inBuffer);
inBuffer.reset();
- if (!CheckHeader(inBuffer, &opcode, &sequence))
+ if (!checkHeader(inBuffer, &opcode, &sequence))
return;
if (opcode == 'B') handleBrokerRequestLH (inBuffer, replyToKey, sequence);
@@ -915,7 +953,7 @@ void ManagementBroker::dispatchAgentCommandLH(Message& msg)
else if (opcode == 'M') handleMethodRequestLH (inBuffer, replyToKey, sequence, msg.getPublisher());
}
-ManagementBroker::PackageMap::iterator ManagementBroker::FindOrAddPackageLH(std::string name)
+ManagementBroker::PackageMap::iterator ManagementBroker::findOrAddPackageLH(std::string name)
{
PackageMap::iterator pIter = packages.find (name);
if (pIter != packages.end ())
@@ -930,19 +968,20 @@ ManagementBroker::PackageMap::iterator ManagementBroker::FindOrAddPackageLH(std:
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader (outBuffer, 'p');
- EncodePackageIndication (outBuffer, result.first);
+ encodeHeader (outBuffer, 'p');
+ encodePackageIndication (outBuffer, result.first);
outLen = MA_BUFFER_SIZE - outBuffer.available ();
outBuffer.reset ();
- SendBuffer (outBuffer, outLen, mExchange, "mgmt." + uuid.str() + ".schema.package");
+ sendBuffer (outBuffer, outLen, mExchange, "mgmt." + uuid.str() + ".schema.package");
return result.first;
}
-void ManagementBroker::AddClass(PackageMap::iterator pIter,
- string className,
- uint8_t* md5Sum,
- ManagementObject::writeSchemaCall_t schemaCall)
+void ManagementBroker::addClassLH(uint8_t kind,
+ PackageMap::iterator pIter,
+ string& className,
+ uint8_t* md5Sum,
+ ManagementObject::writeSchemaCall_t schemaCall)
{
SchemaClassKey key;
ClassMap& cMap = pIter->second;
@@ -958,71 +997,76 @@ void ManagementBroker::AddClass(PackageMap::iterator pIter,
QPID_LOG (debug, "ManagementBroker added class " << pIter->first << "." <<
key.name);
- cMap.insert(pair<SchemaClassKey, SchemaClass>(key, SchemaClass(schemaCall)));
+ cMap.insert(pair<SchemaClassKey, SchemaClass>(key, SchemaClass(kind, schemaCall)));
cIter = cMap.find(key);
}
-void ManagementBroker::EncodePackageIndication (Buffer& buf,
- PackageMap::iterator pIter)
+void ManagementBroker::encodePackageIndication(Buffer& buf,
+ PackageMap::iterator pIter)
{
- buf.putShortString ((*pIter).first);
+ buf.putShortString((*pIter).first);
}
-void ManagementBroker::EncodeClassIndication (Buffer& buf,
- PackageMap::iterator pIter,
- ClassMap::iterator cIter)
+void ManagementBroker::encodeClassIndication(Buffer& buf,
+ PackageMap::iterator pIter,
+ ClassMap::iterator cIter)
{
SchemaClassKey key = (*cIter).first;
- buf.putShortString ((*pIter).first);
- buf.putShortString (key.name);
- buf.putBin128 (key.hash);
+ buf.putOctet((*cIter).second.kind);
+ buf.putShortString((*pIter).first);
+ buf.putShortString(key.name);
+ buf.putBin128(key.hash);
+}
+
+size_t ManagementBroker::validateSchema(Buffer& inBuffer, uint8_t kind)
+{
+ if (kind == ManagementItem::CLASS_KIND_TABLE)
+ return validateTableSchema(inBuffer);
+ else if (kind == ManagementItem::CLASS_KIND_EVENT)
+ return validateEventSchema(inBuffer);
+ return 0;
}
-size_t ManagementBroker::ValidateSchema(Buffer& inBuffer)
+size_t ManagementBroker::validateTableSchema(Buffer& inBuffer)
{
uint32_t start = inBuffer.getPosition();
uint32_t end;
string text;
uint8_t hash[16];
- inBuffer.record();
- inBuffer.getShortString(text);
- inBuffer.getShortString(text);
- inBuffer.getBin128(hash);
+ try {
+ inBuffer.record();
+ uint8_t kind = inBuffer.getOctet();
+ if (kind != ManagementItem::CLASS_KIND_TABLE)
+ return 0;
- uint16_t propCount = inBuffer.getShort();
- uint16_t statCount = inBuffer.getShort();
- uint16_t methCount = inBuffer.getShort();
- uint16_t evntCount = inBuffer.getShort();
+ inBuffer.getShortString(text);
+ inBuffer.getShortString(text);
+ inBuffer.getBin128(hash);
- for (uint16_t idx = 0; idx < propCount + statCount; idx++) {
- FieldTable ft;
- ft.decode(inBuffer);
- }
+ uint16_t propCount = inBuffer.getShort();
+ uint16_t statCount = inBuffer.getShort();
+ uint16_t methCount = inBuffer.getShort();
- for (uint16_t idx = 0; idx < methCount; idx++) {
- FieldTable ft;
- ft.decode(inBuffer);
- if (!ft.isSet("argCount"))
- return 0;
- int argCount = ft.getInt("argCount");
- for (int mIdx = 0; mIdx < argCount; mIdx++) {
- FieldTable aft;
- aft.decode(inBuffer);
+ for (uint16_t idx = 0; idx < propCount + statCount; idx++) {
+ FieldTable ft;
+ ft.decode(inBuffer);
}
- }
- for (uint16_t idx = 0; idx < evntCount; idx++) {
- FieldTable ft;
- ft.decode(inBuffer);
- if (!ft.isSet("argCount"))
- return 0;
- int argCount = ft.getInt("argCount");
- for (int mIdx = 0; mIdx < argCount; mIdx++) {
- FieldTable aft;
- aft.decode(inBuffer);
+ for (uint16_t idx = 0; idx < methCount; idx++) {
+ FieldTable ft;
+ ft.decode(inBuffer);
+ if (!ft.isSet("argCount"))
+ return 0;
+ int argCount = ft.getInt("argCount");
+ for (int mIdx = 0; mIdx < argCount; mIdx++) {
+ FieldTable aft;
+ aft.decode(inBuffer);
+ }
}
+ } catch (std::exception& e) {
+ return 0;
}
end = inBuffer.getPosition();
@@ -1030,24 +1074,34 @@ size_t ManagementBroker::ValidateSchema(Buffer& inBuffer)
return end - start;
}
-Mutex& ManagementBroker::getMutex()
+size_t ManagementBroker::validateEventSchema(Buffer& inBuffer)
{
- return userLock;
-}
+ uint32_t start = inBuffer.getPosition();
+ uint32_t end;
+ string text;
+ uint8_t hash[16];
-Buffer* ManagementBroker::startEventLH()
-{
- Buffer* outBuffer(new Buffer(eventBuffer, MA_BUFFER_SIZE));
- EncodeHeader(*outBuffer, 'e');
- outBuffer->putLongLong(uint64_t(Duration(now())));
- return outBuffer;
-}
+ try {
+ inBuffer.record();
+ uint8_t kind = inBuffer.getOctet();
+ if (kind != ManagementItem::CLASS_KIND_EVENT)
+ return 0;
-void ManagementBroker::finishEventLH(Buffer* outBuffer)
-{
- uint32_t outLen = MA_BUFFER_SIZE - outBuffer->available();
- outBuffer->reset();
- SendBuffer(*outBuffer, outLen, mExchange, "mgmt.event");
- delete outBuffer;
-}
+ inBuffer.getShortString(text);
+ inBuffer.getShortString(text);
+ inBuffer.getBin128(hash);
+
+ uint16_t argCount = inBuffer.getShort();
+ for (uint16_t idx = 0; idx < argCount; idx++) {
+ FieldTable ft;
+ ft.decode(inBuffer);
+ }
+ } catch (std::exception& e) {
+ return 0;
+ }
+
+ end = inBuffer.getPosition();
+ inBuffer.restore(); // restore original position
+ return end - start;
+}
diff --git a/cpp/src/qpid/management/ManagementBroker.h b/cpp/src/qpid/management/ManagementBroker.h
index 23fba74b83..c0e0c50963 100644
--- a/cpp/src/qpid/management/ManagementBroker.h
+++ b/cpp/src/qpid/management/ManagementBroker.h
@@ -38,11 +38,11 @@ namespace management {
class ManagementBroker : public ManagementAgent
{
- private:
+private:
int threadPoolSize;
- public:
+public:
ManagementBroker ();
virtual ~ManagementBroker ();
@@ -52,13 +52,18 @@ class ManagementBroker : public ManagementAgent
void setExchange (qpid::broker::Exchange::shared_ptr mgmtExchange,
qpid::broker::Exchange::shared_ptr directExchange);
int getMaxThreads () { return threadPoolSize; }
- void RegisterClass (std::string packageName,
- std::string className,
+ void registerClass (std::string& packageName,
+ std::string& className,
+ uint8_t* md5Sum,
+ ManagementObject::writeSchemaCall_t schemaCall);
+ void registerEvent (std::string& packageName,
+ std::string& eventName,
uint8_t* md5Sum,
ManagementObject::writeSchemaCall_t schemaCall);
ObjectId addObject (ManagementObject* object,
uint64_t persistId = 0);
- void clientAdded (void);
+ void raiseEvent(const ManagementEvent& event);
+ void clientAdded ();
bool dispatchCommand (qpid::broker::Deliverable& msg,
const std::string& routingKey,
const framing::FieldTable* args);
@@ -68,7 +73,7 @@ class ManagementBroker : public ManagementAgent
uint32_t pollCallbacks (uint32_t) { assert(0); return 0; }
int getSignalFd () { assert(0); return -1; }
- private:
+private:
friend class ManagementAgent;
struct Periodic : public qpid::broker::TimerTask
@@ -127,15 +132,16 @@ class ManagementBroker : public ManagementAgent
struct SchemaClass
{
+ uint8_t kind;
ManagementObject::writeSchemaCall_t writeSchemaCall;
uint32_t pendingSequence;
size_t bufferLen;
uint8_t* buffer;
- SchemaClass(uint32_t seq) :
- writeSchemaCall(0), pendingSequence(seq), bufferLen(0), buffer(0) {}
- SchemaClass(ManagementObject::writeSchemaCall_t call) :
- writeSchemaCall(call), pendingSequence(0), bufferLen(0), buffer(0) {}
+ SchemaClass(uint8_t _kind, uint32_t seq) :
+ kind(_kind), writeSchemaCall(0), pendingSequence(seq), bufferLen(0), buffer(0) {}
+ SchemaClass(uint8_t _kind, ManagementObject::writeSchemaCall_t call) :
+ kind(_kind), writeSchemaCall(call), pendingSequence(0), bufferLen(0), buffer(0) {}
bool hasSchema () { return (writeSchemaCall != 0) || (buffer != 0); }
void appendSchema (framing::Buffer& buf);
};
@@ -154,12 +160,12 @@ class ManagementBroker : public ManagementAgent
framing::Uuid uuid;
sys::Mutex addLock;
sys::Mutex userLock;
- qpid::broker::Timer timer;
+ qpid::broker::Timer timer;
qpid::broker::Exchange::shared_ptr mExchange;
qpid::broker::Exchange::shared_ptr dExchange;
std::string dataDir;
uint16_t interval;
- qpid::broker::Broker* broker;
+ qpid::broker::Broker* broker;
uint16_t bootSequence;
uint32_t nextObjectId;
uint32_t brokerBank;
@@ -173,10 +179,10 @@ class ManagementBroker : public ManagementAgent
char eventBuffer[MA_BUFFER_SIZE];
void writeData ();
- void PeriodicProcessing (void);
- void EncodeHeader (framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0);
- bool CheckHeader (framing::Buffer& buf, uint8_t *opcode, uint32_t *seq);
- void SendBuffer (framing::Buffer& buf,
+ void periodicProcessing (void);
+ void encodeHeader (framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0);
+ bool checkHeader (framing::Buffer& buf, uint8_t *opcode, uint32_t *seq);
+ void sendBuffer (framing::Buffer& buf,
uint32_t length,
qpid::broker::Exchange::shared_ptr exchange,
std::string routingKey);
@@ -185,14 +191,15 @@ class ManagementBroker : public ManagementAgent
bool authorizeAgentMessageLH(qpid::broker::Message& msg);
void dispatchAgentCommandLH(qpid::broker::Message& msg);
- PackageMap::iterator FindOrAddPackageLH(std::string name);
- void AddClass(PackageMap::iterator pIter,
- std::string className,
- uint8_t* md5Sum,
- ManagementObject::writeSchemaCall_t schemaCall);
- void EncodePackageIndication (framing::Buffer& buf,
+ PackageMap::iterator findOrAddPackageLH(std::string name);
+ void addClassLH(uint8_t kind,
+ PackageMap::iterator pIter,
+ std::string& className,
+ uint8_t* md5Sum,
+ ManagementObject::writeSchemaCall_t schemaCall);
+ void encodePackageIndication (framing::Buffer& buf,
PackageMap::iterator pIter);
- void EncodeClassIndication (framing::Buffer& buf,
+ void encodeClassIndication (framing::Buffer& buf,
PackageMap::iterator pIter,
ClassMap::iterator cIter);
bool bankInUse (uint32_t bank);
@@ -212,10 +219,9 @@ class ManagementBroker : public ManagementAgent
void handleGetQueryLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence);
void handleMethodRequestLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence, const qpid::broker::ConnectionToken* connToken);
- size_t ValidateSchema(framing::Buffer&);
- sys::Mutex& getMutex();
- framing::Buffer* startEventLH();
- void finishEventLH(framing::Buffer* outBuffer);
+ size_t validateSchema(framing::Buffer&, uint8_t kind);
+ size_t validateTableSchema(framing::Buffer&);
+ size_t validateEventSchema(framing::Buffer&);
};
}}
diff --git a/cpp/src/qpid/management/ManagementEvent.h b/cpp/src/qpid/management/ManagementEvent.h
new file mode 100644
index 0000000000..d5a47e9144
--- /dev/null
+++ b/cpp/src/qpid/management/ManagementEvent.h
@@ -0,0 +1,48 @@
+#ifndef _ManagementEvent_
+#define _ManagementEvent_
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "ManagementObject.h"
+#include <qpid/framing/Buffer.h>
+#include <string>
+
+namespace qpid {
+namespace management {
+
+class ManagementAgent;
+
+class ManagementEvent : public ManagementItem {
+public:
+ typedef void (*writeSchemaCall_t)(qpid::framing::Buffer&);
+ virtual ~ManagementEvent() {}
+
+ virtual writeSchemaCall_t getWriteSchemaCall(void) = 0;
+ virtual std::string& getEventName() const = 0;
+ virtual std::string& getPackageName() const = 0;
+ virtual uint8_t* getMd5Sum() const = 0;
+ virtual void encode(qpid::framing::Buffer&) const = 0;
+};
+
+}}
+
+#endif /*!_ManagementEvent_*/
diff --git a/cpp/src/qpid/management/ManagementObject.cpp b/cpp/src/qpid/management/ManagementObject.cpp
index e0386ee057..ce65ae3279 100644
--- a/cpp/src/qpid/management/ManagementObject.cpp
+++ b/cpp/src/qpid/management/ManagementObject.cpp
@@ -84,6 +84,21 @@ void ObjectId::decode(framing::Buffer& buffer)
second = buffer.getLongLong();
}
+namespace qpid {
+namespace management {
+
+std::ostream& operator<<(std::ostream& out, const ObjectId& i)
+{
+ out << "[" << ((i.first & 0xF000000000000000LL) >> 60) <<
+ "-" << ((i.first & 0x0FFF000000000000LL) >> 48) <<
+ "-" << ((i.first & 0x0000FFFFF0000000LL) >> 32) <<
+ "-" << (i.first & 0x000000000FFFFFFFLL) <<
+ "-" << i.second << "]";
+ return out;
+}
+
+}}
+
int ManagementObject::nextThreadIndex = 0;
void ManagementObject::writeTimestamps (Buffer& buf)
@@ -109,18 +124,3 @@ int ManagementObject::getThreadIndex() {
}
return thisIndex;
}
-
-Mutex& ManagementObject::getMutex()
-{
- return agent->getMutex();
-}
-
-Buffer* ManagementObject::startEventLH()
-{
- return agent->startEventLH();
-}
-
-void ManagementObject::finishEventLH(Buffer* buf)
-{
- agent->finishEventLH(buf);
-}
diff --git a/cpp/src/qpid/management/ManagementObject.h b/cpp/src/qpid/management/ManagementObject.h
index 1b809f5125..3778d66b7e 100644
--- a/cpp/src/qpid/management/ManagementObject.h
+++ b/cpp/src/qpid/management/ManagementObject.h
@@ -46,7 +46,7 @@ public:
class ObjectId {
-private:
+protected:
const AgentAttachment* agent;
uint64_t first;
uint64_t second;
@@ -59,23 +59,11 @@ public:
bool operator<(const ObjectId &other) const;
void encode(framing::Buffer& buffer);
void decode(framing::Buffer& buffer);
+ friend std::ostream& operator<<(std::ostream&, const ObjectId&);
};
-class ManagementObject
-{
- protected:
-
- uint64_t createTime;
- uint64_t destroyTime;
- ObjectId objectId;
- bool configChanged;
- bool instChanged;
- bool deleted;
- Manageable* coreObject;
- sys::Mutex accessLock;
- ManagementAgent* agent;
- int maxThreads;
-
+class ManagementItem {
+public:
static const uint8_t TYPE_U8 = 1;
static const uint8_t TYPE_U16 = 2;
static const uint8_t TYPE_U32 = 3;
@@ -107,15 +95,35 @@ class ManagementObject
static const uint8_t FLAG_INDEX = 0x02;
static const uint8_t FLAG_END = 0x80;
- static int nextThreadIndex;
+ const static uint8_t CLASS_KIND_TABLE = 1;
+ const static uint8_t CLASS_KIND_EVENT = 2;
+
+
+
+public:
+ virtual ~ManagementItem() {}
+};
+
+class ManagementObject : public ManagementItem
+{
+ protected:
+
+ uint64_t createTime;
+ uint64_t destroyTime;
+ ObjectId objectId;
+ bool configChanged;
+ bool instChanged;
+ bool deleted;
+ Manageable* coreObject;
+ sys::Mutex accessLock;
+ ManagementAgent* agent;
+ int maxThreads;
+
+ static int nextThreadIndex;
int getThreadIndex();
void writeTimestamps (qpid::framing::Buffer& buf);
- sys::Mutex& getMutex();
- framing::Buffer* startEventLH();
- void finishEventLH(framing::Buffer* buf);
-
public:
typedef void (*writeSchemaCall_t) (qpid::framing::Buffer&);
@@ -129,14 +137,14 @@ class ManagementObject
virtual void writeProperties(qpid::framing::Buffer& buf) = 0;
virtual void writeStatistics(qpid::framing::Buffer& buf,
bool skipHeaders = false) = 0;
- virtual void doMethod (std::string methodName,
+ virtual void doMethod (std::string& methodName,
qpid::framing::Buffer& inBuf,
qpid::framing::Buffer& outBuf) = 0;
virtual void setReference (ObjectId objectId);
- virtual std::string& getClassName (void) = 0;
- virtual std::string& getPackageName (void) = 0;
- virtual uint8_t* getMd5Sum (void) = 0;
+ virtual std::string& getClassName (void) const = 0;
+ virtual std::string& getPackageName (void) const = 0;
+ virtual uint8_t* getMd5Sum (void) const = 0;
void setObjectId (ObjectId oid) { objectId = oid; }
ObjectId getObjectId (void) { return objectId; }
View Lorry Controller Status