summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp2
-rw-r--r--qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp9
-rw-r--r--qpid/cpp/src/qpid/management/ManagementAgent.cpp438
-rw-r--r--qpid/cpp/src/qpid/management/ManagementAgent.h38
-rw-r--r--qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp2
-rw-r--r--qpid/cpp/src/qpid/management/ManagementObject.cpp3
-rw-r--r--qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp2
-rw-r--r--qpid/cpp/src/qpid/types/Variant.cpp36
-rw-r--r--qpid/cpp/src/tests/ManagementTest.cpp10
-rw-r--r--qpid/extras/qmf/src/py/qmf/console.py140
-rwxr-xr-xqpid/tools/src/py/qpid-route29
11 files changed, 415 insertions, 294 deletions
diff --git a/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp b/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp
index b0207582ee..890ecd2ca2 100644
--- a/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp
+++ b/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp
@@ -653,7 +653,7 @@ void ManagementAgentImpl::handleGetQuery(const string& body, const string& cid,
}
}
- // end empty "non-partial" message to indicate CommandComplete
+ // Send empty "non-partial" message to indicate CommandComplete
list_.clear();
headers.erase("partial");
ListCodec::encode(list_, content);
diff --git a/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp b/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
index 6806b3e1dd..3192896956 100644
--- a/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
+++ b/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
@@ -146,16 +146,9 @@ struct ClusterPlugin : public Plugin {
}
}
- void disallow(ManagementAgent* agent, const string& className, const string& methodName) {
- string message = "Management method " + className + ":" + methodName + " is not allowed on a clustered broker.";
- agent->disallow(className, methodName, message);
- }
void disallowManagementMethods(ManagementAgent* agent) {
if (!agent) return;
- disallow(agent, "queue", "purge");
- disallow(agent, "session", "detach");
- disallow(agent, "session", "close");
- disallow(agent, "connection", "close");
+ agent->disallowV1Methods();
}
void initialize(Plugin::Target& target) {
diff --git a/qpid/cpp/src/qpid/management/ManagementAgent.cpp b/qpid/cpp/src/qpid/management/ManagementAgent.cpp
index c4b670ca5f..1a1415a06e 100644
--- a/qpid/cpp/src/qpid/management/ManagementAgent.cpp
+++ b/qpid/cpp/src/qpid/management/ManagementAgent.cpp
@@ -54,9 +54,9 @@ namespace _qmf = qmf::org::apache::qpid::broker;
-static Variant::Map mapEncodeSchemaId(const std::string& pname,
- const std::string& cname,
- const std::string& type,
+static Variant::Map mapEncodeSchemaId(const string& pname,
+ const string& cname,
+ const string& type,
const uint8_t *md5Sum)
{
Variant::Map map_;
@@ -81,7 +81,7 @@ ManagementAgent::RemoteAgent::~RemoteAgent ()
ManagementAgent::ManagementAgent (const bool qmfV1, const bool qmfV2) :
threadPoolSize(1), interval(10), broker(0), timer(0),
startTime(sys::now()),
- suppressed(false),
+ suppressed(false), disallowAllV1Methods(false),
qmf1Support(qmfV1), qmf2Support(qmfV2)
{
nextObjectId = 1;
@@ -287,7 +287,7 @@ ObjectId ManagementAgent::addObject(ManagementObject* object, uint64_t persistId
ObjectId ManagementAgent::addObject(ManagementObject* object,
- const std::string& key,
+ const string& key,
bool persistent)
{
uint16_t sequence;
@@ -337,7 +337,7 @@ void ManagementAgent::raiseEvent(const ManagementEvent& event, severity_t severi
outBuffer.putBin128(event.getMd5Sum());
outBuffer.putLongLong(uint64_t(sys::Duration(sys::EPOCH, sys::now())));
outBuffer.putOctet(sev);
- std::string sBuf;
+ string sBuf;
event.encode(sBuf);
outBuffer.putRawData(sBuf);
outLen = MA_BUFFER_SIZE - outBuffer.available();
@@ -391,7 +391,7 @@ void ManagementAgent::Periodic::fire ()
agent.periodicProcessing ();
}
-void ManagementAgent::clientAdded (const std::string& routingKey)
+void ManagementAgent::clientAdded (const string& routingKey)
{
sys::Mutex::ScopedLock lock(userLock);
@@ -492,12 +492,12 @@ void ManagementAgent::sendBufferLH(Buffer& buf,
// NOTE WELL: assumes userLock is held by caller (LH)
-void ManagementAgent::sendBufferLH(const std::string& data,
- const std::string& cid,
+void ManagementAgent::sendBufferLH(const string& data,
+ const string& cid,
const Variant::Map& headers,
- const std::string& content_type,
+ const string& content_type,
qpid::broker::Exchange::shared_ptr exchange,
- const std::string& routingKey)
+ const string& routingKey)
{
Variant::Map::const_iterator i;
@@ -592,7 +592,7 @@ void ManagementAgent::periodicProcessing (void)
uint32_t contentSize;
string routingKey;
list<pair<ObjectId, ManagementObject*> > deleteList;
- std::string sBuf;
+ string sBuf;
uint64_t uptime = sys::Duration(startTime, sys::now());
static_cast<_qmf::Broker*>(broker->GetManagementObject())->set_uptime(uptime);
@@ -844,7 +844,7 @@ void ManagementAgent::deleteObjectNowLH(const ObjectId& oid)
char msgChars[DNOW_BUFSIZE];
uint32_t contentSize;
Buffer msgBuffer(msgChars, DNOW_BUFSIZE);
- std::string sBuf;
+ string sBuf;
encodeHeader(msgBuffer, 'c');
object->writeProperties(sBuf);
@@ -889,8 +889,8 @@ void ManagementAgent::deleteObjectNowLH(const ObjectId& oid)
managementObjects.erase(oid);
}
-void ManagementAgent::sendCommandCompleteLH(string replyToKey, uint32_t sequence,
- uint32_t code, string text)
+void ManagementAgent::sendCommandCompleteLH(const string& replyToKey, uint32_t sequence,
+ uint32_t code, const string& text)
{
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
@@ -905,14 +905,40 @@ void ManagementAgent::sendCommandCompleteLH(string replyToKey, uint32_t sequence
replyToKey << " seq=" << sequence);
}
+void ManagementAgent::sendExceptionLH(const string& replyToKey, const string& cid,
+ const string& text, uint32_t code, bool viaLocal)
+{
+ static const string addr_exchange("qmf.default.direct");
+
+ Variant::Map map;
+ Variant::Map headers;
+ Variant::Map values;
+ string content;
+
+ headers["method"] = "indication";
+ headers["qmf.opcode"] = "_exception";
+ headers["qmf.agent"] = viaLocal ? "broker" : name_address;
+
+ values["error_code"] = code;
+ values["error_text"] = text;
+ map["_values"] = values;
+
+ MapCodec::encode(map, content);
+ sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyToKey);
+
+ QPID_LOG(trace, "SENT Exception code=" << code <<" text=" << text);
+}
+
bool ManagementAgent::dispatchCommand (Deliverable& deliverable,
const string& routingKey,
const FieldTable* /*args*/,
- const bool topic)
+ const bool topic,
+ int qmfVersion)
{
sys::Mutex::ScopedLock lock (userLock);
Message& msg = ((DeliverableMessage&) deliverable).getMessage ();
- if (topic) {
+
+ if (topic && qmfVersion == 1) {
// qmf1 is bound only to the topic management exchange.
// Parse the routing key. This management broker should act as though it
@@ -945,7 +971,7 @@ bool ManagementAgent::dispatchCommand (Deliverable& deliverable,
}
}
- if (qmf2Support) {
+ if (qmfVersion == 2) {
if (topic) {
// Intercept messages bound to:
@@ -962,7 +988,7 @@ bool ManagementAgent::dispatchCommand (Deliverable& deliverable,
// "<name_address>" - the broker agent's proper name
// and do not forward them futher
if (routingKey == "broker" || routingKey == name_address) {
- dispatchAgentCommandLH(msg);
+ dispatchAgentCommandLH(msg, routingKey == "broker");
return false;
}
}
@@ -971,8 +997,7 @@ bool ManagementAgent::dispatchCommand (Deliverable& deliverable,
return true;
}
-void ManagementAgent::handleMethodRequestLH (Buffer& inBuffer, string replyToKey,
- uint32_t sequence, const ConnectionToken* connToken)
+void ManagementAgent::handleMethodRequestLH(Buffer& inBuffer, const string& replyToKey, uint32_t sequence, const ConnectionToken* connToken)
{
string methodName;
string packageName;
@@ -981,9 +1006,9 @@ void ManagementAgent::handleMethodRequestLH (Buffer& inBuffer, string replyToKey
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
AclModule* acl = broker->getAcl();
- std::string inArgs;
+ string inArgs;
- std::string sBuf;
+ string sBuf;
inBuffer.getRawData(sBuf, 16);
ObjectId objId;
objId.decode(sBuf);
@@ -998,7 +1023,17 @@ void ManagementAgent::handleMethodRequestLH (Buffer& inBuffer, string replyToKey
encodeHeader(outBuffer, 'm', sequence);
- DisallowedMethods::const_iterator i = disallowed.find(std::make_pair(className, methodName));
+ if (disallowAllV1Methods) {
+ outBuffer.putLong(Manageable::STATUS_FORBIDDEN);
+ outBuffer.putMediumString("QMFv1 methods forbidden on this broker, use QMFv2");
+ outLen = MA_BUFFER_SIZE - outBuffer.available();
+ outBuffer.reset();
+ sendBufferLH(outBuffer, outLen, dExchange, replyToKey);
+ QPID_LOG(trace, "SEND MethodResponse status=FORBIDDEN reason='All QMFv1 Methods Forbidden' seq=" << sequence);
+ return;
+ }
+
+ DisallowedMethods::const_iterator i = disallowed.find(make_pair(className, methodName));
if (i != disallowed.end()) {
outBuffer.putLong(Manageable::STATUS_FORBIDDEN);
outBuffer.putMediumString(i->second);
@@ -1040,7 +1075,7 @@ void ManagementAgent::handleMethodRequestLH (Buffer& inBuffer, string replyToKey
try {
outBuffer.record();
sys::Mutex::ScopedUnlock u(userLock);
- std::string outBuf;
+ string outBuf;
iter->second->doMethod(methodName, inArgs, outBuf);
outBuffer.putRawData(outBuf);
} catch(exception& e) {
@@ -1057,39 +1092,34 @@ void ManagementAgent::handleMethodRequestLH (Buffer& inBuffer, string replyToKey
}
-void ManagementAgent::handleMethodRequestLH (const std::string& body, string replyTo,
- const std::string& cid, const ConnectionToken* connToken)
+void ManagementAgent::handleMethodRequestLH (const string& body, const string& replyTo,
+ const string& cid, const ConnectionToken* connToken, bool viaLocal)
{
string methodName;
Variant::Map inMap;
MapCodec::decode(body, inMap);
Variant::Map::const_iterator oid, mid;
string content;
+ string error;
+ uint32_t errorCode(0);
Variant::Map outMap;
Variant::Map headers;
headers["method"] = "response";
headers["qmf.opcode"] = "_method_response";
- headers["qmf.agent"] = name_address;
+ headers["qmf.agent"] = viaLocal ? "broker" : name_address;
if ((oid = inMap.find("_object_id")) == inMap.end() ||
- (mid = inMap.find("_method_name")) == inMap.end())
- {
- Variant::Map _values;
- headers["qmf.opcode"] = "_exception";
- _values["_status"] = Manageable::STATUS_PARAMETER_INVALID;
- _values["_status_text"] = Manageable::StatusText(Manageable::STATUS_PARAMETER_INVALID);
- outMap["_values"] = _values;
-
- MapCodec::encode(outMap, content);
- sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyTo);
- QPID_LOG(trace, "SEND MethodResponse (invalid param) to=" << replyTo << " seq=" << cid);
+ (mid = inMap.find("_method_name")) == inMap.end()) {
+ sendExceptionLH(replyTo, cid, Manageable::StatusText(Manageable::STATUS_PARAMETER_INVALID),
+ Manageable::STATUS_PARAMETER_INVALID, viaLocal);
return;
}
ObjectId objId;
Variant::Map inArgs;
+ Variant::Map callMap;
try {
// coversions will throw if input is invalid.
@@ -1101,31 +1131,16 @@ void ManagementAgent::handleMethodRequestLH (const std::string& body, string rep
inArgs = (mid->second).asMap();
}
} catch(exception& e) {
- Variant::Map _values;
- headers["qmf.opcode"] = "_exception";
- _values["_status"] = Manageable::STATUS_EXCEPTION;
- _values["_status_text"] = e.what();
- outMap["_values"] = _values;
-
-
- MapCodec::encode(outMap, content);
- sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyTo);
- QPID_LOG(trace, "SEND MethodResponse (invalid format) to=" << replyTo << " seq=" << cid);
+ sendExceptionLH(replyTo, cid, e.what(), Manageable::STATUS_EXCEPTION, viaLocal);
return;
}
ManagementObjectMap::iterator iter = managementObjects.find(objId);
if (iter == managementObjects.end() || iter->second->isDeleted()) {
- Variant::Map _values;
- headers["qmf.opcode"] = "_exception";
- _values["_status"] = Manageable::STATUS_UNKNOWN_OBJECT;
- _values["_status_text"] = Manageable::StatusText(Manageable::STATUS_UNKNOWN_OBJECT);
- outMap["_values"] = _values;
-
- MapCodec::encode(outMap, content);
- sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyTo);
- QPID_LOG(trace, "SEND MethodResponse (unknown object) to=" << replyTo << " seq=" << cid);
+ stringstream estr;
+ estr << "No object found with ID=" << objId;
+ sendExceptionLH(replyTo, cid, estr.str(), 1, viaLocal);
return;
}
@@ -1133,17 +1148,9 @@ void ManagementAgent::handleMethodRequestLH (const std::string& body, string rep
AclModule* acl = broker->getAcl();
DisallowedMethods::const_iterator i;
- i = disallowed.find(std::make_pair(iter->second->getClassName(), methodName));
+ i = disallowed.find(make_pair(iter->second->getClassName(), methodName));
if (i != disallowed.end()) {
- Variant::Map _values;
- headers["qmf.opcode"] = "_exception";
- _values["_status"] = Manageable::STATUS_FORBIDDEN;
- _values["_status_text"] = i->second;
- outMap["_values"] = _values;
-
- MapCodec::encode(outMap, content);
- sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyTo);
- QPID_LOG(trace, "SEND MethodResponse status=FORBIDDEN text=" << i->second << " seq=" << cid);
+ sendExceptionLH(replyTo, cid, i->second, Manageable::STATUS_FORBIDDEN, viaLocal);
return;
}
@@ -1154,15 +1161,8 @@ void ManagementAgent::handleMethodRequestLH (const std::string& body, string rep
params[acl::PROP_SCHEMACLASS] = iter->second->getClassName();
if (!acl->authorise(userId, acl::ACT_ACCESS, acl::OBJ_METHOD, methodName, &params)) {
- Variant::Map _values;
- headers["qmf.opcode"] = "_exception";
- _values["_status"] = Manageable::STATUS_FORBIDDEN;
- _values["_status_text"] = Manageable::StatusText(Manageable::STATUS_FORBIDDEN);
- outMap["_values"] = _values;
-
- MapCodec::encode(outMap, content);
- sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyTo);
- QPID_LOG(trace, "SEND MethodResponse status=FORBIDDEN" << " seq=" << cid);
+ sendExceptionLH(replyTo, cid, Manageable::StatusText(Manageable::STATUS_FORBIDDEN),
+ Manageable::STATUS_FORBIDDEN, viaLocal);
return;
}
}
@@ -1171,32 +1171,37 @@ void ManagementAgent::handleMethodRequestLH (const std::string& body, string rep
QPID_LOG(trace, "RECV MethodRequest (v2) class=" << iter->second->getPackageName()
<< ":" << iter->second->getClassName() << " method=" <<
- methodName << " replyTo=" << replyTo);
+ methodName << " replyTo=" << replyTo << " objId=" << objId << " inArgs=" << inArgs);
try {
sys::Mutex::ScopedUnlock u(userLock);
- iter->second->doMethod(methodName, inArgs, outMap);
+ iter->second->doMethod(methodName, inArgs, callMap);
+ errorCode = callMap["_status_code"].asUint32();
+ if (errorCode == 0) {
+ outMap["_arguments"] = Variant::Map();
+ for (Variant::Map::const_iterator iter = callMap.begin();
+ iter != callMap.end(); iter++)
+ if (iter->first != "_status_code" && iter->first != "_status_text")
+ outMap["_arguments"].asMap()[iter->first] = iter->second;
+ } else
+ error = callMap["_status_text"].asString();
} catch(exception& e) {
- Variant::Map _values;
- outMap.clear();
- headers["qmf.opcode"] = "_exception";
- _values["_status"] = Manageable::STATUS_EXCEPTION;
- _values["_status_text"] = e.what();
- outMap["_values"] = _values;
-
- MapCodec::encode(outMap, content);
- sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyTo);
- QPID_LOG(trace, "SEND MethodResponse (exception) to=" << replyTo << " seq=" << cid);
+ sendExceptionLH(replyTo, cid, e.what(), Manageable::STATUS_EXCEPTION, viaLocal);
+ return;
+ }
+
+ if (errorCode != 0) {
+ sendExceptionLH(replyTo, cid, error, errorCode, viaLocal);
return;
}
MapCodec::encode(outMap, content);
sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyTo);
- QPID_LOG(trace, "SEND MethodResponse (v2) to=" << replyTo << " seq=" << cid);
+ QPID_LOG(trace, "SEND MethodResponse (v2) to=" << replyTo << " seq=" << cid << " map=" << outMap);
}
-void ManagementAgent::handleBrokerRequestLH (Buffer&, string replyToKey, uint32_t sequence)
+void ManagementAgent::handleBrokerRequestLH (Buffer&, const string& replyToKey, uint32_t sequence)
{
Buffer outBuffer (outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
@@ -1212,7 +1217,7 @@ void ManagementAgent::handleBrokerRequestLH (Buffer&, string replyToKey, uint32_
QPID_LOG(trace, "SEND BrokerResponse to=" << replyToKey);
}
-void ManagementAgent::handlePackageQueryLH (Buffer&, string replyToKey, uint32_t sequence)
+void ManagementAgent::handlePackageQueryLH (Buffer&, const string& replyToKey, uint32_t sequence)
{
QPID_LOG(trace, "RECV PackageQuery replyTo=" << replyToKey);
@@ -1234,7 +1239,7 @@ void ManagementAgent::handlePackageQueryLH (Buffer&, string replyToKey, uint32_t
sendCommandCompleteLH(replyToKey, sequence);
}
-void ManagementAgent::handlePackageIndLH (Buffer& inBuffer, string replyToKey, uint32_t sequence)
+void ManagementAgent::handlePackageIndLH (Buffer& inBuffer, const string& replyToKey, uint32_t sequence)
{
string packageName;
@@ -1245,7 +1250,7 @@ void ManagementAgent::handlePackageIndLH (Buffer& inBuffer, string replyToKey, u
findOrAddPackageLH(packageName);
}
-void ManagementAgent::handleClassQueryLH(Buffer& inBuffer, string replyToKey, uint32_t sequence)
+void ManagementAgent::handleClassQueryLH(Buffer& inBuffer, const string& replyToKey, uint32_t sequence)
{
string packageName;
@@ -1279,7 +1284,7 @@ void ManagementAgent::handleClassQueryLH(Buffer& inBuffer, string replyToKey, ui
sendCommandCompleteLH(replyToKey, sequence);
}
-void ManagementAgent::handleClassIndLH (Buffer& inBuffer, string replyToKey, uint32_t)
+void ManagementAgent::handleClassIndLH (Buffer& inBuffer, const string& replyToKey, uint32_t)
{
string packageName;
SchemaClassKey key;
@@ -1323,14 +1328,14 @@ void ManagementAgent::SchemaClass::appendSchema(Buffer& buf)
// is from a remote management agent, send the stored schema information.
if (writeSchemaCall != 0) {
- std::string schema;
+ string schema;
writeSchemaCall(schema);
buf.putRawData(schema);
} else
buf.putRawData(reinterpret_cast<uint8_t*>(&data[0]), data.size());
}
-void ManagementAgent::handleSchemaRequestLH(Buffer& inBuffer, string replyToKey, uint32_t sequence)
+void ManagementAgent::handleSchemaRequestLH(Buffer& inBuffer, const string& replyToKey, uint32_t sequence)
{
string packageName;
SchemaClassKey key;
@@ -1368,7 +1373,7 @@ void ManagementAgent::handleSchemaRequestLH(Buffer& inBuffer, string replyToKey,
sendCommandCompleteLH(replyToKey, sequence, 1, "Package not found");
}
-void ManagementAgent::handleSchemaResponseLH(Buffer& inBuffer, string /*replyToKey*/, uint32_t sequence)
+void ManagementAgent::handleSchemaResponseLH(Buffer& inBuffer, const string& /*replyToKey*/, uint32_t sequence)
{
string packageName;
SchemaClassKey key;
@@ -1466,7 +1471,7 @@ void ManagementAgent::deleteOrphanedAgentsLH()
deleteList.clear();
}
-void ManagementAgent::handleAttachRequestLH (Buffer& inBuffer, string replyToKey, uint32_t sequence, const ConnectionToken* connToken)
+void ManagementAgent::handleAttachRequestLH (Buffer& inBuffer, const string& replyToKey, uint32_t sequence, const ConnectionToken* connToken)
{
string label;
uint32_t requestedBrokerBank, requestedAgentBank;
@@ -1524,7 +1529,7 @@ void ManagementAgent::handleAttachRequestLH (Buffer& inBuffer, string replyToKey
" to=" << replyToKey << " seq=" << sequence);
}
-void ManagementAgent::handleGetQueryLH (Buffer& inBuffer, string replyToKey, uint32_t sequence)
+void ManagementAgent::handleGetQueryLH(Buffer& inBuffer, const string& replyToKey, uint32_t sequence)
{
FieldTable ft;
FieldTable::ValuePtr value;
@@ -1552,7 +1557,7 @@ void ManagementAgent::handleGetQueryLH (Buffer& inBuffer, string replyToKey, uin
object->setUpdateTime();
if (!object->isDeleted()) {
- std::string sBuf;
+ string sBuf;
encodeHeader(outBuffer, 'g', sequence);
object->writeProperties(sBuf);
outBuffer.putRawData(sBuf);
@@ -1583,7 +1588,7 @@ void ManagementAgent::handleGetQueryLH (Buffer& inBuffer, string replyToKey, uin
object->setUpdateTime();
if (!object->isDeleted()) {
- std::string sBuf;
+ string sBuf;
encodeHeader(outBuffer, 'g', sequence);
object->writeProperties(sBuf);
outBuffer.putRawData(sBuf);
@@ -1602,108 +1607,144 @@ void ManagementAgent::handleGetQueryLH (Buffer& inBuffer, string replyToKey, uin
}
-void ManagementAgent::handleGetQueryLH(const std::string& body, std::string replyTo, const std::string& cid, const std::string& contentType)
+void ManagementAgent::handleGetQueryLH(const string& body, const string& replyTo, const string& cid, bool viaLocal)
{
- FieldTable ft;
- FieldTable::ValuePtr value;
-
moveNewObjectsLH();
- if (contentType != "_query_v1") {
- QPID_LOG(warning, "Support for QMF V2 Query format TBD!!!");
- return;
- }
-
Variant::Map inMap;
- MapCodec::decode(body, inMap);
Variant::Map::const_iterator i;
Variant::Map headers;
+ MapCodec::decode(body, inMap);
QPID_LOG(trace, "RECV GetQuery (v2): map=" << inMap << " seq=" << cid);
headers["method"] = "response";
headers["qmf.opcode"] = "_query_response";
headers["qmf.content"] = "_data";
- headers["qmf.agent"] = name_address;
- headers["partial"];
+ headers["qmf.agent"] = viaLocal ? "broker" : name_address;
+ headers["partial"] = Variant();
Variant::List list_;
Variant::Map map_;
Variant::Map values;
- string className;
+ Variant::Map oidMap;
string content;
- i = inMap.find("_class");
- if (i != inMap.end())
- try {
- className = i->second.asString();
- } catch(exception& /*e*/) {
- className.clear();
- QPID_LOG(trace, "RCVD GetQuery: invalid format - class target ignored.");
- }
+ /*
+ * Unpack the _what element of the query. Currently we only support OBJECT queries.
+ */
+ i = inMap.find("_what");
+ if (i == inMap.end()) {
+ sendExceptionLH(replyTo, cid, "_what element missing in Query");
+ return;
+ }
- if (className.empty()) {
- ObjectId objId;
- i = inMap.find("_object_id");
- if (i != inMap.end()) {
+ if (i->second.getType() != qpid::types::VAR_STRING) {
+ sendExceptionLH(replyTo, cid, "_what element is not a string");
+ return;
+ }
- try {
- objId = ObjectId(i->second.asMap());
- } catch (exception & /*e*/) {
- objId = ObjectId(); // empty object id - won't find a match (I hope).
- QPID_LOG(trace, "RCVD GetQuery (invalid Object Id format) to=" << replyTo << " seq=" << cid);
- }
+ if (i->second.asString() != "OBJECT") {
+ sendExceptionLH(replyTo, cid, "Query for _what => '" + i->second.asString() + "' not supported");
+ return;
+ }
- ManagementObjectMap::iterator iter = managementObjects.find(objId);
- if (iter != managementObjects.end()) {
- ManagementObject* object = iter->second;
+ string className;
+ string packageName;
- if (object->getConfigChanged() || object->getInstChanged())
- object->setUpdateTime();
+ /*
+ * Handle the _schema_id element, if supplied.
+ */
+ i = inMap.find("_schema_id");
+ if (i != inMap.end() && i->second.getType() == qpid::types::VAR_MAP) {
+ const Variant::Map& schemaIdMap(i->second.asMap());
- if (!object->isDeleted()) {
- object->writeTimestamps(map_);
- object->mapEncodeValues(values, true, true); // write both stats and properties
- map_["_values"] = values;
- list_.push_back(map_);
+ Variant::Map::const_iterator s_iter = schemaIdMap.find("_class_name");
+ if (s_iter != schemaIdMap.end() && s_iter->second.getType() == qpid::types::VAR_STRING)
+ className = s_iter->second.asString();
- ListCodec::encode(list_, content);
- sendBufferLH(content, cid, headers, "amqp/list", v2Direct, replyTo);
- }
+ s_iter = schemaIdMap.find("_package_name");
+ if (s_iter != schemaIdMap.end() && s_iter->second.getType() == qpid::types::VAR_STRING)
+ packageName = s_iter->second.asString();
+ }
+
+
+ /*
+ * Unpack the _object_id element of the query if it is present. If it is present, find that one
+ * object and return it. If it is not present, send a class-based result.
+ */
+ i = inMap.find("_object_id");
+ if (i != inMap.end() && i->second.getType() == qpid::types::VAR_MAP) {
+ ObjectId objId(i->second.asMap());
+
+ ManagementObjectMap::iterator iter = managementObjects.find(objId);
+ if (iter != managementObjects.end()) {
+ ManagementObject* object = iter->second;
+
+ if (object->getConfigChanged() || object->getInstChanged())
+ object->setUpdateTime();
+
+ if (!object->isDeleted()) {
+ object->mapEncodeValues(values, true, true); // write both stats and properties
+ objId.mapEncode(oidMap);
+ map_["_values"] = values;
+ map_["_object_id"] = oidMap;
+ map_["_schema_id"] = mapEncodeSchemaId(object->getPackageName(),
+ object->getClassName(),
+ "_data",
+ object->getMd5Sum());
+ list_.push_back(map_);
}
+
+ headers.erase("partial");
+ ListCodec::encode(list_, content);
+ sendBufferLH(content, cid, headers, "amqp/list", v2Direct, replyTo);
+ QPID_LOG(trace, "SENT QueryResponse (query by object_id) to=" << replyTo);
+ return;
}
} else {
for (ManagementObjectMap::iterator iter = managementObjects.begin();
iter != managementObjects.end();
iter++) {
ManagementObject* object = iter->second;
- if (object->getClassName () == className) {
-
- // @todo: support multiple objects per message reply
- values.clear();
- list_.clear();
- if (object->getConfigChanged() || object->getInstChanged())
- object->setUpdateTime();
+ if (object->getClassName() == className &&
+ (packageName.empty() || object->getPackageName() == packageName)) {
+ // @todo support multiple object reply per message
if (!object->isDeleted()) {
+ values.clear();
+ list_.clear();
+ oidMap.clear();
+
+ if (object->getConfigChanged() || object->getInstChanged())
+ object->setUpdateTime();
+
object->writeTimestamps(map_);
object->mapEncodeValues(values, true, true); // write both stats and properties
- map_["_values"] = values;
- list_.push_back(map_);
+ iter->first.mapEncode(oidMap);
+
+ map_["_values"] = values;
+ map_["_object_id"] = oidMap;
+ map_["_schema_id"] = mapEncodeSchemaId(object->getPackageName(),
+ object->getClassName(),
+ "_data",
+ object->getMd5Sum());
+ list_.push_back(map_);
+ }
ListCodec::encode(list_, content);
sendBufferLH(content, cid, headers, "amqp/list", v2Direct, replyTo);
- }
+ QPID_LOG(trace, "SENT QueryResponse (query by schema_id) to=" << replyTo);
}
}
}
- // end empty "non-partial" message to indicate CommandComplete
+ // Send empty "non-partial" message to indicate CommandComplete
list_.clear();
headers.erase("partial");
ListCodec::encode(list_, content);
sendBufferLH(content, cid, headers, "amqp/list", v2Direct, replyTo);
- QPID_LOG(trace, "SEND GetResponse (v2) to=" << replyTo << " seq=" << cid);
+ QPID_LOG(trace, "SENT QueryResponse (empty with no 'partial' indicator) to=" << replyTo);
}
@@ -1742,7 +1783,7 @@ bool ManagementAgent::authorizeAgentMessageLH(Message& msg)
string packageName;
string className;
string methodName;
- std::string cid;
+ string cid;
if (msg.encodedSize() > MA_BUFFER_SIZE)
return false;
@@ -1770,7 +1811,7 @@ bool ManagementAgent::authorizeAgentMessageLH(Message& msg)
// extract object id and method name
- std::string body;
+ string body;
inBuffer.getRawData(body, bufferLen);
Variant::Map inMap;
MapCodec::decode(body, inMap);
@@ -1852,24 +1893,13 @@ bool ManagementAgent::authorizeAgentMessageLH(Message& msg)
if (p && p->hasReplyTo()) {
const framing::ReplyTo& rt = p->getReplyTo();
string replyToKey = rt.getRoutingKey();
+ string cid;
+ if (p && p->hasCorrelationId())
+ cid = p->getCorrelationId();
if (mapMsg) {
- Variant::Map _values;
- Variant::Map outMap;
- Variant::Map headers;
-
- headers["method"] = "response";
- headers["qmf.opcode"] = "_exception";
- headers["qmf.agent"] = name_address;
-
- _values["_status"] = Manageable::STATUS_FORBIDDEN;
- _values["_status_text"] = Manageable::StatusText(Manageable::STATUS_FORBIDDEN);
- outMap["_values"] = _values;
-
- string content;
- MapCodec::encode(outMap, content);
- sendBufferLH(content, cid, headers, "amqp/map", v2Direct, replyToKey);
-
+ sendExceptionLH(replyToKey, cid, Manageable::StatusText(Manageable::STATUS_FORBIDDEN),
+ Manageable::STATUS_FORBIDDEN, false);
} else {
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
@@ -1892,7 +1922,7 @@ bool ManagementAgent::authorizeAgentMessageLH(Message& msg)
return true;
}
-void ManagementAgent::dispatchAgentCommandLH(Message& msg)
+void ManagementAgent::dispatchAgentCommandLH(Message& msg, bool viaLocal)
{
string replyToKey;
const framing::MessageProperties* p =
@@ -1920,10 +1950,10 @@ void ManagementAgent::dispatchAgentCommandLH(Message& msg)
const framing::FieldTable *headers = msg.getApplicationHeaders();
if (headers && msg.getAppId() == "qmf2")
{
- std::string opcode = headers->getAsString("qmf.opcode");
- std::string contentType = headers->getAsString("qmf.content");
- std::string body;
- std::string cid;
+ string opcode = headers->getAsString("qmf.opcode");
+ string contentType = headers->getAsString("qmf.content");
+ string body;
+ string cid;
inBuffer.getRawData(body, bufferLen);
if (p && p->hasCorrelationId()) {
@@ -1931,9 +1961,9 @@ void ManagementAgent::dispatchAgentCommandLH(Message& msg)
}
if (opcode == "_method_request")
- return handleMethodRequestLH(body, replyToKey, cid, msg.getPublisher());
+ return handleMethodRequestLH(body, replyToKey, cid, msg.getPublisher(), viaLocal);
else if (opcode == "_query_request")
- return handleGetQueryLH(body, replyToKey, cid, contentType);
+ return handleGetQueryLH(body, replyToKey, cid, viaLocal);
else if (opcode == "_agent_locate_request")
return handleLocateRequestLH(body, replyToKey, cid);
@@ -2140,7 +2170,7 @@ ManagementObjectMap::iterator ManagementAgent::numericFind(const ObjectId& oid)
return iter;
}
-void ManagementAgent::setAllocator(std::auto_ptr<IdAllocator> a)
+void ManagementAgent::setAllocator(auto_ptr<IdAllocator> a)
{
sys::Mutex::ScopedLock lock (userLock);
allocator = a;
@@ -2153,8 +2183,8 @@ uint64_t ManagementAgent::allocateId(Manageable* object)
return 0;
}
-void ManagementAgent::disallow(const std::string& className, const std::string& methodName, const std::string& message) {
- disallowed[std::make_pair(className, methodName)] = message;
+void ManagementAgent::disallow(const string& className, const string& methodName, const string& message) {
+ disallowed[make_pair(className, methodName)] = message;
}
void ManagementAgent::SchemaClassKey::mapEncode(Variant::Map& _map) const {
@@ -2211,7 +2241,7 @@ void ManagementAgent::SchemaClass::mapDecode(const Variant::Map& _map) {
}
}
-void ManagementAgent::exportSchemas(std::string& out) {
+void ManagementAgent::exportSchemas(string& out) {
Variant::List list_;
Variant::Map map_, kmap, cmap;
@@ -2318,7 +2348,7 @@ void ManagementAgent::RemoteAgent::mapDecode(const Variant::Map& map_) {
mgmtObject->set_connectionRef(connectionRef);
}
-void ManagementAgent::exportAgents(std::string& out) {
+void ManagementAgent::exportAgents(string& out) {
Variant::List list_;
Variant::Map map_, omap, amap;
@@ -2348,7 +2378,7 @@ void ManagementAgent::importAgents(qpid::framing::Buffer& inBuf) {
sys::Mutex::ScopedLock lock(userLock);
for (l = content.begin(); l != content.end(); l++) {
- std::auto_ptr<RemoteAgent> agent(new RemoteAgent(*this));
+ auto_ptr<RemoteAgent> agent(new RemoteAgent(*this));
Variant::Map map_;
Variant::Map::const_iterator i;
@@ -2364,8 +2394,8 @@ void ManagementAgent::importAgents(qpid::framing::Buffer& inBuf) {
}
}
-std::string ManagementAgent::debugSnapshot() {
- std::ostringstream msg;
+string ManagementAgent::debugSnapshot() {
+ ostringstream msg;
msg << " management snapshot:";
for (RemoteAgentMap::const_iterator i=remoteAgents.begin();
i != remoteAgents.end(); ++i)
@@ -2466,14 +2496,14 @@ boost::shared_ptr<FieldValue> ManagementAgent::toFieldValue(const Variant& in)
// stolen from qpid/client/amqp0_10/Codecs.cpp - TODO: make Codecs public, and remove this dup.
Variant ManagementAgent::toVariant(const boost::shared_ptr<FieldValue>& in)
{
- const std::string iso885915("iso-8859-15");
- const std::string utf8("utf8");
- const std::string utf16("utf16");
- //const std::string binary("binary");
- const std::string amqp0_10_binary("amqp0-10:binary");
- //const std::string amqp0_10_bit("amqp0-10:bit");
- const std::string amqp0_10_datetime("amqp0-10:datetime");
- const std::string amqp0_10_struct("amqp0-10:struct");
+ const string iso885915("iso-8859-15");
+ const string utf8("utf8");
+ const string utf16("utf16");
+ //const string binary("binary");
+ const string amqp0_10_binary("amqp0-10:binary");
+ //const string amqp0_10_bit("amqp0-10:bit");
+ const string amqp0_10_datetime("amqp0-10:datetime");
+ const string amqp0_10_struct("amqp0-10:struct");
Variant out;
//based on AMQP 0-10 typecode, pick most appropriate variant type
@@ -2522,30 +2552,30 @@ Variant ManagementAgent::toVariant(const boost::shared_ptr<FieldValue>& in)
case 0x80: // str8
case 0x90: // str16
case 0xa0: // str32
- out = in->get<std::string>();
+ out = in->get<string>();
out.setEncoding(amqp0_10_binary);
break;
case 0x84: // str8
case 0x94: // str16
- out = in->get<std::string>();
+ out = in->get<string>();
out.setEncoding(iso885915);
break;
case 0x85: // str8
case 0x95: // str16
- out = in->get<std::string>();
+ out = in->get<string>();
out.setEncoding(utf8);
break;
case 0x86: // str8
case 0x96: // str16
- out = in->get<std::string>();
+ out = in->get<string>();
out.setEncoding(utf16);
break;
case 0xab: // str32
- out = in->get<std::string>();
+ out = in->get<string>();
out.setEncoding(amqp0_10_struct);
break;
diff --git a/qpid/cpp/src/qpid/management/ManagementAgent.h b/qpid/cpp/src/qpid/management/ManagementAgent.h
index ca3f742cb7..dd781c36d3 100644
--- a/qpid/cpp/src/qpid/management/ManagementAgent.h
+++ b/qpid/cpp/src/qpid/management/ManagementAgent.h
@@ -108,7 +108,8 @@ public:
bool dispatchCommand (qpid::broker::Deliverable& msg,
const std::string& routingKey,
const framing::FieldTable* args,
- const bool topic);
+ const bool topic,
+ int qmfVersion);
const framing::Uuid& getUuid() const { return uuid; }
@@ -118,6 +119,9 @@ public:
/** Disallow a method. Attempts to call it will receive an exception with message. */
void disallow(const std::string& className, const std::string& methodName, const std::string& message);
+ /** Disallow all QMFv1 methods (used in clustered brokers). */
+ void disallowV1Methods() { disallowAllV1Methods = true; }
+
/** Serialize my schemas as a binary blob into schemaOut */
void exportSchemas(std::string& schemaOut);
@@ -281,6 +285,7 @@ private:
typedef std::pair<std::string,std::string> MethodName;
typedef std::map<MethodName, std::string> DisallowedMethods;
DisallowedMethods disallowed;
+ bool disallowAllV1Methods;
// Agent name and address
qpid::types::Variant::Map attrMap;
@@ -314,7 +319,7 @@ private:
void moveNewObjectsLH();
bool authorizeAgentMessageLH(qpid::broker::Message& msg);
- void dispatchAgentCommandLH(qpid::broker::Message& msg);
+ void dispatchAgentCommandLH(qpid::broker::Message& msg, bool viaLocal=false);
PackageMap::iterator findOrAddPackageLH(std::string name);
void addClassLH(uint8_t kind,
@@ -331,20 +336,21 @@ private:
uint32_t allocateNewBank ();
uint32_t assignBankLH (uint32_t requestedPrefix);
void deleteOrphanedAgentsLH();
- void sendCommandCompleteLH(std::string replyToKey, uint32_t sequence,
- uint32_t code = 0, std::string text = std::string("OK"));
- void handleBrokerRequestLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence);
- void handlePackageQueryLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence);
- void handlePackageIndLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence);
- void handleClassQueryLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence);
- void handleClassIndLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence);
- void handleSchemaRequestLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence);
- void handleSchemaResponseLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence);
- void handleAttachRequestLH (framing::Buffer& inBuffer, std::string replyToKey, uint32_t sequence, const qpid::broker::ConnectionToken* connToken);
- 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);
- void handleGetQueryLH (const std::string& body, std::string replyToKey, const std::string& cid, const std::string& contentType);
- void handleMethodRequestLH (const std::string& body, std::string replyToKey, const std::string& cid, const qpid::broker::ConnectionToken* connToken);
+ void sendCommandCompleteLH(const std::string& replyToKey, uint32_t sequence,
+ uint32_t code = 0, const std::string& text = "OK");
+ void sendExceptionLH(const std::string& replyToKey, const std::string& cid, const std::string& text, uint32_t code=1, bool viaLocal=false);
+ void handleBrokerRequestLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
+ void handlePackageQueryLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
+ void handlePackageIndLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
+ void handleClassQueryLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
+ void handleClassIndLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
+ void handleSchemaRequestLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
+ void handleSchemaResponseLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
+ void handleAttachRequestLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence, const qpid::broker::ConnectionToken* connToken);
+ void handleGetQueryLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence);
+ void handleMethodRequestLH (framing::Buffer& inBuffer, const std::string& replyToKey, uint32_t sequence, const qpid::broker::ConnectionToken* connToken);
+ void handleGetQueryLH (const std::string& body, const std::string& replyToKey, const std::string& cid, bool viaLocal);
+ void handleMethodRequestLH (const std::string& body, const std::string& replyToKey, const std::string& cid, const qpid::broker::ConnectionToken* connToken, bool viaLocal);
void handleLocateRequestLH (const std::string& body, const std::string &replyToKey, const std::string& cid);
diff --git a/qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp b/qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp
index 6dc41ef073..1d5f8bbd6b 100644
--- a/qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp
+++ b/qpid/cpp/src/qpid/management/ManagementDirectExchange.cpp
@@ -47,7 +47,7 @@ void ManagementDirectExchange::route(Deliverable& msg,
bool routeIt = true;
if (managementAgent)
- routeIt = managementAgent->dispatchCommand(msg, routingKey, args, false /*direct*/);
+ routeIt = managementAgent->dispatchCommand(msg, routingKey, args, false, qmfVersion);
if (routeIt)
DirectExchange::route(msg, routingKey, args);
diff --git a/qpid/cpp/src/qpid/management/ManagementObject.cpp b/qpid/cpp/src/qpid/management/ManagementObject.cpp
index a29770f459..209c935947 100644
--- a/qpid/cpp/src/qpid/management/ManagementObject.cpp
+++ b/qpid/cpp/src/qpid/management/ManagementObject.cpp
@@ -236,7 +236,8 @@ std::ostream& operator<<(std::ostream& out, const ObjectId& i)
"-" << ((virtFirst & 0x0FFF000000000000LL) >> 48) <<
"-" << ((virtFirst & 0x0000FFFFF0000000LL) >> 28) <<
"-" << i.agentName <<
- "-" << i.second;
+ "-" << i.second <<
+ "(" << i.v2Key << ")";
return out;
}
diff --git a/qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp b/qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp
index 7fdce133e5..ee8657646f 100644
--- a/qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp
+++ b/qpid/cpp/src/qpid/management/ManagementTopicExchange.cpp
@@ -47,7 +47,7 @@ void ManagementTopicExchange::route(Deliverable& msg,
// Intercept management agent commands
if (managementAgent)
- routeIt = managementAgent->dispatchCommand(msg, routingKey, args, true /* topic */);
+ routeIt = managementAgent->dispatchCommand(msg, routingKey, args, true, qmfVersion);
if (routeIt)
TopicExchange::route(msg, routingKey, args);
diff --git a/qpid/cpp/src/qpid/types/Variant.cpp b/qpid/cpp/src/qpid/types/Variant.cpp
index 2126c67fec..f22483fa65 100644
--- a/qpid/cpp/src/qpid/types/Variant.cpp
+++ b/qpid/cpp/src/qpid/types/Variant.cpp
@@ -217,17 +217,25 @@ uint8_t VariantImpl::asUint8() const
switch(type) {
case VAR_UINT8: return value.ui8;
case VAR_STRING: return convertFromString<uint8_t>();
- default: throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_UINT8)));
+ case VAR_INT64:
+ if (value.i64 >= 0 && value.i64 <= 0x00000000000000ff)
+ return uint8_t(value.i64);
+ default: break;
}
+ throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_UINT8)));
}
uint16_t VariantImpl::asUint16() const
{
switch(type) {
case VAR_UINT8: return value.ui8;
case VAR_UINT16: return value.ui16;
+ case VAR_INT64:
+ if (value.i64 >= 0 && value.i64 <= 0x000000000000ffff)
+ return uint16_t(value.i64);
case VAR_STRING: return convertFromString<uint16_t>();
- default: throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_UINT16)));
+ default: break;
}
+ throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_UINT16)));
}
uint32_t VariantImpl::asUint32() const
{
@@ -235,9 +243,13 @@ uint32_t VariantImpl::asUint32() const
case VAR_UINT8: return value.ui8;
case VAR_UINT16: return value.ui16;
case VAR_UINT32: return value.ui32;
+ case VAR_INT64:
+ if (value.i64 >= 0 && value.i64 <= 0x00000000ffffffff)
+ return uint32_t(value.i64);
case VAR_STRING: return convertFromString<uint32_t>();
- default: throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_UINT32)));
+ default: break;
}
+ throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_UINT32)));
}
uint64_t VariantImpl::asUint64() const
{
@@ -254,18 +266,26 @@ int8_t VariantImpl::asInt8() const
{
switch(type) {
case VAR_INT8: return value.i8;
+ case VAR_INT64:
+ if (value.i64 <= 0x000000000000007f)
+ return int8_t(value.i64);
case VAR_STRING: return convertFromString<int8_t>();
- default: throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_INT8)));
+ default: break;
}
+ throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_INT8)));
}
int16_t VariantImpl::asInt16() const
{
switch(type) {
case VAR_INT8: return value.i8;
case VAR_INT16: return value.i16;
+ case VAR_INT64:
+ if (value.i64 <= 0x0000000000007fff)
+ return int16_t(value.i64);
case VAR_STRING: return convertFromString<int16_t>();
- default: throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_INT16)));
+ default: break;
}
+ throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_INT16)));
}
int32_t VariantImpl::asInt32() const
{
@@ -273,9 +293,13 @@ int32_t VariantImpl::asInt32() const
case VAR_INT8: return value.i8;
case VAR_INT16: return value.i16;
case VAR_INT32: return value.i32;
+ case VAR_INT64:
+ if (value.i64 <= 0x000000007fffffff)
+ return int32_t(value.i64);
case VAR_STRING: return convertFromString<int32_t>();
- default: throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_INT32)));
+ default: break;
}
+ throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_INT32)));
}
int64_t VariantImpl::asInt64() const
{
diff --git a/qpid/cpp/src/tests/ManagementTest.cpp b/qpid/cpp/src/tests/ManagementTest.cpp
index 99b9c1f03e..e9b8ac32b9 100644
--- a/qpid/cpp/src/tests/ManagementTest.cpp
+++ b/qpid/cpp/src/tests/ManagementTest.cpp
@@ -33,7 +33,7 @@ using namespace qpid::framing;
using namespace qpid::management;
QPID_AUTO_TEST_CASE(testObjectIdSerializeStream) {
- std::string text("0-10-4-2500-80000000000");
+ std::string text("0-10-4-2500-80000000000()");
std::stringstream input(text);
ObjectId oid(input);
@@ -45,7 +45,7 @@ QPID_AUTO_TEST_CASE(testObjectIdSerializeStream) {
}
QPID_AUTO_TEST_CASE(testObjectIdSerializeString) {
- std::string text("0-10-4-2500-80000000000");
+ std::string text("0-10-4-2500-80000000000()");
ObjectId oid(text);
@@ -65,7 +65,7 @@ QPID_AUTO_TEST_CASE(testObjectIdEncode) {
std::stringstream out1;
out1 << oid;
- BOOST_CHECK_EQUAL(out1.str(), "1-2-3-myAgent-9999");
+ BOOST_CHECK_EQUAL(out1.str(), "1-2-3-myAgent-9999(testkey)");
}
QPID_AUTO_TEST_CASE(testObjectIdAttach) {
@@ -77,13 +77,13 @@ QPID_AUTO_TEST_CASE(testObjectIdAttach) {
std::stringstream out1;
out1 << oid;
- BOOST_CHECK_EQUAL(out1.str(), "10-20-0-MrSmith-0");
+ BOOST_CHECK_EQUAL(out1.str(), "10-20-0-MrSmith-0(GabbaGabbaHey)");
agent.setBanks(30, 40);
std::stringstream out2;
out2 << oid;
- BOOST_CHECK_EQUAL(out2.str(), "10-20-30-MrSmith-0");
+ BOOST_CHECK_EQUAL(out2.str(), "10-20-30-MrSmith-0(GabbaGabbaHey)");
}
QPID_AUTO_TEST_CASE(testConsoleObjectId) {
diff --git a/qpid/extras/qmf/src/py/qmf/console.py b/qpid/extras/qmf/src/py/qmf/console.py
index 96f26e205c..fa8708d717 100644
--- a/qpid/extras/qmf/src/py/qmf/console.py
+++ b/qpid/extras/qmf/src/py/qmf/console.py
@@ -141,6 +141,10 @@ class Object(object):
self._schema = schema
self._properties = []
self._statistics = []
+ self._currentTime = None
+ self._createTime = None
+ self._deleteTime = 0
+ self._objectId = None
if v2Map:
self.v2Init(v2Map, agentName)
return
@@ -150,11 +154,6 @@ class Object(object):
self._createTime = codec.read_uint64()
self._deleteTime = codec.read_uint64()
self._objectId = ObjectId(codec)
- else:
- self._currentTime = None
- self._createTime = None
- self._deleteTime = None
- self._objectId = None
if codec:
if prop:
notPresent = self._parsePresenceMasks(codec, schema)
@@ -184,7 +183,12 @@ class Object(object):
values = omap['_values']
for prop in self._schema.getProperties():
if prop.name in values:
- self._properties.append((prop, values[prop.name]))
+ if prop.type == 10: # Reference
+ self._properties.append((prop, ObjectId(values[prop.name], agentName=agentName)))
+ else:
+ self._properties.append((prop, values[prop.name]))
+ else:
+ self._properties.append((prop, None))
for stat in self._schema.getStatistics():
if stat.name in values:
self._statistics.append((stat, values[stat.name]))
@@ -211,6 +215,14 @@ class Object(object):
""" Return the broker from which this object was sent """
return self._broker
+ def getAgent(self):
+ """ Return the agent from which this object was sent """
+ return self._agent
+
+ def getV2RoutingKey(self):
+ """ Get the QMFv2 routing key to address this object """
+ return self._agent.getV2RoutingKey()
+
def getObjectId(self):
""" Return the object identifier for this object """
return self._objectId
@@ -356,7 +368,7 @@ class Object(object):
call['_arguments'] = argMap
dp = self._broker.amqpSession.delivery_properties()
- dp.routing_key = self._objectId.getAgentBank()
+ dp.routing_key = self.getV2RoutingKey()
mp = self._broker.amqpSession.message_properties()
mp.content_type = "amqp/map"
mp.user_id = self._broker.authUser
@@ -599,7 +611,8 @@ class Session:
self.brokers.append(broker)
if not self.manageConnections:
- agent = broker.getAgent(1,0)
+ broker._waitForStable()
+ agent = broker.getBrokerAgent()
if agent:
agent.getObjects(_class="agent")
return broker
@@ -933,6 +946,15 @@ class Session:
except:
return
+ ##
+ ## For now, ignore heartbeats from messaging brokers. We already have the "local-broker"
+ ## agent in our list.
+ ##
+ if '_vendor' in values and values['_vendor'] == 'apache.org' and \
+ '_product' in values and values['_product'] == 'qpidd':
+ broker._ageAgents()
+ return
+
agent = broker.getAgent(1, agentName)
if agent == None:
agent = Agent(broker, agentName, "QMFv2 Agent", True, interval)
@@ -1185,7 +1207,7 @@ class Session:
call['_arguments'] = args
dp = broker.amqpSession.delivery_properties()
- dp.routing_key = objectId.getAgentBank()
+ dp.routing_key = objectId.getV2RoutingKey()
mp = broker.amqpSession.message_properties()
mp.content_type = "amqp/map"
mp.user_id = broker.authUser
@@ -1821,6 +1843,8 @@ class Broker:
self.error = None
self.brokerId = None
self.connected = False
+ self.brokerAgent = None
+ self.brokerSupportsV2 = None
self.amqpSessionId = "%s.%d.%d" % (platform.uname()[1], os.getpid(), Broker.nextSeq)
Broker.nextSeq += 1
if self.session.manageConnections:
@@ -1859,6 +1883,9 @@ class Broker:
self.cv.release()
return None
+ def getBrokerAgent(self):
+ return self.brokerAgent
+
def getSessionId(self):
""" Get the identifier of the AMQP session to the broker """
return self.amqpSessionId
@@ -1915,7 +1942,6 @@ class Broker:
try:
self.cv.acquire()
self.agents = {}
- self.agents['0'] = Agent(self, 0, "BrokerAgent")
finally:
self.cv.release()
@@ -1968,22 +1994,38 @@ class Broker:
self.amqpSession.message_flow(destination="tdest", unit=1, value=0xFFFFFFFFL)
##
+ ## Check to see if the broker has QMFv2 exchanges configured
+ ##
+ direct_result = self.amqpSession.exchange_query("qmf.default.direct")
+ topic_result = self.amqpSession.exchange_query("qmf.default.topic")
+ self.brokerSupportsV2 = not (direct_result.not_found or topic_result.not_found)
+
+ try:
+ self.cv.acquire()
+ self.agents = {}
+ self.brokerAgent = Agent(self, 0, "BrokerAgent", isV2=self.brokerSupportsV2)
+ self.agents['0'] = self.brokerAgent
+ finally:
+ self.cv.release()
+
+ ##
## Set up connectivity for QMFv2
##
- self.v2_queue_name = "qmfc-v2-%s" % self.amqpSessionId
- self.amqpSession.queue_declare(queue=self.v2_queue_name, exclusive=True, auto_delete=True)
- self.amqpSession.exchange_bind(exchange="qmf.default.direct",
- queue=self.v2_queue_name, binding_key=self.v2_queue_name)
- self.amqpSession.exchange_bind(exchange="qmf.default.topic",
- queue=self.v2_queue_name, binding_key="agent.#")
- ## Other bindings here...
- self.amqpSession.message_subscribe(queue=self.v2_queue_name, destination="v2dest",
- accept_mode=self.amqpSession.accept_mode.none,
- acquire_mode=self.amqpSession.acquire_mode.pre_acquired)
- self.amqpSession.incoming("v2dest").listen(self._v2Cb, self._exceptionCb)
- self.amqpSession.message_set_flow_mode(destination="v2dest", flow_mode=1)
- self.amqpSession.message_flow(destination="v2dest", unit=0, value=0xFFFFFFFFL)
- self.amqpSession.message_flow(destination="v2dest", unit=1, value=0xFFFFFFFFL)
+ if self.brokerSupportsV2:
+ self.v2_queue_name = "qmfc-v2-%s" % self.amqpSessionId
+ self.amqpSession.queue_declare(queue=self.v2_queue_name, exclusive=True, auto_delete=True)
+ self.amqpSession.exchange_bind(exchange="qmf.default.direct",
+ queue=self.v2_queue_name, binding_key=self.v2_queue_name)
+ self.amqpSession.exchange_bind(exchange="qmf.default.topic",
+ queue=self.v2_queue_name, binding_key="agent.#")
+ ## Other bindings here...
+ self.amqpSession.message_subscribe(queue=self.v2_queue_name, destination="v2dest",
+ accept_mode=self.amqpSession.accept_mode.none,
+ acquire_mode=self.amqpSession.acquire_mode.pre_acquired)
+ self.amqpSession.incoming("v2dest").listen(self._v2Cb, self._exceptionCb)
+ self.amqpSession.message_set_flow_mode(destination="v2dest", flow_mode=1)
+ self.amqpSession.message_flow(destination="v2dest", unit=0, value=0xFFFFFFFFL)
+ self.amqpSession.message_flow(destination="v2dest", unit=1, value=0xFFFFFFFFL)
self.connected = True
self.session._handleBrokerConnect(self)
@@ -1992,7 +2034,8 @@ class Broker:
self._setHeader(codec, 'B')
msg = self._message(codec.encoded)
self._send(msg)
- self._v2SendAgentLocate()
+ if self.brokerSupportsV2:
+ self._v2SendAgentLocate()
except socket.error, e:
self.error = "Socket Error %s - %s" % (e.__class__.__name__, e)
@@ -2257,6 +2300,8 @@ class Broker:
## of the message.
##
agent_addr = ah['qmf.agent']
+ if agent_addr == 'broker':
+ agent_addr = '0'
if agent_addr in self.agents:
agent = self.agents[agent_addr]
agent._handleQmfV2Message(opcode, mp, ah, content)
@@ -2373,6 +2418,13 @@ class Agent:
return self.agentBank
+ def getV2RoutingKey(self):
+ self._checkClosed()
+ if self.agentBank == '0':
+ return 'broker'
+ return self.agentBank
+
+
def getObjects(self, notifiable=None, **kwargs):
""" Get a list of objects from QMF agents.
All arguments are passed by name(keyword).
@@ -2634,19 +2686,26 @@ class Agent:
else:
return
+ values = {}
+ if '_values' in content:
+ values = content['_values']
+
+ code = 7
+ text = "error"
+ if 'error_code' in values:
+ code = values['error_code']
+ if 'error_text' in values:
+ text = values['error_text']
+
pair = self.seqMgr._release(seq)
if pair == None:
return
- method, synchronous = pair
- code = 7
- text = ""
- if '_status_code' in content:
- code = content['_status_code']
- if '_status_text' in content:
- text = content['_status_text']
- else:
- text = content
+ if pair.__class__ == RequestContext:
+ pair.cancel(text)
+ return
+
+ method, synchronous = pair
result = MethodResult(code, text, {})
if synchronous:
@@ -2713,7 +2772,7 @@ class Agent:
# Construct and transmit the message
#
dp = self.broker.amqpSession.delivery_properties()
- dp.routing_key = self.agentBank
+ dp.routing_key = self.getV2RoutingKey()
mp = self.broker.amqpSession.message_properties()
mp.content_type = "amqp/map"
mp.user_id = self.broker.authUser
@@ -2812,8 +2871,12 @@ class RequestContext(object):
values = data['_values']
for key in values:
val = values[key]
- if key in self.selectors and val != self.selectors[key]:
- return
+ if key in self.selectors:
+ sel_val = self.selectors[key]
+ if sel_val.__class__ == ObjectId:
+ val = ObjectId(val, agentName=self.agent.getAgentBank())
+ if val != sel_val:
+ return
self.rawQueryResults.append(data)
@@ -2908,6 +2971,9 @@ class RequestContext(object):
self.agent._v2SendSchemaRequest(schemaId)
for result in queryResults:
+ key = result.getClassKey()
+ if key.getPackageName() == "org.apache.qpid.broker" and key.getClassName() == "agent":
+ self.agent.broker._updateAgent(result)
if self.notifiable:
self.notifiable(qmf_object=result)
else:
diff --git a/qpid/tools/src/py/qpid-route b/qpid/tools/src/py/qpid-route
index 0e1c078c2a..156ab696f1 100755
--- a/qpid/tools/src/py/qpid-route
+++ b/qpid/tools/src/py/qpid-route
@@ -73,12 +73,14 @@ class RouteManager:
self.remote = None
self.qmf = Session()
self.broker = self.qmf.addBroker(localBroker, _connTimeout)
+ self.broker._waitForStable()
+ self.agent = self.broker.getBrokerAgent()
def disconnect(self):
self.qmf.delBroker(self.broker)
def getLink(self):
- links = self.qmf.getObjects(_class="link")
+ links = self.agent.getObjects(_class="link")
for link in links:
if self.remote.match(link.host, link.port):
return link
@@ -89,7 +91,7 @@ class RouteManager:
if self.local.match(self.remote.host, self.remote.port):
raise Exception("Linking broker to itself is not permitted")
- brokers = self.qmf.getObjects(_class="broker")
+ brokers = self.agent.getObjects(_class="broker")
broker = brokers[0]
link = self.getLink()
if link == None:
@@ -105,7 +107,7 @@ class RouteManager:
def delLink(self, remoteBroker):
self.remote = BrokerURL(remoteBroker)
- brokers = self.qmf.getObjects(_class="broker")
+ brokers = self.agent.getObjects(_class="broker")
broker = brokers[0]
link = self.getLink()
if link == None:
@@ -116,7 +118,7 @@ class RouteManager:
print "Close method returned:", res.status, res.text
def listLinks(self):
- links = self.qmf.getObjects(_class="link")
+ links = self.agent.getObjects(_class="link")
if len(links) == 0:
print "No Links Found"
else:
@@ -128,7 +130,6 @@ class RouteManager:
(link.host, link.port, link.transport, YN(link.durable), link.state, link.lastError)
def mapRoutes(self):
- qmf = self.qmf
print
print "Finding Linked Brokers:"
@@ -139,13 +140,13 @@ class RouteManager:
added = True
while added:
added = False
- links = qmf.getObjects(_class="link")
+ links = self.qmf.getObjects(_class="link")
for link in links:
url = BrokerURL("%s:%d" % (link.host, link.port))
if url.name() not in brokerList:
print " %s..." % url.name(),
try:
- b = qmf.addBroker("%s:%d" % (link.host, link.port), _connTimeout)
+ b = self.qmf.addBroker("%s:%d" % (link.host, link.port), _connTimeout)
brokerList[url.name()] = b
added = True
print "Ok"
@@ -154,7 +155,7 @@ class RouteManager:
print
print "Dynamic Routes:"
- bridges = qmf.getObjects(_class="bridge", dynamic=True)
+ bridges = self.qmf.getObjects(_class="bridge", dynamic=True)
fedExchanges = []
for bridge in bridges:
if bridge.src not in fedExchanges:
@@ -182,7 +183,7 @@ class RouteManager:
print
print "Static Routes:"
- bridges = qmf.getObjects(_class="bridge", dynamic=False)
+ bridges = self.qmf.getObjects(_class="bridge", dynamic=False)
if len(bridges) == 0:
print " none found"
print
@@ -216,7 +217,7 @@ class RouteManager:
for broker in brokerList:
if broker != self.local.name():
- qmf.delBroker(brokerList[broker])
+ self.qmf.delBroker(brokerList[broker])
def addRoute(self, remoteBroker, exchange, routingKey, tag, excludes, dynamic=False):
@@ -228,7 +229,7 @@ class RouteManager:
if link == None:
raise Exception("Link failed to create")
- bridges = self.qmf.getObjects(_class="bridge")
+ bridges = self.agent.getObjects(_class="bridge")
for bridge in bridges:
if bridge.linkRef == link.getObjectId() and \
bridge.dest == exchange and bridge.key == routingKey and not bridge.srcIsQueue:
@@ -250,7 +251,7 @@ class RouteManager:
if link == None:
raise Exception("Link failed to create")
- bridges = self.qmf.getObjects(_class="bridge")
+ bridges = self.agent.getObjects(_class="bridge")
for bridge in bridges:
if bridge.linkRef == link.getObjectId() and \
bridge.dest == exchange and bridge.src == queue and bridge.srcIsQueue:
@@ -274,7 +275,7 @@ class RouteManager:
raise Exception("No link found from %s to %s" % (self.remote.name(), self.local.name()))
sys.exit(0)
- bridges = self.qmf.getObjects(_class="bridge")
+ bridges = self.agent.getObjects(_class="bridge")
for bridge in bridges:
if bridge.linkRef == link.getObjectId() and \
bridge.dest == exchange and bridge.src == queue and bridge.srcIsQueue:
@@ -304,7 +305,7 @@ class RouteManager:
raise Exception("No link found from %s to %s" % (self.remote.name(), self.local.name()))
sys.exit(0)
- bridges = self.qmf.getObjects(_class="bridge")
+ bridges = self.agent.getObjects(_class="bridge")
for bridge in bridges:
if bridge.linkRef == link.getObjectId() and bridge.dest == exchange and bridge.key == routingKey \
and bridge.dynamic == dynamic:
View Lorry Controller Status