diff options
Diffstat (limited to 'qpid/cpp/src/qmf')
21 files changed, 2050 insertions, 713 deletions
diff --git a/qpid/cpp/src/qmf/Agent.cpp b/qpid/cpp/src/qmf/AgentEngine.cpp index 6d59ae2750..d3204042d5 100644 --- a/qpid/cpp/src/qmf/Agent.cpp +++ b/qpid/cpp/src/qmf/AgentEngine.cpp @@ -17,7 +17,7 @@ * under the License. */ -#include "qmf/Agent.h" +#include "qmf/AgentEngine.h" #include "qmf/MessageImpl.h" #include "qmf/SchemaImpl.h" #include "qmf/Typecode.h" @@ -25,6 +25,7 @@ #include "qmf/ObjectIdImpl.h" #include "qmf/QueryImpl.h" #include "qmf/ValueImpl.h" +#include "qmf/Protocol.h" #include <qpid/framing/Buffer.h> #include <qpid/framing/Uuid.h> #include <qpid/framing/FieldTable.h> @@ -77,17 +78,17 @@ namespace qmf { AgentQueryContext() : schemaMethod(0) {} }; - class AgentImpl { + class AgentEngineImpl { public: - AgentImpl(char* label, bool internalStore); - ~AgentImpl(); + AgentEngineImpl(char* label, bool internalStore); + ~AgentEngineImpl(); - void setStoreDir(char* path); - void setTransferDir(char* path); + void setStoreDir(const char* path); + void setTransferDir(const char* path); void handleRcvMessage(Message& message); - bool getXmtMessage(Message& item); + bool getXmtMessage(Message& item) const; void popXmt(); - bool getEvent(AgentEvent& event); + bool getEvent(AgentEvent& event) const; void popEvent(); void newSession(); void startProtocol(); @@ -103,7 +104,7 @@ namespace qmf { void raiseEvent(Event& event); private: - Mutex lock; + mutable Mutex lock; Mutex addLock; string label; string queueName; @@ -134,13 +135,13 @@ namespace qmf { # define MA_BUFFER_SIZE 65536 char outputBuffer[MA_BUFFER_SIZE]; - struct SchemaClassKey { + struct AgentClassKey { string name; uint8_t hash[16]; - SchemaClassKey(const string& n, const uint8_t* h) : name(n) { + AgentClassKey(const string& n, const uint8_t* h) : name(n) { memcpy(hash, h, 16); } - SchemaClassKey(Buffer& buffer) { + AgentClassKey(Buffer& buffer) { buffer.getShortString(name); buffer.getBin128(hash); } @@ -149,8 +150,8 @@ namespace qmf { } }; - struct SchemaClassKeyComp { - bool operator() (const SchemaClassKey& lhs, const SchemaClassKey& rhs) const + struct AgentClassKeyComp { + bool operator() (const AgentClassKey& lhs, const AgentClassKey& rhs) const { if (lhs.name != rhs.name) return lhs.name < rhs.name; @@ -162,8 +163,8 @@ namespace qmf { } }; - typedef map<SchemaClassKey, SchemaObjectClassImpl*, SchemaClassKeyComp> ObjectClassMap; - typedef map<SchemaClassKey, SchemaEventClassImpl*, SchemaClassKeyComp> EventClassMap; + typedef map<AgentClassKey, SchemaObjectClassImpl*, AgentClassKeyComp> ObjectClassMap; + typedef map<AgentClassKey, SchemaEventClassImpl*, AgentClassKeyComp> EventClassMap; struct ClassMaps { ObjectClassMap objectClasses; @@ -172,8 +173,6 @@ namespace qmf { map<string, ClassMaps> packages; - bool checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq); - void encodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq = 0); AgentEventImpl::Ptr eventDeclareQueue(const string& queueName); AgentEventImpl::Ptr eventBind(const string& exchange, const string& queue, const string& key); AgentEventImpl::Ptr eventSetupComplete(); @@ -185,7 +184,7 @@ namespace qmf { void sendBufferLH(Buffer& buf, const string& destination, const string& routingKey); void sendPackageIndicationLH(const string& packageName); - void sendClassIndicationLH(ClassKind kind, const string& packageName, const SchemaClassKey& key); + void sendClassIndicationLH(ClassKind kind, const string& packageName, const AgentClassKey& key); void sendCommandCompleteLH(const string& exchange, const string& key, uint32_t seq, uint32_t code = 0, const string& text = "OK"); void sendMethodErrorLH(uint32_t sequence, const string& key, uint32_t code, const string& text=""); @@ -200,9 +199,9 @@ namespace qmf { }; } -const char* AgentImpl::QMF_EXCHANGE = "qpid.management"; -const char* AgentImpl::DIR_EXCHANGE = "amq.direct"; -const char* AgentImpl::BROKER_KEY = "broker"; +const char* AgentEngineImpl::QMF_EXCHANGE = "qpid.management"; +const char* AgentEngineImpl::DIR_EXCHANGE = "amq.direct"; +const char* AgentEngineImpl::BROKER_KEY = "broker"; #define STRING_REF(s) {if (!s.empty()) item.s = const_cast<char*>(s.c_str());} @@ -228,7 +227,7 @@ AgentEvent AgentEventImpl::copy() return item; } -AgentImpl::AgentImpl(char* _label, bool i) : +AgentEngineImpl::AgentEngineImpl(char* _label, bool i) : label(_label), queueName("qmfa-"), internalStore(i), nextTransientId(1), requestedBrokerBank(0), requestedAgentBank(0), assignedBrokerBank(0), assignedAgentBank(0), @@ -237,11 +236,11 @@ AgentImpl::AgentImpl(char* _label, bool i) : queueName += label; } -AgentImpl::~AgentImpl() +AgentEngineImpl::~AgentEngineImpl() { } -void AgentImpl::setStoreDir(char* path) +void AgentEngineImpl::setStoreDir(const char* path) { Mutex::ScopedLock _lock(lock); if (path) @@ -250,7 +249,7 @@ void AgentImpl::setStoreDir(char* path) storeDir.clear(); } -void AgentImpl::setTransferDir(char* path) +void AgentEngineImpl::setTransferDir(const char* path) { Mutex::ScopedLock _lock(lock); if (path) @@ -259,7 +258,7 @@ void AgentImpl::setTransferDir(char* path) transferDir.clear(); } -void AgentImpl::handleRcvMessage(Message& message) +void AgentEngineImpl::handleRcvMessage(Message& message) { Buffer inBuffer(message.body, message.length); uint8_t opcode; @@ -268,16 +267,20 @@ void AgentImpl::handleRcvMessage(Message& message) string replyToKey(message.replyKey ? message.replyKey : ""); string userId(message.userId ? message.userId : ""); - if (checkHeader(inBuffer, &opcode, &sequence)) { - if (opcode == 'a') handleAttachResponse(inBuffer); - else if (opcode == 'S') handleSchemaRequest(inBuffer, sequence, replyToExchange, replyToKey); - else if (opcode == 'x') handleConsoleAddedIndication(); - else if (opcode == 'G') handleGetQuery(inBuffer, sequence, replyToKey, userId); - else if (opcode == 'M') handleMethodRequest(inBuffer, sequence, replyToKey, userId); + while (Protocol::checkHeader(inBuffer, &opcode, &sequence)) { + if (opcode == Protocol::OP_ATTACH_RESPONSE) handleAttachResponse(inBuffer); + else if (opcode == Protocol::OP_SCHEMA_REQUEST) handleSchemaRequest(inBuffer, sequence, replyToExchange, replyToKey); + else if (opcode == Protocol::OP_CONSOLE_ADDED_INDICATION) handleConsoleAddedIndication(); + else if (opcode == Protocol::OP_GET_QUERY) handleGetQuery(inBuffer, sequence, replyToKey, userId); + else if (opcode == Protocol::OP_METHOD_REQUEST) handleMethodRequest(inBuffer, sequence, replyToKey, userId); + else { + QPID_LOG(error, "AgentEngineImpl::handleRcvMessage invalid opcode=" << opcode); + break; + } } } -bool AgentImpl::getXmtMessage(Message& item) +bool AgentEngineImpl::getXmtMessage(Message& item) const { Mutex::ScopedLock _lock(lock); if (xmtQueue.empty()) @@ -286,14 +289,14 @@ bool AgentImpl::getXmtMessage(Message& item) return true; } -void AgentImpl::popXmt() +void AgentEngineImpl::popXmt() { Mutex::ScopedLock _lock(lock); if (!xmtQueue.empty()) xmtQueue.pop_front(); } -bool AgentImpl::getEvent(AgentEvent& event) +bool AgentEngineImpl::getEvent(AgentEvent& event) const { Mutex::ScopedLock _lock(lock); if (eventQueue.empty()) @@ -302,14 +305,14 @@ bool AgentImpl::getEvent(AgentEvent& event) return true; } -void AgentImpl::popEvent() +void AgentEngineImpl::popEvent() { Mutex::ScopedLock _lock(lock); if (!eventQueue.empty()) eventQueue.pop_front(); } -void AgentImpl::newSession() +void AgentEngineImpl::newSession() { Mutex::ScopedLock _lock(lock); eventQueue.clear(); @@ -319,13 +322,13 @@ void AgentImpl::newSession() eventQueue.push_back(eventSetupComplete()); } -void AgentImpl::startProtocol() +void AgentEngineImpl::startProtocol() { Mutex::ScopedLock _lock(lock); char rawbuffer[512]; Buffer buffer(rawbuffer, 512); - encodeHeader(buffer, 'A'); + Protocol::encodeHeader(buffer, Protocol::OP_ATTACH_REQUEST); buffer.putShortString("qmfa"); systemId.encode(buffer); buffer.putLong(requestedBrokerBank); @@ -335,12 +338,12 @@ void AgentImpl::startProtocol() " reqAgent=" << requestedAgentBank); } -void AgentImpl::heartbeat() +void AgentEngineImpl::heartbeat() { Mutex::ScopedLock _lock(lock); Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'h'); + Protocol::encodeHeader(buffer, Protocol::OP_HEARTBEAT_INDICATION); buffer.putLongLong(uint64_t(Duration(now()))); stringstream key; key << "console.heartbeat." << assignedBrokerBank << "." << assignedAgentBank; @@ -348,8 +351,8 @@ void AgentImpl::heartbeat() QPID_LOG(trace, "SENT HeartbeatIndication"); } -void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, - const Value& argMap) +void AgentEngineImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, + const Value& argMap) { Mutex::ScopedLock _lock(lock); map<uint32_t, AgentQueryContext::Ptr>::iterator iter = contextMap.find(sequence); @@ -359,7 +362,7 @@ void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, contextMap.erase(iter); Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'm', context->sequence); + Protocol::encodeHeader(buffer, Protocol::OP_METHOD_RESPONSE, context->sequence); buffer.putLong(status); buffer.putMediumString(text); if (status == 0) { @@ -381,7 +384,7 @@ void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, QPID_LOG(trace, "SENT MethodResponse"); } -void AgentImpl::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) +void AgentEngineImpl::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) { Mutex::ScopedLock _lock(lock); map<uint32_t, AgentQueryContext::Ptr>::iterator iter = contextMap.find(sequence); @@ -390,7 +393,7 @@ void AgentImpl::queryResponse(uint32_t sequence, Object& object, bool prop, bool AgentQueryContext::Ptr context = iter->second; Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'g', context->sequence); + Protocol::encodeHeader(buffer, Protocol::OP_OBJECT_INDICATION, context->sequence); object.impl->encodeSchemaKey(buffer); object.impl->encodeManagedObjectData(buffer); @@ -403,7 +406,7 @@ void AgentImpl::queryResponse(uint32_t sequence, Object& object, bool prop, bool QPID_LOG(trace, "SENT ContentIndication"); } -void AgentImpl::queryComplete(uint32_t sequence) +void AgentEngineImpl::queryComplete(uint32_t sequence) { Mutex::ScopedLock _lock(lock); map<uint32_t, AgentQueryContext::Ptr>::iterator iter = contextMap.find(sequence); @@ -415,7 +418,7 @@ void AgentImpl::queryComplete(uint32_t sequence) sendCommandCompleteLH(context->exchange, context->key, context->sequence, 0, "OK"); } -void AgentImpl::registerClass(SchemaObjectClass* cls) +void AgentEngineImpl::registerClass(SchemaObjectClass* cls) { Mutex::ScopedLock _lock(lock); SchemaObjectClassImpl* impl = cls->impl; @@ -423,17 +426,17 @@ void AgentImpl::registerClass(SchemaObjectClass* cls) map<string, ClassMaps>::iterator iter = packages.find(impl->package); if (iter == packages.end()) { packages[impl->package] = ClassMaps(); - iter = packages.find(impl->package); + iter = packages.find(impl->getClassKey()->getPackageName()); // TODO: Indicate this package if connected } - SchemaClassKey key(impl->name, impl->getHash()); + AgentClassKey key(impl->getClassKey()->getClassName(), impl->getClassKey()->getHash()); iter->second.objectClasses[key] = impl; // TODO: Indicate this schema if connected. } -void AgentImpl::registerClass(SchemaEventClass* cls) +void AgentEngineImpl::registerClass(SchemaEventClass* cls) { Mutex::ScopedLock _lock(lock); SchemaEventClassImpl* impl = cls->impl; @@ -441,23 +444,23 @@ void AgentImpl::registerClass(SchemaEventClass* cls) map<string, ClassMaps>::iterator iter = packages.find(impl->package); if (iter == packages.end()) { packages[impl->package] = ClassMaps(); - iter = packages.find(impl->package); + iter = packages.find(impl->getClassKey()->getPackageName()); // TODO: Indicate this package if connected } - SchemaClassKey key(impl->name, impl->getHash()); + AgentClassKey key(impl->getClassKey()->getClassName(), impl->getClassKey()->getHash()); iter->second.eventClasses[key] = impl; // TODO: Indicate this schema if connected. } -const ObjectId* AgentImpl::addObject(Object&, uint64_t) +const ObjectId* AgentEngineImpl::addObject(Object&, uint64_t) { Mutex::ScopedLock _lock(lock); return 0; } -const ObjectId* AgentImpl::allocObjectId(uint64_t persistId) +const ObjectId* AgentEngineImpl::allocObjectId(uint64_t persistId) { Mutex::ScopedLock _lock(lock); uint16_t sequence = persistId ? 0 : bootSequence; @@ -467,41 +470,17 @@ const ObjectId* AgentImpl::allocObjectId(uint64_t persistId) return oid->envelope; } -const ObjectId* AgentImpl::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) +const ObjectId* AgentEngineImpl::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) { return allocObjectId(((uint64_t) persistIdHi) << 32 | (uint64_t) persistIdLo); } -void AgentImpl::raiseEvent(Event&) +void AgentEngineImpl::raiseEvent(Event&) { Mutex::ScopedLock _lock(lock); } -void AgentImpl::encodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq) -{ - buf.putOctet('A'); - buf.putOctet('M'); - buf.putOctet('3'); - buf.putOctet(opcode); - buf.putLong (seq); -} - -bool AgentImpl::checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq) -{ - if (buf.getSize() < 8) - return false; - - uint8_t h1 = buf.getOctet(); - uint8_t h2 = buf.getOctet(); - uint8_t h3 = buf.getOctet(); - - *opcode = buf.getOctet(); - *seq = buf.getLong(); - - return h1 == 'A' && h2 == 'M' && h3 == '3'; -} - -AgentEventImpl::Ptr AgentImpl::eventDeclareQueue(const string& name) +AgentEventImpl::Ptr AgentEngineImpl::eventDeclareQueue(const string& name) { AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::DECLARE_QUEUE)); event->name = name; @@ -509,8 +488,8 @@ AgentEventImpl::Ptr AgentImpl::eventDeclareQueue(const string& name) return event; } -AgentEventImpl::Ptr AgentImpl::eventBind(const string& exchange, const string& queue, - const string& key) +AgentEventImpl::Ptr AgentEngineImpl::eventBind(const string& exchange, const string& queue, + const string& key) { AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::BIND)); event->name = queue; @@ -520,14 +499,14 @@ AgentEventImpl::Ptr AgentImpl::eventBind(const string& exchange, const string& q return event; } -AgentEventImpl::Ptr AgentImpl::eventSetupComplete() +AgentEventImpl::Ptr AgentEngineImpl::eventSetupComplete() { AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::SETUP_COMPLETE)); return event; } -AgentEventImpl::Ptr AgentImpl::eventQuery(uint32_t num, const string& userId, const string& package, - const string& cls, boost::shared_ptr<ObjectId> oid) +AgentEventImpl::Ptr AgentEngineImpl::eventQuery(uint32_t num, const string& userId, const string& package, + const string& cls, boost::shared_ptr<ObjectId> oid) { AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::GET_QUERY)); event->sequence = num; @@ -538,9 +517,9 @@ AgentEventImpl::Ptr AgentImpl::eventQuery(uint32_t num, const string& userId, co return event; } -AgentEventImpl::Ptr AgentImpl::eventMethod(uint32_t num, const string& userId, const string& method, - boost::shared_ptr<ObjectId> oid, boost::shared_ptr<Value> argMap, - SchemaObjectClass* objectClass) +AgentEventImpl::Ptr AgentEngineImpl::eventMethod(uint32_t num, const string& userId, const string& method, + boost::shared_ptr<ObjectId> oid, boost::shared_ptr<Value> argMap, + SchemaObjectClass* objectClass) { AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::METHOD_CALL)); event->sequence = num; @@ -552,7 +531,7 @@ AgentEventImpl::Ptr AgentImpl::eventMethod(uint32_t num, const string& userId, c return event; } -void AgentImpl::sendBufferLH(Buffer& buf, const string& destination, const string& routingKey) +void AgentEngineImpl::sendBufferLH(Buffer& buf, const string& destination, const string& routingKey) { uint32_t length = buf.getPosition(); MessageImpl::Ptr message(new MessageImpl); @@ -567,19 +546,19 @@ void AgentImpl::sendBufferLH(Buffer& buf, const string& destination, const strin xmtQueue.push_back(message); } -void AgentImpl::sendPackageIndicationLH(const string& packageName) +void AgentEngineImpl::sendPackageIndicationLH(const string& packageName) { Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'p'); + Protocol::encodeHeader(buffer, Protocol::OP_PACKAGE_INDICATION); buffer.putShortString(packageName); sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); QPID_LOG(trace, "SENT PackageIndication: package_name=" << packageName); } -void AgentImpl::sendClassIndicationLH(ClassKind kind, const string& packageName, const SchemaClassKey& key) +void AgentEngineImpl::sendClassIndicationLH(ClassKind kind, const string& packageName, const AgentClassKey& key) { Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'q'); + Protocol::encodeHeader(buffer, Protocol::OP_CLASS_INDICATION); buffer.putOctet((int) kind); buffer.putShortString(packageName); buffer.putShortString(key.name); @@ -588,21 +567,21 @@ void AgentImpl::sendClassIndicationLH(ClassKind kind, const string& packageName, QPID_LOG(trace, "SENT ClassIndication: package_name=" << packageName << " class_name=" << key.name); } -void AgentImpl::sendCommandCompleteLH(const string& exchange, const string& replyToKey, - uint32_t sequence, uint32_t code, const string& text) +void AgentEngineImpl::sendCommandCompleteLH(const string& exchange, const string& replyToKey, + uint32_t sequence, uint32_t code, const string& text) { Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'z', sequence); + Protocol::encodeHeader(buffer, Protocol::OP_COMMAND_COMPLETE, sequence); buffer.putLong(code); buffer.putShortString(text); sendBufferLH(buffer, exchange, replyToKey); QPID_LOG(trace, "SENT CommandComplete: seq=" << sequence << " code=" << code << " text=" << text); } -void AgentImpl::sendMethodErrorLH(uint32_t sequence, const string& key, uint32_t code, const string& text) +void AgentEngineImpl::sendMethodErrorLH(uint32_t sequence, const string& key, uint32_t code, const string& text) { Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'm', sequence); + Protocol::encodeHeader(buffer, Protocol::OP_METHOD_RESPONSE, sequence); buffer.putLong(code); string fulltext; @@ -625,7 +604,7 @@ void AgentImpl::sendMethodErrorLH(uint32_t sequence, const string& key, uint32_t QPID_LOG(trace, "SENT MethodResponse: errorCode=" << code << " text=" << fulltext); } -void AgentImpl::handleAttachResponse(Buffer& inBuffer) +void AgentEngineImpl::handleAttachResponse(Buffer& inBuffer) { Mutex::ScopedLock _lock(lock); @@ -672,25 +651,25 @@ void AgentImpl::handleAttachResponse(Buffer& inBuffer) } } -void AgentImpl::handlePackageRequest(Buffer&) +void AgentEngineImpl::handlePackageRequest(Buffer&) { Mutex::ScopedLock _lock(lock); } -void AgentImpl::handleClassQuery(Buffer&) +void AgentEngineImpl::handleClassQuery(Buffer&) { Mutex::ScopedLock _lock(lock); } -void AgentImpl::handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, - const string& replyExchange, const string& replyKey) +void AgentEngineImpl::handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, + const string& replyExchange, const string& replyKey) { Mutex::ScopedLock _lock(lock); string rExchange(replyExchange); string rKey(replyKey); string packageName; inBuffer.getShortString(packageName); - SchemaClassKey key(inBuffer); + AgentClassKey key(inBuffer); if (rExchange.empty()) rExchange = QMF_EXCHANGE; @@ -710,7 +689,7 @@ void AgentImpl::handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, if (ocIter != cMap.objectClasses.end()) { SchemaObjectClassImpl* oImpl = ocIter->second; Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 's', sequence); + Protocol::encodeHeader(buffer, Protocol::OP_SCHEMA_RESPONSE, sequence); oImpl->encode(buffer); sendBufferLH(buffer, rExchange, rKey); QPID_LOG(trace, "SENT SchemaResponse: (object) package=" << packageName << " class=" << key.name); @@ -721,7 +700,7 @@ void AgentImpl::handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, if (ecIter != cMap.eventClasses.end()) { SchemaEventClassImpl* eImpl = ecIter->second; Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 's', sequence); + Protocol::encodeHeader(buffer, Protocol::OP_SCHEMA_RESPONSE, sequence); eImpl->encode(buffer); sendBufferLH(buffer, rExchange, rKey); QPID_LOG(trace, "SENT SchemaResponse: (event) package=" << packageName << " class=" << key.name); @@ -731,7 +710,7 @@ void AgentImpl::handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, sendCommandCompleteLH(rExchange, rKey, sequence, 1, "class not found"); } -void AgentImpl::handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId) +void AgentEngineImpl::handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId) { Mutex::ScopedLock _lock(lock); FieldTable ft; @@ -783,7 +762,7 @@ void AgentImpl::handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string eventQueue.push_back(eventQuery(contextNum, userId, pname, cname, oid)); } -void AgentImpl::handleMethodRequest(Buffer& buffer, uint32_t sequence, const string& replyTo, const string& userId) +void AgentEngineImpl::handleMethodRequest(Buffer& buffer, uint32_t sequence, const string& replyTo, const string& userId) { Mutex::ScopedLock _lock(lock); string pname; @@ -791,7 +770,7 @@ void AgentImpl::handleMethodRequest(Buffer& buffer, uint32_t sequence, const str ObjectIdImpl* oidImpl = new ObjectIdImpl(buffer); boost::shared_ptr<ObjectId> oid(oidImpl->envelope); buffer.getShortString(pname); - SchemaClassKey classKey(buffer); + AgentClassKey classKey(buffer); buffer.getShortString(method); map<string, ClassMaps>::const_iterator pIter = packages.find(pname); @@ -842,7 +821,7 @@ void AgentImpl::handleMethodRequest(Buffer& buffer, uint32_t sequence, const str eventQueue.push_back(eventMethod(contextNum, userId, method, oid, argMap, schema->envelope)); } -void AgentImpl::handleConsoleAddedIndication() +void AgentEngineImpl::handleConsoleAddedIndication() { Mutex::ScopedLock _lock(lock); } @@ -851,108 +830,25 @@ void AgentImpl::handleConsoleAddedIndication() // Wrappers //================================================================== -Agent::Agent(char* label, bool internalStore) -{ - impl = new AgentImpl(label, internalStore); -} - -Agent::~Agent() -{ - delete impl; -} - -void Agent::setStoreDir(char* path) -{ - impl->setStoreDir(path); -} - -void Agent::setTransferDir(char* path) -{ - impl->setTransferDir(path); -} - -void Agent::handleRcvMessage(Message& message) -{ - impl->handleRcvMessage(message); -} - -bool Agent::getXmtMessage(Message& item) -{ - return impl->getXmtMessage(item); -} - -void Agent::popXmt() -{ - impl->popXmt(); -} - -bool Agent::getEvent(AgentEvent& event) -{ - return impl->getEvent(event); -} - -void Agent::popEvent() -{ - impl->popEvent(); -} - -void Agent::newSession() -{ - impl->newSession(); -} - -void Agent::startProtocol() -{ - impl->startProtocol(); -} - -void Agent::heartbeat() -{ - impl->heartbeat(); -} - -void Agent::methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments) -{ - impl->methodResponse(sequence, status, text, arguments); -} - -void Agent::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) -{ - impl->queryResponse(sequence, object, prop, stat); -} - -void Agent::queryComplete(uint32_t sequence) -{ - impl->queryComplete(sequence); -} - -void Agent::registerClass(SchemaObjectClass* cls) -{ - impl->registerClass(cls); -} - -void Agent::registerClass(SchemaEventClass* cls) -{ - impl->registerClass(cls); -} - -const ObjectId* Agent::addObject(Object& obj, uint64_t persistId) -{ - return impl->addObject(obj, persistId); -} - -const ObjectId* Agent::allocObjectId(uint64_t persistId) -{ - return impl->allocObjectId(persistId); -} - -const ObjectId* Agent::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) -{ - return impl->allocObjectId(persistIdLo, persistIdHi); -} - -void Agent::raiseEvent(Event& event) -{ - impl->raiseEvent(event); -} +AgentEngine::AgentEngine(char* label, bool internalStore) { impl = new AgentEngineImpl(label, internalStore); } +AgentEngine::~AgentEngine() { delete impl; } +void AgentEngine::setStoreDir(const char* path) { impl->setStoreDir(path); } +void AgentEngine::setTransferDir(const char* path) { impl->setTransferDir(path); } +void AgentEngine::handleRcvMessage(Message& message) { impl->handleRcvMessage(message); } +bool AgentEngine::getXmtMessage(Message& item) const { return impl->getXmtMessage(item); } +void AgentEngine::popXmt() { impl->popXmt(); } +bool AgentEngine::getEvent(AgentEvent& event) const { return impl->getEvent(event); } +void AgentEngine::popEvent() { impl->popEvent(); } +void AgentEngine::newSession() { impl->newSession(); } +void AgentEngine::startProtocol() { impl->startProtocol(); } +void AgentEngine::heartbeat() { impl->heartbeat(); } +void AgentEngine::methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments) { impl->methodResponse(sequence, status, text, arguments); } +void AgentEngine::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) { impl->queryResponse(sequence, object, prop, stat); } +void AgentEngine::queryComplete(uint32_t sequence) { impl->queryComplete(sequence); } +void AgentEngine::registerClass(SchemaObjectClass* cls) { impl->registerClass(cls); } +void AgentEngine::registerClass(SchemaEventClass* cls) { impl->registerClass(cls); } +const ObjectId* AgentEngine::addObject(Object& obj, uint64_t persistId) { return impl->addObject(obj, persistId); } +const ObjectId* AgentEngine::allocObjectId(uint64_t persistId) { return impl->allocObjectId(persistId); } +const ObjectId* AgentEngine::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) { return impl->allocObjectId(persistIdLo, persistIdHi); } +void AgentEngine::raiseEvent(Event& event) { impl->raiseEvent(event); } diff --git a/qpid/cpp/src/qmf/Agent.h b/qpid/cpp/src/qmf/AgentEngine.h index d8f784e9d8..c88ef33657 100644 --- a/qpid/cpp/src/qmf/Agent.h +++ b/qpid/cpp/src/qmf/AgentEngine.h @@ -1,5 +1,5 @@ -#ifndef _QmfAgent_ -#define _QmfAgent_ +#ifndef _QmfAgentEngine_ +#define _QmfAgentEngine_ /* * Licensed to the Apache Software Foundation (ASF) under one @@ -64,15 +64,15 @@ namespace qmf { SchemaObjectClass* objectClass; // (METHOD_CALL) }; - class AgentImpl; + class AgentEngineImpl; /** - * Agent - Protocol engine for the QMF agent + * AgentEngine - Protocol engine for the QMF agent */ - class Agent { + class AgentEngine { public: - Agent(char* label, bool internalStore=true); - ~Agent(); + AgentEngine(char* label, bool internalStore=true); + ~AgentEngine(); /** * Configure the directory path for storing persistent data. @@ -80,7 +80,7 @@ namespace qmf { * created, written, and read. If NULL, no persistent storage will be * attempted. */ - void setStoreDir(char* path); + void setStoreDir(const char* path); /** * Configure the directory path for files transferred over QMF. @@ -88,7 +88,7 @@ namespace qmf { * created, deleted, written, and read. If NULL, file transfers shall not * be permitted. */ - void setTransferDir(char* path); + void setTransferDir(const char* path); /** * Pass messages received from the AMQP session to the Agent engine. @@ -101,7 +101,7 @@ namespace qmf { *@param item The Message structure describing the message to be produced. *@return true if the Message is valid, false if there are no messages to send. */ - bool getXmtMessage(Message& item); + bool getXmtMessage(Message& item) const; /** * Remove and discard one message from the head of the transmit queue. @@ -113,7 +113,7 @@ namespace qmf { *@param event The event iff the return value is true *@return true if event is valid, false if there are no events to process */ - bool getEvent(AgentEvent& event); + bool getEvent(AgentEvent& event) const; /** * Remove and discard one event from the head of the event queue. @@ -182,11 +182,12 @@ namespace qmf { *@return The objectId of the managed object. */ const ObjectId* addObject(Object& obj, uint64_t persistId); + // const ObjectId* addObject(Object& obj, uint32_t persistIdLo, uint32_t persistIdHi); /** - * Allocate an objecc-id for an object that will be managed by the application. + * Allocate an object-id for an object that will be managed by the application. *@param persistId A unique non-zero value if the object-id is to be persistent. - @return The objectId structure for the allocated ID. + *@return The objectId structure for the allocated ID. */ const ObjectId* allocObjectId(uint64_t persistId); const ObjectId* allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi); @@ -198,7 +199,7 @@ namespace qmf { void raiseEvent(Event& event); private: - AgentImpl* impl; + AgentEngineImpl* impl; }; } diff --git a/qpid/cpp/src/qmf/ConnectionSettingsImpl.cpp b/qpid/cpp/src/qmf/ConnectionSettingsImpl.cpp new file mode 100644 index 0000000000..034ab18395 --- /dev/null +++ b/qpid/cpp/src/qmf/ConnectionSettingsImpl.cpp @@ -0,0 +1,323 @@ +/* + * 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 "qmf/ConnectionSettingsImpl.h" +#include "qmf/Typecode.h" + +using namespace std; +using namespace qmf; +using namespace qpid; + +const string attrProtocol("protocol"); +const string attrHost("host"); +const string attrPort("port"); +const string attrVirtualhost("virtualhost"); +const string attrUsername("username"); +const string attrPassword("password"); +const string attrMechanism("mechanism"); +const string attrLocale("locale"); +const string attrHeartbeat("heartbeat"); +const string attrMaxChannels("maxChannels"); +const string attrMaxFrameSize("maxFrameSize"); +const string attrBounds("bounds"); +const string attrTcpNoDelay("tcpNoDelay"); +const string attrService("service"); +const string attrMinSsf("minSsf"); +const string attrMaxSsf("maxSsf"); +const string attrRetryDelayMin("retryDelayMin"); +const string attrRetryDelayMax("retryDelayMax"); +const string attrRetryDelayFactor("retryDelayFactor"); + +ConnectionSettingsImpl::ConnectionSettingsImpl(ConnectionSettings* e) : + envelope(e), retryDelayMin(1), retryDelayMax(64), retryDelayFactor(2) +{ +} + +ConnectionSettingsImpl::ConnectionSettingsImpl(ConnectionSettings* e, const string& /*url*/) : + envelope(e), retryDelayMin(1), retryDelayMax(64), retryDelayFactor(2) +{ + // TODO: Parse the URL +} + +void ConnectionSettingsImpl::setAttr(const string& key, const Value& value) +{ + if (key == attrProtocol) clientSettings.protocol = value.asString(); + else if (key == attrHost) clientSettings.host = value.asString(); + else if (key == attrPort) clientSettings.port = value.asUint(); + else if (key == attrVirtualhost) clientSettings.virtualhost = value.asString(); + else if (key == attrUsername) clientSettings.username = value.asString(); + else if (key == attrPassword) clientSettings.password = value.asString(); + else if (key == attrMechanism) clientSettings.mechanism = value.asString(); + else if (key == attrLocale) clientSettings.locale = value.asString(); + else if (key == attrHeartbeat) clientSettings.heartbeat = value.asUint(); + else if (key == attrMaxChannels) clientSettings.maxChannels = value.asUint(); + else if (key == attrMaxFrameSize) clientSettings.maxFrameSize = value.asUint(); + else if (key == attrBounds) clientSettings.bounds = value.asUint(); + else if (key == attrTcpNoDelay) clientSettings.tcpNoDelay = value.asBool(); + else if (key == attrService) clientSettings.service = value.asString(); + else if (key == attrMinSsf) clientSettings.minSsf = value.asUint(); + else if (key == attrMaxSsf) clientSettings.maxSsf = value.asUint(); + + else if (key == attrRetryDelayMin) retryDelayMin = value.asUint(); + else if (key == attrRetryDelayMax) retryDelayMax = value.asUint(); + else if (key == attrRetryDelayFactor) retryDelayFactor = value.asUint(); +} + +Value ConnectionSettingsImpl::getAttr(const string& key) const +{ + Value strval(TYPE_LSTR); + Value intval(TYPE_UINT32); + Value boolval(TYPE_BOOL); + + if (key == attrProtocol) { + strval.setString(clientSettings.protocol.c_str()); + return strval; + } + + if (key == attrHost) { + strval.setString(clientSettings.host.c_str()); + return strval; + } + + if (key == attrPort) { + intval.setUint(clientSettings.port); + return intval; + } + + if (key == attrVirtualhost) { + strval.setString(clientSettings.virtualhost.c_str()); + return strval; + } + + if (key == attrUsername) { + strval.setString(clientSettings.username.c_str()); + return strval; + } + + if (key == attrPassword) { + strval.setString(clientSettings.password.c_str()); + return strval; + } + + if (key == attrMechanism) { + strval.setString(clientSettings.mechanism.c_str()); + return strval; + } + + if (key == attrLocale) { + strval.setString(clientSettings.locale.c_str()); + return strval; + } + + if (key == attrHeartbeat) { + intval.setUint(clientSettings.heartbeat); + return intval; + } + + if (key == attrMaxChannels) { + intval.setUint(clientSettings.maxChannels); + return intval; + } + + if (key == attrMaxFrameSize) { + intval.setUint(clientSettings.maxFrameSize); + return intval; + } + + if (key == attrBounds) { + intval.setUint(clientSettings.bounds); + return intval; + } + + if (key == attrTcpNoDelay) { + boolval.setBool(clientSettings.tcpNoDelay); + return boolval; + } + + if (key == attrService) { + strval.setString(clientSettings.service.c_str()); + return strval; + } + + if (key == attrMinSsf) { + intval.setUint(clientSettings.minSsf); + return intval; + } + + if (key == attrMaxSsf) { + intval.setUint(clientSettings.maxSsf); + return intval; + } + + if (key == attrRetryDelayMin) { + intval.setUint(retryDelayMin); + return intval; + } + + if (key == attrRetryDelayMax) { + intval.setUint(retryDelayMax); + return intval; + } + + if (key == attrRetryDelayFactor) { + intval.setUint(retryDelayFactor); + return intval; + } + + return strval; +} + +const string& ConnectionSettingsImpl::getAttrString() const +{ + // TODO: build and return attribute string + return attrString; +} + +void ConnectionSettingsImpl::transportTcp(uint16_t port) +{ + clientSettings.protocol = "tcp"; + clientSettings.port = port; +} + +void ConnectionSettingsImpl::transportSsl(uint16_t port) +{ + clientSettings.protocol = "ssl"; + clientSettings.port = port; +} + +void ConnectionSettingsImpl::transportRdma(uint16_t port) +{ + clientSettings.protocol = "rdma"; + clientSettings.port = port; +} + +void ConnectionSettingsImpl::authAnonymous(const string& username) +{ + clientSettings.mechanism = "ANONYMOUS"; + clientSettings.username = username; +} + +void ConnectionSettingsImpl::authPlain(const string& username, const string& password) +{ + clientSettings.mechanism = "PLAIN"; + clientSettings.username = username; + clientSettings.password = password; +} + +void ConnectionSettingsImpl::authGssapi(const string& serviceName, uint32_t minSsf, uint32_t maxSsf) +{ + clientSettings.mechanism = "GSSAPI"; + clientSettings.service = serviceName; + clientSettings.minSsf = minSsf; + clientSettings.maxSsf = maxSsf; +} + +void ConnectionSettingsImpl::setRetry(int delayMin, int delayMax, int delayFactor) +{ + retryDelayMin = delayMin; + retryDelayMax = delayMax; + retryDelayFactor = delayFactor; +} + +const client::ConnectionSettings& ConnectionSettingsImpl::getClientSettings() const +{ + return clientSettings; +} + +void ConnectionSettingsImpl::getRetrySettings(int* min, int* max, int* factor) const +{ + *min = retryDelayMin; + *max = retryDelayMax; + *factor = retryDelayFactor; +} + +//================================================================== +// Wrappers +//================================================================== + +ConnectionSettings::ConnectionSettings(const ConnectionSettings& from) +{ + impl = new ConnectionSettingsImpl(*from.impl); +} + +ConnectionSettings::ConnectionSettings() +{ + impl = new ConnectionSettingsImpl(this); +} + +ConnectionSettings::ConnectionSettings(const char* url) +{ + impl = new ConnectionSettingsImpl(this, url); +} + +ConnectionSettings::~ConnectionSettings() +{ + delete impl; +} + +void ConnectionSettings::setAttr(const char* key, const Value& value) +{ + impl->setAttr(key, value); +} + +Value ConnectionSettings::getAttr(const char* key) const +{ + return impl->getAttr(key); +} + +const char* ConnectionSettings::getAttrString() const +{ + return impl->getAttrString().c_str(); +} + +void ConnectionSettings::transportTcp(uint16_t port) +{ + impl->transportTcp(port); +} + +void ConnectionSettings::transportSsl(uint16_t port) +{ + impl->transportSsl(port); +} + +void ConnectionSettings::transportRdma(uint16_t port) +{ + impl->transportRdma(port); +} + +void ConnectionSettings::authAnonymous(const char* username) +{ + impl->authAnonymous(username); +} + +void ConnectionSettings::authPlain(const char* username, const char* password) +{ + impl->authPlain(username, password); +} + +void ConnectionSettings::authGssapi(const char* serviceName, uint32_t minSsf, uint32_t maxSsf) +{ + impl->authGssapi(serviceName, minSsf, maxSsf); +} + +void ConnectionSettings::setRetry(int delayMin, int delayMax, int delayFactor) +{ + impl->setRetry(delayMin, delayMax, delayFactor); +} + diff --git a/qpid/cpp/src/qmf/ConnectionSettingsImpl.h b/qpid/cpp/src/qmf/ConnectionSettingsImpl.h new file mode 100644 index 0000000000..a177233cf3 --- /dev/null +++ b/qpid/cpp/src/qmf/ConnectionSettingsImpl.h @@ -0,0 +1,60 @@ +#ifndef _QmfConnectionSettingsImpl_ +#define _QmfConnectionSettingsImpl_ + +/* + * 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 "qmf/ConnectionSettings.h" +#include "qmf/Value.h" +#include "qpid/client/ConnectionSettings.h" +#include <string> +#include <map> + +namespace qmf { + + class ConnectionSettingsImpl { + ConnectionSettings* envelope; + qpid::client::ConnectionSettings clientSettings; + mutable std::string attrString; + int retryDelayMin; + int retryDelayMax; + int retryDelayFactor; + + public: + ConnectionSettingsImpl(ConnectionSettings* e); + ConnectionSettingsImpl(ConnectionSettings* e, const std::string& url); + ~ConnectionSettingsImpl() {} + void setAttr(const std::string& key, const Value& value); + Value getAttr(const std::string& key) const; + const std::string& getAttrString() const; + void transportTcp(uint16_t port); + void transportSsl(uint16_t port); + void transportRdma(uint16_t port); + void authAnonymous(const std::string& username); + void authPlain(const std::string& username, const std::string& password); + void authGssapi(const std::string& serviceName, uint32_t minSsf, uint32_t maxSsf); + void setRetry(int delayMin, int delayMax, int delayFactor); + + const qpid::client::ConnectionSettings& getClientSettings() const; + void getRetrySettings(int* delayMin, int* delayMax, int* delayFactor) const; + }; + +} + +#endif diff --git a/qpid/cpp/src/qmf/Console.h b/qpid/cpp/src/qmf/Console.h deleted file mode 100644 index de7949e1de..0000000000 --- a/qpid/cpp/src/qmf/Console.h +++ /dev/null @@ -1,82 +0,0 @@ -#ifndef _QmfConsole_ -#define _QmfConsole_ - -/* - * 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 <qmf/ManagedConnection.h> -#include <qmf/Agent.h> -#include <qmf/Broker.h> -#include <qmf/Package.h> -#include <qmf/SchemaClassTable.h> -#include <qmf/Object.h> -#include <qmf/ConsoleHandler.h> -#include <set> -#include <vector> -#include <string> - -namespace qmf { - - struct ConsoleSettings { - bool rcvObjects; - bool rcvEvents; - bool rcvHeartbeats; - bool userBindings; - uint32_t methodTimeout; - uint32_t getTimeout; - - ConsoleSettings() : - rcvObjects(true), - rcvEvents(true), - rcvHeartbeats(true), - userBindings(false), - methodTimeout(20), - getTimeout(20) {} - }; - - class Console { - public: - Console(ConsoleHandler* handler = 0, ConsoleSettings settings = ConsoleSettings()); - ~Console(); - - Broker* addConnection(ManagedConnection& connection); - void delConnection(Broker* broker); - void delConnection(ManagedConnection& connection); - - const PackageMap& getPackages() const; - - void bindPackage(const Package& package); - void bindPackage(const std::string& packageName); - void bindClass(const SchemaClass& otype); - void bindClass(const std::string& packageName, const std::string& className); - - void getAgents(std::set<Agent>& agents, Broker* = 0); - void getObjects(std::vector<Object>& objects, const std::string& typeName, - const std::string& packageName = "", - Broker* broker = 0, - Agent* agent = 0); - void getObjects(std::vector<Object>& objects, - const std::map<std::string, std::string>& query, - Broker* broker = 0, - Agent* agent = 0); - }; -} - -#endif - diff --git a/qpid/cpp/src/qmf/ConsoleEngine.cpp b/qpid/cpp/src/qmf/ConsoleEngine.cpp new file mode 100644 index 0000000000..3d1b378b68 --- /dev/null +++ b/qpid/cpp/src/qmf/ConsoleEngine.cpp @@ -0,0 +1,886 @@ +/* + * 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 "qmf/ConsoleEngine.h" +#include "qmf/MessageImpl.h" +#include "qmf/SchemaImpl.h" +#include "qmf/Typecode.h" +#include "qmf/ObjectImpl.h" +#include "qmf/ObjectIdImpl.h" +#include "qmf/QueryImpl.h" +#include "qmf/ValueImpl.h" +#include "qmf/Protocol.h" +#include "qmf/SequenceManager.h" +#include <qpid/framing/Buffer.h> +#include <qpid/framing/Uuid.h> +#include <qpid/framing/FieldTable.h> +#include <qpid/framing/FieldValue.h> +#include <qpid/sys/Mutex.h> +#include <qpid/log/Statement.h> +#include <qpid/sys/Time.h> +#include <string.h> +#include <string> +#include <deque> +#include <map> +#include <vector> +#include <iostream> +#include <fstream> +#include <boost/shared_ptr.hpp> + +using namespace std; +using namespace qmf; +using namespace qpid::framing; +using namespace qpid::sys; + +namespace qmf { + + struct MethodResponseImpl { + typedef boost::shared_ptr<MethodResponseImpl> Ptr; + MethodResponse* envelope; + uint32_t status; + auto_ptr<Value> exception; + auto_ptr<Value> arguments; + + MethodResponseImpl(Buffer& buf); + ~MethodResponseImpl() {} + uint32_t getStatus() const { return status; } + const Value* getException() const { return exception.get(); } + const Value* getArgs() const { return arguments.get(); } + }; + + struct ConsoleEventImpl { + typedef boost::shared_ptr<ConsoleEventImpl> Ptr; + ConsoleEvent::EventKind kind; + boost::shared_ptr<AgentProxyImpl> agent; + string name; + boost::shared_ptr<SchemaClassKey> classKey; + Object* object; + void* context; + Event* event; + uint64_t timestamp; + uint32_t methodHandle; + MethodResponseImpl::Ptr methodResponse; + + ConsoleEventImpl(ConsoleEvent::EventKind k) : + kind(k), object(0), context(0), event(0), timestamp(0), methodHandle(0) {} + ~ConsoleEventImpl() {} + ConsoleEvent copy(); + }; + + struct BrokerEventImpl { + typedef boost::shared_ptr<BrokerEventImpl> Ptr; + BrokerEvent::EventKind kind; + string name; + string exchange; + string bindingKey; + + BrokerEventImpl(BrokerEvent::EventKind k) : kind(k) {} + ~BrokerEventImpl() {} + BrokerEvent copy(); + }; + + class BrokerProxyImpl : public SequenceContext { + public: + typedef boost::shared_ptr<BrokerProxyImpl> Ptr; + + BrokerProxyImpl(BrokerProxy* e, ConsoleEngine& _console); + ~BrokerProxyImpl() {} + + void sessionOpened(SessionHandle& sh); + void sessionClosed(); + void startProtocol(); + + void sendBufferLH(Buffer& buf, const string& destination, const string& routingKey); + void handleRcvMessage(Message& message); + bool getXmtMessage(Message& item) const; + void popXmt(); + + bool getEvent(BrokerEvent& event) const; + void popEvent(); + + // From SequenceContext + void complete(); + + void addBinding(const string& exchange, const string& key); + + private: + mutable Mutex lock; + BrokerProxy* envelope; + ConsoleEngineImpl* console; + string queueName; + Uuid brokerId; + SequenceManager seqMgr; + uint32_t requestsOutstanding; + bool topicBound; + deque<MessageImpl::Ptr> xmtQueue; + deque<BrokerEventImpl::Ptr> eventQueue; + +# define MA_BUFFER_SIZE 65536 + char outputBuffer[MA_BUFFER_SIZE]; + + BrokerEventImpl::Ptr eventDeclareQueue(const string& queueName); + BrokerEventImpl::Ptr eventBind(const string& exchange, const string& queue, const string& key); + BrokerEventImpl::Ptr eventSetupComplete(); + BrokerEventImpl::Ptr eventStable(); + + void handleBrokerResponse(Buffer& inBuffer, uint32_t seq); + void handlePackageIndication(Buffer& inBuffer, uint32_t seq); + void handleCommandComplete(Buffer& inBuffer, uint32_t seq); + void handleClassIndication(Buffer& inBuffer, uint32_t seq); + void handleMethodResponse(Buffer& inBuffer, uint32_t seq); + void handleHeartbeatIndication(Buffer& inBuffer, uint32_t seq); + void handleEventIndication(Buffer& inBuffer, uint32_t seq); + void handleSchemaResponse(Buffer& inBuffer, uint32_t seq); + void handleObjectIndication(Buffer& inBuffer, uint32_t seq, bool prop, bool stat); + void incOutstandingLH(); + void decOutstanding(); + }; + + struct AgentProxyImpl { + typedef boost::shared_ptr<AgentProxyImpl> Ptr; + AgentProxy* envelope; + ConsoleEngineImpl* console; + + AgentProxyImpl(AgentProxy* e, ConsoleEngine& _console) : + envelope(e), console(_console.impl) {} + ~AgentProxyImpl() {} + }; + + class ConsoleEngineImpl { + public: + ConsoleEngineImpl(ConsoleEngine* e, const ConsoleSettings& settings = ConsoleSettings()); + ~ConsoleEngineImpl(); + + bool getEvent(ConsoleEvent& event) const; + void popEvent(); + + void addConnection(BrokerProxy& broker, void* context); + void delConnection(BrokerProxy& broker); + + uint32_t packageCount() const; + const string& getPackageName(uint32_t idx) const; + + uint32_t classCount(const char* packageName) const; + const SchemaClassKey* getClass(const char* packageName, uint32_t idx) const; + + ClassKind getClassKind(const SchemaClassKey* key) const; + const SchemaObjectClass* getObjectClass(const SchemaClassKey* key) const; + const SchemaEventClass* getEventClass(const SchemaClassKey* key) const; + + void bindPackage(const char* packageName); + void bindClass(const SchemaClassKey* key); + void bindClass(const char* packageName, const char* className); + + uint32_t agentCount() const; + const AgentProxy* getAgent(uint32_t idx) const; + + void sendQuery(const Query& query, void* context); + + /* + void startSync(const Query& query, void* context, SyncQuery& sync); + void touchSync(SyncQuery& sync); + void endSync(SyncQuery& sync); + */ + + private: + friend class BrokerProxyImpl; + ConsoleEngine* envelope; + const ConsoleSettings& settings; + mutable Mutex lock; + deque<ConsoleEventImpl::Ptr> eventQueue; + vector<BrokerProxyImpl*> brokerList; + vector<pair<string, string> > bindingList; // exchange/key (empty exchange => QMF_EXCHANGE) + + // Declare a compare class for the class maps that compares the dereferenced + // class key pointers. The default behavior would be to compare the pointer + // addresses themselves. + struct KeyCompare { + bool operator()(const SchemaClassKeyImpl* left, const SchemaClassKeyImpl* right) const { + return *left < *right; + } + }; + + typedef map<const SchemaClassKeyImpl*, SchemaObjectClassImpl::Ptr, KeyCompare> ObjectClassList; + typedef map<const SchemaClassKeyImpl*, SchemaEventClassImpl::Ptr, KeyCompare> EventClassList; + typedef map<string, pair<ObjectClassList, EventClassList> > PackageList; + + PackageList packages; + + void learnPackage(const string& packageName); + void learnClass(SchemaObjectClassImpl::Ptr cls); + void learnClass(SchemaEventClassImpl::Ptr cls); + bool haveClass(const SchemaClassKeyImpl& key) const; + }; +} + +namespace { +const char* QMF_EXCHANGE = "qpid.management"; +const char* DIR_EXCHANGE = "amq.direct"; +const char* BROKER_KEY = "broker"; +} + +#define STRING_REF(s) {if (!s.empty()) item.s = const_cast<char*>(s.c_str());} + +ConsoleEvent ConsoleEventImpl::copy() +{ + ConsoleEvent item; + + ::memset(&item, 0, sizeof(ConsoleEvent)); + item.kind = kind; + item.agent = agent.get() ? agent->envelope : 0; + item.classKey = classKey.get(); + item.object = object; + item.context = context; + item.event = event; + item.timestamp = timestamp; + item.methodHandle = methodHandle; + item.methodResponse = methodResponse.get() ? methodResponse->envelope : 0; + + STRING_REF(name); + + return item; +} + +BrokerEvent BrokerEventImpl::copy() +{ + BrokerEvent item; + + ::memset(&item, 0, sizeof(BrokerEvent)); + item.kind = kind; + + STRING_REF(name); + STRING_REF(exchange); + STRING_REF(bindingKey); + + return item; +} + +BrokerProxyImpl::BrokerProxyImpl(BrokerProxy* e, ConsoleEngine& _console) : + envelope(e), console(_console.impl), queueName("qmfc-") +{ + // TODO: Give the queue name a unique suffix +} + +void BrokerProxyImpl::sessionOpened(SessionHandle& /*sh*/) +{ + Mutex::ScopedLock _lock(lock); + eventQueue.clear(); + xmtQueue.clear(); + eventQueue.push_back(eventDeclareQueue(queueName)); + eventQueue.push_back(eventBind(DIR_EXCHANGE, queueName, queueName)); + eventQueue.push_back(eventSetupComplete()); + + // TODO: Store session handle +} + +void BrokerProxyImpl::sessionClosed() +{ + Mutex::ScopedLock _lock(lock); + eventQueue.clear(); + xmtQueue.clear(); +} + +void BrokerProxyImpl::startProtocol() +{ + Mutex::ScopedLock _lock(lock); + char rawbuffer[512]; + Buffer buffer(rawbuffer, 512); + + requestsOutstanding = 1; + topicBound = false; + Protocol::encodeHeader(buffer, Protocol::OP_BROKER_REQUEST); + sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT BrokerRequest"); +} + +void BrokerProxyImpl::sendBufferLH(Buffer& buf, const string& destination, const string& routingKey) +{ + uint32_t length = buf.getPosition(); + MessageImpl::Ptr message(new MessageImpl); + + buf.reset(); + buf.getRawData(message->body, length); + message->destination = destination; + message->routingKey = routingKey; + message->replyExchange = DIR_EXCHANGE; + message->replyKey = queueName; + + xmtQueue.push_back(message); +} + +void BrokerProxyImpl::handleRcvMessage(Message& message) +{ + Buffer inBuffer(message.body, message.length); + uint8_t opcode; + uint32_t sequence; + + while (Protocol::checkHeader(inBuffer, &opcode, &sequence)) { + if (opcode == Protocol::OP_BROKER_RESPONSE) handleBrokerResponse(inBuffer, sequence); + else if (opcode == Protocol::OP_PACKAGE_INDICATION) handlePackageIndication(inBuffer, sequence); + else if (opcode == Protocol::OP_COMMAND_COMPLETE) handleCommandComplete(inBuffer, sequence); + else if (opcode == Protocol::OP_CLASS_INDICATION) handleClassIndication(inBuffer, sequence); + else if (opcode == Protocol::OP_METHOD_RESPONSE) handleMethodResponse(inBuffer, sequence); + else if (opcode == Protocol::OP_HEARTBEAT_INDICATION) handleHeartbeatIndication(inBuffer, sequence); + else if (opcode == Protocol::OP_EVENT_INDICATION) handleEventIndication(inBuffer, sequence); + else if (opcode == Protocol::OP_SCHEMA_RESPONSE) handleSchemaResponse(inBuffer, sequence); + else if (opcode == Protocol::OP_PROPERTY_INDICATION) handleObjectIndication(inBuffer, sequence, true, false); + else if (opcode == Protocol::OP_STATISTIC_INDICATION) handleObjectIndication(inBuffer, sequence, false, true); + else if (opcode == Protocol::OP_OBJECT_INDICATION) handleObjectIndication(inBuffer, sequence, true, true); + else { + QPID_LOG(trace, "BrokerProxyImpl::handleRcvMessage invalid opcode: " << opcode); + break; + } + } +} + +bool BrokerProxyImpl::getXmtMessage(Message& item) const +{ + Mutex::ScopedLock _lock(lock); + if (xmtQueue.empty()) + return false; + item = xmtQueue.front()->copy(); + return true; +} + +void BrokerProxyImpl::popXmt() +{ + Mutex::ScopedLock _lock(lock); + if (!xmtQueue.empty()) + xmtQueue.pop_front(); +} + +bool BrokerProxyImpl::getEvent(BrokerEvent& event) const +{ + Mutex::ScopedLock _lock(lock); + if (eventQueue.empty()) + return false; + event = eventQueue.front()->copy(); + return true; +} + +void BrokerProxyImpl::popEvent() +{ + Mutex::ScopedLock _lock(lock); + if (!eventQueue.empty()) + eventQueue.pop_front(); +} + +void BrokerProxyImpl::complete() +{ + decOutstanding(); +} + +void BrokerProxyImpl::addBinding(const string& exchange, const string& key) +{ + eventQueue.push_back(eventBind(exchange, queueName, key)); +} + +BrokerEventImpl::Ptr BrokerProxyImpl::eventDeclareQueue(const string& queueName) +{ + BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::DECLARE_QUEUE)); + event->name = queueName; + return event; +} + +BrokerEventImpl::Ptr BrokerProxyImpl::eventBind(const string& exchange, const string& queue, const string& key) +{ + BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::BIND)); + event->name = queue; + event->exchange = exchange; + event->bindingKey = key; + + return event; +} + +BrokerEventImpl::Ptr BrokerProxyImpl::eventSetupComplete() +{ + BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::SETUP_COMPLETE)); + return event; +} + +BrokerEventImpl::Ptr BrokerProxyImpl::eventStable() +{ + BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::STABLE)); + return event; +} + +void BrokerProxyImpl::handleBrokerResponse(Buffer& inBuffer, uint32_t seq) +{ + // Note that this function doesn't touch requestsOutstanding. This is because + // it accounts for one request completed (the BrokerRequest) and one request + // started (the PackageRequest) which cancel each other out. + + brokerId.decode(inBuffer); + QPID_LOG(trace, "RCVD BrokerResponse seq=" << seq << " brokerId=" << brokerId); + Mutex::ScopedLock _lock(lock); + Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + uint32_t sequence(seqMgr.reserve(this)); + Protocol::encodeHeader(outBuffer, Protocol::OP_PACKAGE_REQUEST, sequence); + sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT PackageRequest seq=" << sequence); +} + +void BrokerProxyImpl::handlePackageIndication(Buffer& inBuffer, uint32_t seq) +{ + string package; + + inBuffer.getShortString(package); + QPID_LOG(trace, "RCVD PackageIndication seq=" << seq << " package=" << package); + console->learnPackage(package); + + Mutex::ScopedLock _lock(lock); + Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + uint32_t sequence(seqMgr.reserve(this)); + incOutstandingLH(); + Protocol::encodeHeader(outBuffer, Protocol::OP_CLASS_QUERY, sequence); + outBuffer.putShortString(package); + sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT ClassQuery seq=" << sequence << " package=" << package); +} + +void BrokerProxyImpl::handleCommandComplete(Buffer& inBuffer, uint32_t seq) +{ + string text; + uint32_t code = inBuffer.getLong(); + inBuffer.getShortString(text); + QPID_LOG(trace, "RCVD CommandComplete seq=" << seq << " code=" << code << " text=" << text); + seqMgr.release(seq); +} + +void BrokerProxyImpl::handleClassIndication(Buffer& inBuffer, uint32_t seq) +{ + string package; + string clsName; + SchemaHash hash; + uint8_t kind = inBuffer.getOctet(); + inBuffer.getShortString(package); + inBuffer.getShortString(clsName); + hash.decode(inBuffer); + Uuid printableHash(hash.get()); + SchemaClassKeyImpl classKey(package, clsName, hash); + + QPID_LOG(trace, "RCVD ClassIndication seq=" << seq << " kind=" << (int) kind << " key=" << classKey.str()); + + if (!console->haveClass(classKey)) { + Mutex::ScopedLock _lock(lock); + incOutstandingLH(); + Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + uint32_t sequence(seqMgr.reserve(this)); + Protocol::encodeHeader(outBuffer, Protocol::OP_SCHEMA_REQUEST, sequence); + classKey.encode(outBuffer); + sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT SchemaRequest seq=" << sequence <<" key=" << classKey.str()); + } +} + +void BrokerProxyImpl::handleMethodResponse(Buffer& /*inBuffer*/, uint32_t /*seq*/) +{ + // TODO +} + +void BrokerProxyImpl::handleHeartbeatIndication(Buffer& /*inBuffer*/, uint32_t /*seq*/) +{ + // TODO +} + +void BrokerProxyImpl::handleEventIndication(Buffer& /*inBuffer*/, uint32_t /*seq*/) +{ + // TODO +} + +void BrokerProxyImpl::handleSchemaResponse(Buffer& inBuffer, uint32_t seq) +{ + SchemaObjectClassImpl::Ptr oClassPtr; + SchemaEventClassImpl::Ptr eClassPtr; + uint8_t kind = inBuffer.getOctet(); + const SchemaClassKeyImpl* key; + if (kind == CLASS_OBJECT) { + oClassPtr.reset(new SchemaObjectClassImpl(inBuffer)); + console->learnClass(oClassPtr); + key = oClassPtr->getClassKey()->impl; + QPID_LOG(trace, "RCVD SchemaResponse seq=" << seq << " kind=object key=" << key->str()); + } else if (kind == CLASS_EVENT) { + eClassPtr.reset(new SchemaEventClassImpl(inBuffer)); + console->learnClass(eClassPtr); + key = eClassPtr->getClassKey()->impl; + QPID_LOG(trace, "RCVD SchemaResponse seq=" << seq << " kind=event key=" << key->str()); + } + else { + QPID_LOG(error, "BrokerProxyImpl::handleSchemaResponse received unknown class kind: " << (int) kind); + } + + decOutstanding(); +} + +void BrokerProxyImpl::handleObjectIndication(Buffer& /*inBuffer*/, uint32_t /*seq*/, bool /*prop*/, bool /*stat*/) +{ + // TODO +} + +void BrokerProxyImpl::incOutstandingLH() +{ + requestsOutstanding++; +} + +void BrokerProxyImpl::decOutstanding() +{ + Mutex::ScopedLock _lock(lock); + requestsOutstanding--; + if (requestsOutstanding == 0 && !topicBound) { + topicBound = true; + for (vector<pair<string, string> >::const_iterator iter = console->bindingList.begin(); + iter != console->bindingList.end(); iter++) { + string exchange(iter->first.empty() ? QMF_EXCHANGE : iter->first); + string key(iter->second); + eventQueue.push_back(eventBind(exchange, queueName, key)); + } + eventQueue.push_back(eventStable()); + } +} + +MethodResponseImpl::MethodResponseImpl(Buffer& buf) : envelope(new MethodResponse(this)) +{ + string text; + + status = buf.getLong(); + buf.getMediumString(text); + exception.reset(new Value(TYPE_LSTR)); + exception->setString(text.c_str()); + + // TODO: Parse schema-specific output arguments. + arguments.reset(new Value(TYPE_MAP)); +} + +ConsoleEngineImpl::ConsoleEngineImpl(ConsoleEngine* e, const ConsoleSettings& s) : + envelope(e), settings(s) +{ + bindingList.push_back(pair<string, string>(string(), "schema.#")); + if (settings.rcvObjects && settings.rcvEvents && settings.rcvHeartbeats && !settings.userBindings) { + bindingList.push_back(pair<string, string>(string(), "console.#")); + } else { + if (settings.rcvObjects && !settings.userBindings) + bindingList.push_back(pair<string, string>(string(), "console.obj.#")); + else + bindingList.push_back(pair<string, string>(string(), "console.obj.*.*.org.apache.qpid.broker.agent")); + if (settings.rcvEvents) + bindingList.push_back(pair<string, string>(string(), "console.event.#")); + if (settings.rcvHeartbeats) + bindingList.push_back(pair<string, string>(string(), "console.heartbeat.#")); + } +} + +ConsoleEngineImpl::~ConsoleEngineImpl() +{ + // This function intentionally left blank. +} + +bool ConsoleEngineImpl::getEvent(ConsoleEvent& event) const +{ + Mutex::ScopedLock _lock(lock); + if (eventQueue.empty()) + return false; + event = eventQueue.front()->copy(); + return true; +} + +void ConsoleEngineImpl::popEvent() +{ + Mutex::ScopedLock _lock(lock); + if (!eventQueue.empty()) + eventQueue.pop_front(); +} + +void ConsoleEngineImpl::addConnection(BrokerProxy& broker, void* /*context*/) +{ + Mutex::ScopedLock _lock(lock); + brokerList.push_back(broker.impl); +} + +void ConsoleEngineImpl::delConnection(BrokerProxy& broker) +{ + Mutex::ScopedLock _lock(lock); + for (vector<BrokerProxyImpl*>::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + if (*iter == broker.impl) { + brokerList.erase(iter); + break; + } +} + +uint32_t ConsoleEngineImpl::packageCount() const +{ + Mutex::ScopedLock _lock(lock); + return packages.size(); +} + +const string& ConsoleEngineImpl::getPackageName(uint32_t idx) const +{ + const static string empty; + + Mutex::ScopedLock _lock(lock); + if (idx >= packages.size()) + return empty; + + PackageList::const_iterator iter = packages.begin(); + for (uint32_t i = 0; i < idx; i++) iter++; + return iter->first; +} + +uint32_t ConsoleEngineImpl::classCount(const char* packageName) const +{ + Mutex::ScopedLock _lock(lock); + PackageList::const_iterator pIter = packages.find(packageName); + if (pIter == packages.end()) + return 0; + + const ObjectClassList& oList = pIter->second.first; + const EventClassList& eList = pIter->second.second; + + return oList.size() + eList.size(); +} + +const SchemaClassKey* ConsoleEngineImpl::getClass(const char* packageName, uint32_t idx) const +{ + Mutex::ScopedLock _lock(lock); + PackageList::const_iterator pIter = packages.find(packageName); + if (pIter == packages.end()) + return 0; + + const ObjectClassList& oList = pIter->second.first; + const EventClassList& eList = pIter->second.second; + uint32_t count = 0; + + for (ObjectClassList::const_iterator oIter = oList.begin(); + oIter != oList.end(); oIter++) { + if (count == idx) + return oIter->second->getClassKey(); + count++; + } + + for (EventClassList::const_iterator eIter = eList.begin(); + eIter != eList.end(); eIter++) { + if (count == idx) + return eIter->second->getClassKey(); + count++; + } + + return 0; +} + +ClassKind ConsoleEngineImpl::getClassKind(const SchemaClassKey* key) const +{ + Mutex::ScopedLock _lock(lock); + PackageList::const_iterator pIter = packages.find(key->getPackageName()); + if (pIter == packages.end()) + return CLASS_OBJECT; + + const EventClassList& eList = pIter->second.second; + if (eList.find(key->impl) != eList.end()) + return CLASS_EVENT; + return CLASS_OBJECT; +} + +const SchemaObjectClass* ConsoleEngineImpl::getObjectClass(const SchemaClassKey* key) const +{ + Mutex::ScopedLock _lock(lock); + PackageList::const_iterator pIter = packages.find(key->getPackageName()); + if (pIter == packages.end()) + return 0; + + const ObjectClassList& oList = pIter->second.first; + ObjectClassList::const_iterator iter = oList.find(key->impl); + if (iter == oList.end()) + return 0; + return iter->second->envelope; +} + +const SchemaEventClass* ConsoleEngineImpl::getEventClass(const SchemaClassKey* key) const +{ + Mutex::ScopedLock _lock(lock); + PackageList::const_iterator pIter = packages.find(key->getPackageName()); + if (pIter == packages.end()) + return 0; + + const EventClassList& eList = pIter->second.second; + EventClassList::const_iterator iter = eList.find(key->impl); + if (iter == eList.end()) + return 0; + return iter->second->envelope; +} + +void ConsoleEngineImpl::bindPackage(const char* packageName) +{ + stringstream key; + key << "console.obj.*.*." << packageName << ".#"; + Mutex::ScopedLock _lock(lock); + bindingList.push_back(pair<string, string>(string(), key.str())); + for (vector<BrokerProxyImpl*>::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + (*iter)->addBinding(QMF_EXCHANGE, key.str()); +} + +void ConsoleEngineImpl::bindClass(const SchemaClassKey* classKey) +{ + stringstream key; + key << "console.obj.*.*." << classKey->getPackageName() << "." << classKey->getClassName() << ".#"; + Mutex::ScopedLock _lock(lock); + bindingList.push_back(pair<string, string>(string(), key.str())); + for (vector<BrokerProxyImpl*>::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + (*iter)->addBinding(QMF_EXCHANGE, key.str()); +} + +void ConsoleEngineImpl::bindClass(const char* packageName, const char* className) +{ + stringstream key; + key << "console.obj.*.*." << packageName << "." << className << ".#"; + Mutex::ScopedLock _lock(lock); + bindingList.push_back(pair<string, string>(string(), key.str())); + for (vector<BrokerProxyImpl*>::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + (*iter)->addBinding(QMF_EXCHANGE, key.str()); +} + +uint32_t ConsoleEngineImpl::agentCount() const +{ + // TODO + return 0; +} + +const AgentProxy* ConsoleEngineImpl::getAgent(uint32_t /*idx*/) const +{ + // TODO + return 0; +} + +void ConsoleEngineImpl::sendQuery(const Query& /*query*/, void* /*context*/) +{ + // TODO +} + +/* +void ConsoleEngineImpl::startSync(const Query& query, void* context, SyncQuery& sync) +{ +} + +void ConsoleEngineImpl::touchSync(SyncQuery& sync) +{ +} + +void ConsoleEngineImpl::endSync(SyncQuery& sync) +{ +} +*/ + +void ConsoleEngineImpl::learnPackage(const string& packageName) +{ + Mutex::ScopedLock _lock(lock); + if (packages.find(packageName) == packages.end()) + packages.insert(pair<string, pair<ObjectClassList, EventClassList> > + (packageName, pair<ObjectClassList, EventClassList>(ObjectClassList(), EventClassList()))); +} + +void ConsoleEngineImpl::learnClass(SchemaObjectClassImpl::Ptr cls) +{ + Mutex::ScopedLock _lock(lock); + const SchemaClassKey* key = cls->getClassKey(); + PackageList::iterator pIter = packages.find(key->getPackageName()); + if (pIter == packages.end()) + return; + + ObjectClassList& list = pIter->second.first; + if (list.find(key->impl) == list.end()) + list[key->impl] = cls; +} + +void ConsoleEngineImpl::learnClass(SchemaEventClassImpl::Ptr cls) +{ + Mutex::ScopedLock _lock(lock); + const SchemaClassKey* key = cls->getClassKey(); + PackageList::iterator pIter = packages.find(key->getPackageName()); + if (pIter == packages.end()) + return; + + EventClassList& list = pIter->second.second; + if (list.find(key->impl) == list.end()) + list[key->impl] = cls; +} + +bool ConsoleEngineImpl::haveClass(const SchemaClassKeyImpl& key) const +{ + Mutex::ScopedLock _lock(lock); + PackageList::const_iterator pIter = packages.find(key.getPackageName()); + if (pIter == packages.end()) + return false; + + const ObjectClassList& oList = pIter->second.first; + const EventClassList& eList = pIter->second.second; + + return oList.find(&key) != oList.end() || eList.find(&key) != eList.end(); +} + + +//================================================================== +// Wrappers +//================================================================== + +BrokerProxy::BrokerProxy(ConsoleEngine& console) : impl(new BrokerProxyImpl(this, console)) {} +BrokerProxy::~BrokerProxy() { delete impl; } +void BrokerProxy::sessionOpened(SessionHandle& sh) { impl->sessionOpened(sh); } +void BrokerProxy::sessionClosed() { impl->sessionClosed(); } +void BrokerProxy::startProtocol() { impl->startProtocol(); } +void BrokerProxy::handleRcvMessage(Message& message) { impl->handleRcvMessage(message); } +bool BrokerProxy::getXmtMessage(Message& item) const { return impl->getXmtMessage(item); } +void BrokerProxy::popXmt() { impl->popXmt(); } +bool BrokerProxy::getEvent(BrokerEvent& event) const { return impl->getEvent(event); } +void BrokerProxy::popEvent() { impl->popEvent(); } + +AgentProxy::AgentProxy(ConsoleEngine& console) : impl(new AgentProxyImpl(this, console)) {} +AgentProxy::~AgentProxy() { delete impl; } + +MethodResponse::MethodResponse(MethodResponseImpl* i) : impl(i) {} +MethodResponse::~MethodResponse() { delete impl; } // TODO: correct to delete here? +uint32_t MethodResponse::getStatus() const { return impl->getStatus(); } +const Value* MethodResponse::getException() const { return impl->getException(); } +const Value* MethodResponse::getArgs() const { return impl->getArgs(); } + +ConsoleEngine::ConsoleEngine(const ConsoleSettings& settings) : impl(new ConsoleEngineImpl(this, settings)) {} +ConsoleEngine::~ConsoleEngine() { delete impl; } +bool ConsoleEngine::getEvent(ConsoleEvent& event) const { return impl->getEvent(event); } +void ConsoleEngine::popEvent() { impl->popEvent(); } +void ConsoleEngine::addConnection(BrokerProxy& broker, void* context) { impl->addConnection(broker, context); } +void ConsoleEngine::delConnection(BrokerProxy& broker) { impl->delConnection(broker); } +uint32_t ConsoleEngine::packageCount() const { return impl->packageCount(); } +const char* ConsoleEngine::getPackageName(uint32_t idx) const { return impl->getPackageName(idx).c_str(); } +uint32_t ConsoleEngine::classCount(const char* packageName) const { return impl->classCount(packageName); } +const SchemaClassKey* ConsoleEngine::getClass(const char* packageName, uint32_t idx) const { return impl->getClass(packageName, idx); } +ClassKind ConsoleEngine::getClassKind(const SchemaClassKey* key) const { return impl->getClassKind(key); } +const SchemaObjectClass* ConsoleEngine::getObjectClass(const SchemaClassKey* key) const { return impl->getObjectClass(key); } +const SchemaEventClass* ConsoleEngine::getEventClass(const SchemaClassKey* key) const { return impl->getEventClass(key); } +void ConsoleEngine::bindPackage(const char* packageName) { impl->bindPackage(packageName); } +void ConsoleEngine::bindClass(const SchemaClassKey* key) { impl->bindClass(key); } +void ConsoleEngine::bindClass(const char* packageName, const char* className) { impl->bindClass(packageName, className); } +uint32_t ConsoleEngine::agentCount() const { return impl->agentCount(); } +const AgentProxy* ConsoleEngine::getAgent(uint32_t idx) const { return impl->getAgent(idx); } +void ConsoleEngine::sendQuery(const Query& query, void* context) { impl->sendQuery(query, context); } +//void ConsoleEngine::startSync(const Query& query, void* context, SyncQuery& sync) { impl->startSync(query, context, sync); } +//void ConsoleEngine::touchSync(SyncQuery& sync) { impl->touchSync(sync); } +//void ConsoleEngine::endSync(SyncQuery& sync) { impl->endSync(sync); } + + diff --git a/qpid/cpp/src/qmf/ConsoleEngine.h b/qpid/cpp/src/qmf/ConsoleEngine.h new file mode 100644 index 0000000000..84ac78cd69 --- /dev/null +++ b/qpid/cpp/src/qmf/ConsoleEngine.h @@ -0,0 +1,200 @@ +#ifndef _QmfConsoleEngine_ +#define _QmfConsoleEngine_ + +/* + * 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 <qmf/ResilientConnection.h> +#include <qmf/Schema.h> +#include <qmf/ObjectId.h> +#include <qmf/Object.h> +#include <qmf/Event.h> +#include <qmf/Query.h> +#include <qmf/Value.h> +#include <qmf/Message.h> + +namespace qmf { + + class ConsoleEngine; + class ConsoleEngineImpl; + class BrokerProxyImpl; + class AgentProxy; + class AgentProxyImpl; + class MethodResponseImpl; + + /** + * + */ + class MethodResponse { + public: + MethodResponse(MethodResponseImpl* impl); + ~MethodResponse(); + uint32_t getStatus() const; + const Value* getException() const; + const Value* getArgs() const; + + private: + friend class ConsoleEngineImpl; + MethodResponseImpl* impl; + }; + + /** + * + */ + struct ConsoleEvent { + enum EventKind { + AGENT_ADDED = 1, + AGENT_DELETED = 2, + NEW_PACKAGE = 3, + NEW_CLASS = 4, + OBJECT_UPDATE = 5, + QUERY_COMPLETE = 6, + EVENT_RECEIVED = 7, + AGENT_HEARTBEAT = 8, + METHOD_RESPONSE = 9 + }; + + EventKind kind; + AgentProxy* agent; // (AGENT_[ADDED|DELETED|HEARTBEAT]) + char* name; // (NEW_PACKAGE) + SchemaClassKey* classKey; // (NEW_CLASS) + Object* object; // (OBJECT_UPDATE) + void* context; // (OBJECT_UPDATE, QUERY_COMPLETE) + Event* event; // (EVENT_RECEIVED) + uint64_t timestamp; // (AGENT_HEARTBEAT) + uint32_t methodHandle; // (METHOD_RESPONSE) + MethodResponse* methodResponse; // (METHOD_RESPONSE) + }; + + /** + * + */ + struct BrokerEvent { + enum EventKind { + BROKER_INFO = 10, + DECLARE_QUEUE = 11, + DELETE_QUEUE = 12, + BIND = 13, + UNBIND = 14, + SETUP_COMPLETE = 15, + STABLE = 16 + }; + + EventKind kind; + char* name; // ([DECLARE|DELETE]_QUEUE, [UN]BIND) + char* exchange; // ([UN]BIND) + char* bindingKey; // ([UN]BIND) + }; + + /** + * + */ + class BrokerProxy { + public: + BrokerProxy(ConsoleEngine& console); + ~BrokerProxy(); + + void sessionOpened(SessionHandle& sh); + void sessionClosed(); + void startProtocol(); + + void handleRcvMessage(Message& message); + bool getXmtMessage(Message& item) const; + void popXmt(); + + bool getEvent(BrokerEvent& event) const; + void popEvent(); + + private: + friend class ConsoleEngineImpl; + BrokerProxyImpl* impl; + }; + + /** + * + */ + class AgentProxy { + public: + AgentProxy(ConsoleEngine& console); + ~AgentProxy(); + + private: + friend class ConsoleEngineImpl; + AgentProxyImpl* impl; + }; + + // TODO - move this to a public header + struct ConsoleSettings { + bool rcvObjects; + bool rcvEvents; + bool rcvHeartbeats; + bool userBindings; + + ConsoleSettings() : + rcvObjects(true), + rcvEvents(true), + rcvHeartbeats(true), + userBindings(false) {} + }; + + class ConsoleEngine { + public: + ConsoleEngine(const ConsoleSettings& settings = ConsoleSettings()); + ~ConsoleEngine(); + + bool getEvent(ConsoleEvent& event) const; + void popEvent(); + + void addConnection(BrokerProxy& broker, void* context); + void delConnection(BrokerProxy& broker); + + uint32_t packageCount() const; + const char* getPackageName(uint32_t idx) const; + + uint32_t classCount(const char* packageName) const; + const SchemaClassKey* getClass(const char* packageName, uint32_t idx) const; + + ClassKind getClassKind(const SchemaClassKey* key) const; + const SchemaObjectClass* getObjectClass(const SchemaClassKey* key) const; + const SchemaEventClass* getEventClass(const SchemaClassKey* key) const; + + void bindPackage(const char* packageName); + void bindClass(const SchemaClassKey* key); + void bindClass(const char* packageName, const char* className); + + uint32_t agentCount() const; + const AgentProxy* getAgent(uint32_t idx) const; + + void sendQuery(const Query& query, void* context); + + /* + void startSync(const Query& query, void* context, SyncQuery& sync); + void touchSync(SyncQuery& sync); + void endSync(SyncQuery& sync); + */ + + private: + friend class BrokerProxyImpl; + friend class AgentProxyImpl; + ConsoleEngineImpl* impl; + }; +} + +#endif + diff --git a/qpid/cpp/src/qmf/Object.h b/qpid/cpp/src/qmf/Object.h index 8caab8d6dc..eb92cbbe45 100644 --- a/qpid/cpp/src/qmf/Object.h +++ b/qpid/cpp/src/qmf/Object.h @@ -31,7 +31,7 @@ namespace qmf { public: Object(const SchemaObjectClass* type); Object(ObjectImpl* impl); - ~Object(); + virtual ~Object(); void destroy(); const ObjectId* getObjectId() const; diff --git a/qpid/cpp/src/qmf/ObjectId.h b/qpid/cpp/src/qmf/ObjectId.h index 1ceae20bd8..ffd1b6978b 100644 --- a/qpid/cpp/src/qmf/ObjectId.h +++ b/qpid/cpp/src/qmf/ObjectId.h @@ -39,7 +39,6 @@ namespace qmf { bool isDurable() const; bool operator==(const ObjectId& other) const; - bool operator!=(const ObjectId& other) const; bool operator<(const ObjectId& other) const; bool operator>(const ObjectId& other) const; bool operator<=(const ObjectId& other) const; diff --git a/qpid/cpp/src/qmf/ObjectIdImpl.cpp b/qpid/cpp/src/qmf/ObjectIdImpl.cpp index efa8e7119b..75661fdb47 100644 --- a/qpid/cpp/src/qmf/ObjectIdImpl.cpp +++ b/qpid/cpp/src/qmf/ObjectIdImpl.cpp @@ -166,11 +166,6 @@ bool ObjectId::operator==(const ObjectId& other) const return *impl == *other.impl; } -bool ObjectId::operator!=(const ObjectId& other) const -{ - return !(*impl == *other.impl); -} - bool ObjectId::operator<(const ObjectId& other) const { return *impl < *other.impl; diff --git a/qpid/cpp/src/qmf/ObjectImpl.cpp b/qpid/cpp/src/qmf/ObjectImpl.cpp index d3882935e4..645ccd5c81 100644 --- a/qpid/cpp/src/qmf/ObjectImpl.cpp +++ b/qpid/cpp/src/qmf/ObjectImpl.cpp @@ -123,9 +123,9 @@ void ObjectImpl::parsePresenceMasks(Buffer& buffer, set<string>& excludeList) void ObjectImpl::encodeSchemaKey(qpid::framing::Buffer& buffer) const { - buffer.putShortString(objectClass->getPackage()); - buffer.putShortString(objectClass->getName()); - buffer.putBin128(const_cast<uint8_t*>(objectClass->getHash())); + buffer.putShortString(objectClass->getClassKey()->getPackageName()); + buffer.putShortString(objectClass->getClassKey()->getClassName()); + buffer.putBin128(const_cast<uint8_t*>(objectClass->getClassKey()->getHash())); } void ObjectImpl::encodeManagedObjectData(qpid::framing::Buffer& buffer) const diff --git a/qpid/cpp/src/qmf/Protocol.cpp b/qpid/cpp/src/qmf/Protocol.cpp new file mode 100644 index 0000000000..0a3beeb276 --- /dev/null +++ b/qpid/cpp/src/qmf/Protocol.cpp @@ -0,0 +1,52 @@ +/* + * 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 "qmf/Protocol.h" +#include "qpid/framing/Buffer.h" + +using namespace std; +using namespace qmf; +using namespace qpid::framing; + + +bool Protocol::checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq) +{ + if (buf.available() < 8) + return false; + + uint8_t h1 = buf.getOctet(); + uint8_t h2 = buf.getOctet(); + uint8_t h3 = buf.getOctet(); + + *opcode = buf.getOctet(); + *seq = buf.getLong(); + + return h1 == 'A' && h2 == 'M' && h3 == '3'; +} + +void Protocol::encodeHeader(qpid::framing::Buffer& buf, uint8_t opcode, uint32_t seq) +{ + buf.putOctet('A'); + buf.putOctet('M'); + buf.putOctet('3'); + buf.putOctet(opcode); + buf.putLong (seq); +} + + diff --git a/qpid/cpp/src/qmf/Protocol.h b/qpid/cpp/src/qmf/Protocol.h new file mode 100644 index 0000000000..d5da08c1db --- /dev/null +++ b/qpid/cpp/src/qmf/Protocol.h @@ -0,0 +1,67 @@ +#ifndef _QmfProtocol_ +#define _QmfProtocol_ + +/* + * 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 <qpid/sys/IntegerTypes.h> + +namespace qpid { + namespace framing { + class Buffer; + } +} + +namespace qmf { + + class Protocol { + public: + static bool checkHeader(qpid::framing::Buffer& buf, uint8_t *opcode, uint32_t *seq); + static void encodeHeader(qpid::framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0); + + const static uint8_t OP_ATTACH_REQUEST = 'A'; + const static uint8_t OP_ATTACH_RESPONSE = 'a'; + + const static uint8_t OP_BROKER_REQUEST = 'B'; + const static uint8_t OP_BROKER_RESPONSE = 'b'; + + const static uint8_t OP_CONSOLE_ADDED_INDICATION = 'x'; + const static uint8_t OP_COMMAND_COMPLETE = 'z'; + const static uint8_t OP_HEARTBEAT_INDICATION = 'h'; + + const static uint8_t OP_PACKAGE_REQUEST = 'P'; + const static uint8_t OP_PACKAGE_INDICATION = 'p'; + const static uint8_t OP_CLASS_QUERY = 'Q'; + const static uint8_t OP_CLASS_INDICATION = 'q'; + const static uint8_t OP_SCHEMA_REQUEST = 'S'; + const static uint8_t OP_SCHEMA_RESPONSE = 's'; + + const static uint8_t OP_METHOD_REQUEST = 'M'; + const static uint8_t OP_METHOD_RESPONSE = 'm'; + const static uint8_t OP_GET_QUERY = 'G'; + const static uint8_t OP_OBJECT_INDICATION = 'g'; + const static uint8_t OP_PROPERTY_INDICATION = 'c'; + const static uint8_t OP_STATISTIC_INDICATION = 'i'; + const static uint8_t OP_EVENT_INDICATION = 'e'; + }; + +} + +#endif + diff --git a/qpid/cpp/src/qmf/ResilientConnection.cpp b/qpid/cpp/src/qmf/ResilientConnection.cpp index 610306f896..7ec03cf4da 100644 --- a/qpid/cpp/src/qmf/ResilientConnection.cpp +++ b/qpid/cpp/src/qmf/ResilientConnection.cpp @@ -19,6 +19,8 @@ #include "qmf/ResilientConnection.h" #include "qmf/MessageImpl.h" +#include "qmf/ConnectionSettingsImpl.h" +#include <qpid/client/Connection.h> #include <qpid/client/Session.h> #include <qpid/client/MessageListener.h> #include <qpid/client/SubscriptionManager.h> @@ -39,7 +41,7 @@ using namespace std; using namespace qmf; -using namespace qpid::client; +using namespace qpid; using qpid::sys::Mutex; namespace qmf { @@ -55,30 +57,29 @@ namespace qmf { ResilientConnectionEvent copy(); }; - struct RCSession : public MessageListener, public qpid::sys::Runnable, public qpid::RefCounted { + struct RCSession : public client::MessageListener, public qpid::sys::Runnable, public qpid::RefCounted { typedef boost::intrusive_ptr<RCSession> Ptr; ResilientConnectionImpl& connImpl; string name; - Connection& connection; - Session session; - SubscriptionManager* subscriptions; + client::Connection& connection; + client::Session session; + client::SubscriptionManager* subscriptions; void* userContext; vector<string> dests; qpid::sys::Thread thread; - RCSession(ResilientConnectionImpl& ci, const string& n, Connection& c, void* uc) : + RCSession(ResilientConnectionImpl& ci, const string& n, client::Connection& c, void* uc) : connImpl(ci), name(n), connection(c), session(connection.newSession(name)), - subscriptions(new SubscriptionManager(session)), userContext(uc), thread(*this) {} + subscriptions(new client::SubscriptionManager(session)), userContext(uc), thread(*this) {} ~RCSession(); - void received(qpid::client::Message& msg); + void received(client::Message& msg); void run(); void stop(); }; class ResilientConnectionImpl : public qpid::sys::Runnable { public: - ResilientConnectionImpl(ConnectionSettings& settings, - int dmin, int dmax, int dfactor); + ResilientConnectionImpl(const ConnectionSettings& settings); ~ResilientConnectionImpl(); bool isConnected() const; @@ -107,8 +108,8 @@ namespace qmf { bool connected; bool shutdown; string lastError; - ConnectionSettings settings; - Connection connection; + const ConnectionSettings settings; + client::Connection connection; mutable qpid::sys::Mutex lock; int delayMin; int delayMax; @@ -155,7 +156,7 @@ void RCSession::stop() subscriptions->stop(); } -void RCSession::received(qpid::client::Message& msg) +void RCSession::received(client::Message& msg) { qmf::MessageImpl qmsg; qmsg.body = msg.getData(); @@ -174,12 +175,11 @@ void RCSession::received(qpid::client::Message& msg) connImpl.EnqueueEvent(ResilientConnectionEvent::RECV, userContext, qmsg); } -ResilientConnectionImpl::ResilientConnectionImpl(ConnectionSettings& _settings, - int dmin, int dmax, int dfactor) : - notifyFd(-1), connected(false), shutdown(false), settings(_settings), - delayMin(dmin), delayMax(dmax), delayFactor(dfactor), connThread(*this) +ResilientConnectionImpl::ResilientConnectionImpl(const ConnectionSettings& _settings) : + notifyFd(-1), connected(false), shutdown(false), settings(_settings), delayMin(1), connThread(*this) { connection.registerFailureCallback(boost::bind(&ResilientConnectionImpl::failure, this)); + settings.impl->getRetrySettings(&delayMin, &delayMax, &delayFactor); } ResilientConnectionImpl::~ResilientConnectionImpl() @@ -222,7 +222,7 @@ bool ResilientConnectionImpl::createSession(const char* name, void* sessionConte RCSession::Ptr sess = RCSession::Ptr(new RCSession(*this, name, connection, sessionContext)); - handle.handle = (void*) sess.get(); + handle.impl = (void*) sess.get(); sessions.insert(sess); return true; @@ -231,7 +231,7 @@ bool ResilientConnectionImpl::createSession(const char* name, void* sessionConte void ResilientConnectionImpl::destroySession(SessionHandle handle) { Mutex::ScopedLock _lock(lock); - RCSession::Ptr sess = RCSession::Ptr((RCSession*) handle.handle); + RCSession::Ptr sess = RCSession::Ptr((RCSession*) handle.impl); set<RCSession::Ptr>::iterator iter = sessions.find(sess); if (iter != sessions.end()) { for (vector<string>::iterator dIter = sess->dests.begin(); dIter != sess->dests.end(); dIter++) @@ -247,7 +247,7 @@ void ResilientConnectionImpl::destroySession(SessionHandle handle) void ResilientConnectionImpl::sendMessage(SessionHandle handle, qmf::Message& message) { Mutex::ScopedLock _lock(lock); - RCSession::Ptr sess = RCSession::Ptr((RCSession*) handle.handle); + RCSession::Ptr sess = RCSession::Ptr((RCSession*) handle.impl); set<RCSession::Ptr>::iterator iter = sessions.find(sess); qpid::client::Message msg; string data(message.body, message.length); @@ -256,7 +256,7 @@ void ResilientConnectionImpl::sendMessage(SessionHandle handle, qmf::Message& me msg.setData(data); try { - sess->session.messageTransfer(arg::content=msg, arg::destination=message.destination); + sess->session.messageTransfer(client::arg::content=msg, client::arg::destination=message.destination); } catch(exception& e) { QPID_LOG(error, "Session Exception during message-transfer: " << e.what()); sessions.erase(iter); @@ -267,19 +267,22 @@ void ResilientConnectionImpl::sendMessage(SessionHandle handle, qmf::Message& me void ResilientConnectionImpl::declareQueue(SessionHandle handle, char* queue) { Mutex::ScopedLock _lock(lock); - RCSession* sess = (RCSession*) handle.handle; + RCSession* sess = (RCSession*) handle.impl; - sess->session.queueDeclare(arg::queue=queue, arg::autoDelete=true, arg::exclusive=true); + sess->session.queueDeclare(client::arg::queue=queue, client::arg::autoDelete=true, client::arg::exclusive=true); + sess->subscriptions->setAcceptMode(client::ACCEPT_MODE_NONE); + sess->subscriptions->setAcquireMode(client::ACQUIRE_MODE_PRE_ACQUIRED); sess->subscriptions->subscribe(*sess, queue, queue); + sess->subscriptions->setFlowControl(queue, client::FlowControl::unlimited()); sess->dests.push_back(string(queue)); } void ResilientConnectionImpl::deleteQueue(SessionHandle handle, char* queue) { Mutex::ScopedLock _lock(lock); - RCSession* sess = (RCSession*) handle.handle; + RCSession* sess = (RCSession*) handle.impl; - sess->session.queueDelete(arg::queue=queue); + sess->session.queueDelete(client::arg::queue=queue); for (vector<string>::iterator iter = sess->dests.begin(); iter != sess->dests.end(); iter++) if (*iter == queue) { @@ -293,18 +296,18 @@ void ResilientConnectionImpl::bind(SessionHandle handle, char* exchange, char* queue, char* key) { Mutex::ScopedLock _lock(lock); - RCSession* sess = (RCSession*) handle.handle; + RCSession* sess = (RCSession*) handle.impl; - sess->session.exchangeBind(arg::exchange=exchange, arg::queue=queue, arg::bindingKey=key); + sess->session.exchangeBind(client::arg::exchange=exchange, client::arg::queue=queue, client::arg::bindingKey=key); } void ResilientConnectionImpl::unbind(SessionHandle handle, char* exchange, char* queue, char* key) { Mutex::ScopedLock _lock(lock); - RCSession* sess = (RCSession*) handle.handle; + RCSession* sess = (RCSession*) handle.impl; - sess->session.exchangeUnbind(arg::exchange=exchange, arg::queue=queue, arg::bindingKey=key); + sess->session.exchangeUnbind(client::arg::exchange=exchange, client::arg::queue=queue, client::arg::bindingKey=key); } void ResilientConnectionImpl::setNotifyFd(int fd) @@ -318,7 +321,8 @@ void ResilientConnectionImpl::run() while (true) { try { - connection.open(settings); + QPID_LOG(trace, "Trying to open connection..."); + connection.open(settings.impl->getClientSettings()); { Mutex::ScopedLock _lock(lock); connected = true; @@ -326,6 +330,7 @@ void ResilientConnectionImpl::run() while (connected) cond.wait(lock); + delay = delayMin; while (!sessions.empty()) { set<RCSession::Ptr>::iterator iter = sessions.begin(); @@ -334,6 +339,11 @@ void ResilientConnectionImpl::run() EnqueueEvent(ResilientConnectionEvent::SESSION_CLOSED, sess->userContext); Mutex::ScopedUnlock _u(lock); sess->stop(); + + // Nullify the intrusive pointer within the scoped unlock, otherwise, + // the reference is held until overwritted above (under lock) which causes + // the session destructor to be called with the lock held. + sess = 0; } EnqueueEvent(ResilientConnectionEvent::DISCONNECTED); @@ -341,7 +351,6 @@ void ResilientConnectionImpl::run() if (shutdown) return; } - delay = delayMin; connection.close(); } catch (exception &e) { QPID_LOG(debug, "connection.open exception: " << e.what()); @@ -396,10 +405,9 @@ void ResilientConnectionImpl::EnqueueEvent(ResilientConnectionEvent::EventKind k // Wrappers //================================================================== -ResilientConnection::ResilientConnection(ConnectionSettings& settings, - int delayMin, int delayMax, int delayFactor) +ResilientConnection::ResilientConnection(const ConnectionSettings& settings) { - impl = new ResilientConnectionImpl(settings, delayMin, delayMax, delayFactor); + impl = new ResilientConnectionImpl(settings); } ResilientConnection::~ResilientConnection() diff --git a/qpid/cpp/src/qmf/ResilientConnection.h b/qpid/cpp/src/qmf/ResilientConnection.h index bb565e27ae..03f1b9c0d5 100644 --- a/qpid/cpp/src/qmf/ResilientConnection.h +++ b/qpid/cpp/src/qmf/ResilientConnection.h @@ -21,12 +21,13 @@ */ #include <qmf/Message.h> -#include <qpid/client/Connection.h> -#include <qpid/client/ConnectionSettings.h> +#include <qmf/ConnectionSettings.h> #include <string> namespace qmf { + class ResilientConnectionImpl; + /** * Represents events that occur, unsolicited, from ResilientConnection. */ @@ -44,12 +45,11 @@ namespace qmf { Message message; // RECV }; - struct SessionHandle { - void* handle; + class SessionHandle { + friend class ResilientConnectionImpl; + void* impl; }; - class ResilientConnectionImpl; - /** * ResilientConnection represents a Qpid connection that is resilient. * @@ -68,10 +68,7 @@ namespace qmf { *@param delayMax Maximum delay (in seconds) between retries. *@param delayFactor Factor to multiply retry delay by after each failure. */ - ResilientConnection(qpid::client::ConnectionSettings& settings, - int delayMin = 1, - int delayMax = 128, - int delayFactor = 2); + ResilientConnection(const ConnectionSettings& settings); ~ResilientConnection(); /** diff --git a/qpid/cpp/src/qmf/Schema.h b/qpid/cpp/src/qmf/Schema.h index e3ab90e3e3..1123acc3b8 100644 --- a/qpid/cpp/src/qmf/Schema.h +++ b/qpid/cpp/src/qmf/Schema.h @@ -35,6 +35,7 @@ namespace qmf { struct SchemaStatisticImpl; struct SchemaObjectClassImpl; struct SchemaEventClassImpl; + struct SchemaClassKeyImpl; /** */ @@ -114,6 +115,20 @@ namespace qmf { /** */ + class SchemaClassKey { + public: + SchemaClassKey(SchemaClassKeyImpl* impl); + ~SchemaClassKey(); + + const char* getPackageName() const; + const char* getClassName() const; + const uint8_t* getHash() const; + + SchemaClassKeyImpl* impl; + }; + + /** + */ class SchemaObjectClass { public: SchemaObjectClass(const char* package, const char* name); @@ -123,9 +138,7 @@ namespace qmf { void addStatistic(const SchemaStatistic& statistic); void addMethod(const SchemaMethod& method); - const char* getPackage() const; - const char* getName() const; - const uint8_t* getHash() const; + const SchemaClassKey* getClassKey() const; int getPropertyCount() const; int getStatisticCount() const; int getMethodCount() const; @@ -146,9 +159,7 @@ namespace qmf { void addArgument(const SchemaArgument& argument); void setDesc(const char* desc); - const char* getPackage() const; - const char* getName() const; - const uint8_t* getHash() const; + const SchemaClassKey* getClassKey() const; int getArgumentCount() const; const SchemaArgument* getArgument(int idx) const; diff --git a/qpid/cpp/src/qmf/SchemaImpl.cpp b/qpid/cpp/src/qmf/SchemaImpl.cpp index 665c94f2a1..ae7d6ca689 100644 --- a/qpid/cpp/src/qmf/SchemaImpl.cpp +++ b/qpid/cpp/src/qmf/SchemaImpl.cpp @@ -20,6 +20,8 @@ #include "qmf/SchemaImpl.h" #include <qpid/framing/Buffer.h> #include <qpid/framing/FieldTable.h> +#include <qpid/framing/Uuid.h> +#include <string.h> #include <string> #include <vector> @@ -27,6 +29,7 @@ using namespace std; using namespace qmf; using qpid::framing::Buffer; using qpid::framing::FieldTable; +using qpid::framing::Uuid; SchemaHash::SchemaHash() { @@ -34,7 +37,7 @@ SchemaHash::SchemaHash() hash[idx] = 0x5A; } -void SchemaHash::encode(Buffer& buffer) +void SchemaHash::encode(Buffer& buffer) const { buffer.putBin128(hash); } @@ -63,6 +66,21 @@ void SchemaHash::update(const char* data, uint32_t len) } } +bool SchemaHash::operator==(const SchemaHash& other) const +{ + return ::memcmp(&hash, &other.hash, 16) == 0; +} + +bool SchemaHash::operator<(const SchemaHash& other) const +{ + return ::memcmp(&hash, &other.hash, 16) < 0; +} + +bool SchemaHash::operator>(const SchemaHash& other) const +{ + return ::memcmp(&hash, &other.hash, 16) > 0; +} + SchemaArgumentImpl::SchemaArgumentImpl(Buffer& buffer) : envelope(new SchemaArgument(this)) { FieldTable map; @@ -240,15 +258,51 @@ void SchemaStatisticImpl::updateHash(SchemaHash& hash) const hash.update(description); } -SchemaObjectClassImpl::SchemaObjectClassImpl(Buffer& buffer) : envelope(new SchemaObjectClass(this)), hasHash(true) +SchemaClassKeyImpl::SchemaClassKeyImpl(const string& p, const string& n, const SchemaHash& h) : + envelope(new SchemaClassKey(this)), package(p), name(n), hash(h) {} + +void SchemaClassKeyImpl::encode(qpid::framing::Buffer& buffer) const +{ + buffer.putShortString(package); + buffer.putShortString(name); + hash.encode(buffer); +} + +bool SchemaClassKeyImpl::operator==(const SchemaClassKeyImpl& other) const +{ + return package == other.package && + name == other.name && + hash == other.hash; +} + +bool SchemaClassKeyImpl::operator<(const SchemaClassKeyImpl& other) const +{ + if (package < other.package) return true; + if (package > other.package) return false; + if (name < other.name) return true; + if (name > other.name) return false; + return hash < other.hash; +} + +string SchemaClassKeyImpl::str() const +{ + Uuid printableHash(hash.get()); + stringstream str; + str << package << ":" << name << "(" << printableHash << ")"; + return str.str(); +} + +SchemaObjectClassImpl::SchemaObjectClassImpl(Buffer& buffer) : + envelope(new SchemaObjectClass(this)), hasHash(true), classKey(package, name, hash) { buffer.getShortString(package); buffer.getShortString(name); hash.decode(buffer); - uint16_t propCount = buffer.getShort(); - uint16_t statCount = buffer.getShort(); - uint16_t methodCount = buffer.getShort(); + /*uint8_t hasParentClass =*/ buffer.getOctet(); // TODO: Parse parent-class indicator + uint16_t propCount = buffer.getShort(); + uint16_t statCount = buffer.getShort(); + uint16_t methodCount = buffer.getShort(); for (uint16_t idx = 0; idx < propCount; idx++) { SchemaPropertyImpl* property = new SchemaPropertyImpl(buffer); @@ -288,7 +342,7 @@ void SchemaObjectClassImpl::encode(Buffer& buffer) const (*iter)->encode(buffer); } -const uint8_t* SchemaObjectClassImpl::getHash() const +const SchemaClassKey* SchemaObjectClassImpl::getClassKey() const { if (!hasHash) { hasHash = true; @@ -305,7 +359,7 @@ const uint8_t* SchemaObjectClassImpl::getHash() const (*iter)->updateHash(hash); } - return hash.get(); + return classKey.envelope; } void SchemaObjectClassImpl::addProperty(const SchemaProperty& property) @@ -353,7 +407,8 @@ const SchemaMethod* SchemaObjectClassImpl::getMethod(int idx) const return 0; } -SchemaEventClassImpl::SchemaEventClassImpl(Buffer& buffer) : envelope(new SchemaEventClass(this)), hasHash(true) +SchemaEventClassImpl::SchemaEventClassImpl(Buffer& buffer) : + envelope(new SchemaEventClass(this)), hasHash(true), classKey(package, name, hash) { buffer.getShortString(package); buffer.getShortString(name); @@ -380,7 +435,7 @@ void SchemaEventClassImpl::encode(Buffer& buffer) const (*iter)->encode(buffer); } -const uint8_t* SchemaEventClassImpl::getHash() const +const SchemaClassKey* SchemaEventClassImpl::getClassKey() const { if (!hasHash) { hasHash = true; @@ -390,7 +445,7 @@ const uint8_t* SchemaEventClassImpl::getHash() const iter != arguments.end(); iter++) (*iter)->updateHash(hash); } - return hash.get(); + return classKey.envelope; } void SchemaEventClassImpl::addArgument(const SchemaArgument& argument) @@ -408,334 +463,79 @@ const SchemaArgument* SchemaEventClassImpl::getArgument(int idx) const return 0; } + //================================================================== // Wrappers //================================================================== -SchemaArgument::SchemaArgument(const char* name, Typecode typecode) -{ - impl = new SchemaArgumentImpl(this, name, typecode); -} - +SchemaArgument::SchemaArgument(const char* name, Typecode typecode) { impl = new SchemaArgumentImpl(this, name, typecode); } SchemaArgument::SchemaArgument(SchemaArgumentImpl* i) : impl(i) {} - -SchemaArgument::~SchemaArgument() -{ - delete impl; -} - -void SchemaArgument::setDirection(Direction dir) -{ - impl->setDirection(dir); -} - -void SchemaArgument::setUnit(const char* val) -{ - impl->setUnit(val); -} - -void SchemaArgument::setDesc(const char* desc) -{ - impl->setDesc(desc); -} - -const char* SchemaArgument::getName() const -{ - return impl->getName().c_str(); -} - -Typecode SchemaArgument::getType() const -{ - return impl->getType(); -} - -Direction SchemaArgument::getDirection() const -{ - return impl->getDirection(); -} - -const char* SchemaArgument::getUnit() const -{ - return impl->getUnit().c_str(); -} - -const char* SchemaArgument::getDesc() const -{ - return impl->getDesc().c_str(); -} - -SchemaMethod::SchemaMethod(const char* name) -{ - impl = new SchemaMethodImpl(this, name); -} - +SchemaArgument::~SchemaArgument() { delete impl; } +void SchemaArgument::setDirection(Direction dir) { impl->setDirection(dir); } +void SchemaArgument::setUnit(const char* val) { impl->setUnit(val); } +void SchemaArgument::setDesc(const char* desc) { impl->setDesc(desc); } +const char* SchemaArgument::getName() const { return impl->getName().c_str(); } +Typecode SchemaArgument::getType() const { return impl->getType(); } +Direction SchemaArgument::getDirection() const { return impl->getDirection(); } +const char* SchemaArgument::getUnit() const { return impl->getUnit().c_str(); } +const char* SchemaArgument::getDesc() const { return impl->getDesc().c_str(); } +SchemaMethod::SchemaMethod(const char* name) { impl = new SchemaMethodImpl(this, name); } SchemaMethod::SchemaMethod(SchemaMethodImpl* i) : impl(i) {} - -SchemaMethod::~SchemaMethod() -{ - delete impl; -} - -void SchemaMethod::addArgument(const SchemaArgument& argument) -{ - impl->addArgument(argument); -} - -void SchemaMethod::setDesc(const char* desc) -{ - impl->setDesc(desc); -} - -const char* SchemaMethod::getName() const -{ - return impl->getName().c_str(); -} - -const char* SchemaMethod::getDesc() const -{ - return impl->getDesc().c_str(); -} - -int SchemaMethod::getArgumentCount() const -{ - return impl->getArgumentCount(); -} - -const SchemaArgument* SchemaMethod::getArgument(int idx) const -{ - return impl->getArgument(idx); -} - -SchemaProperty::SchemaProperty(const char* name, Typecode typecode) -{ - impl = new SchemaPropertyImpl(this, name, typecode); -} - +SchemaMethod::~SchemaMethod() { delete impl; } +void SchemaMethod::addArgument(const SchemaArgument& argument) { impl->addArgument(argument); } +void SchemaMethod::setDesc(const char* desc) { impl->setDesc(desc); } +const char* SchemaMethod::getName() const { return impl->getName().c_str(); } +const char* SchemaMethod::getDesc() const { return impl->getDesc().c_str(); } +int SchemaMethod::getArgumentCount() const { return impl->getArgumentCount(); } +const SchemaArgument* SchemaMethod::getArgument(int idx) const { return impl->getArgument(idx); } +SchemaProperty::SchemaProperty(const char* name, Typecode typecode) { impl = new SchemaPropertyImpl(this, name, typecode); } SchemaProperty::SchemaProperty(SchemaPropertyImpl* i) : impl(i) {} - -SchemaProperty::~SchemaProperty() -{ - delete impl; -} - -void SchemaProperty::setAccess(Access access) -{ - impl->setAccess(access); -} - -void SchemaProperty::setIndex(bool val) -{ - impl->setIndex(val); -} - -void SchemaProperty::setOptional(bool val) -{ - impl->setOptional(val); -} - -void SchemaProperty::setUnit(const char* val) -{ - impl->setUnit(val); -} - -void SchemaProperty::setDesc(const char* desc) -{ - impl->setDesc(desc); -} - -const char* SchemaProperty::getName() const -{ - return impl->getName().c_str(); -} - -Typecode SchemaProperty::getType() const -{ - return impl->getType(); -} - -Access SchemaProperty::getAccess() const -{ - return impl->getAccess(); -} - -bool SchemaProperty::isIndex() const -{ - return impl->isIndex(); -} - -bool SchemaProperty::isOptional() const -{ - return impl->isOptional(); -} - -const char* SchemaProperty::getUnit() const -{ - return impl->getUnit().c_str(); -} - -const char* SchemaProperty::getDesc() const -{ - return impl->getDesc().c_str(); -} - -SchemaStatistic::SchemaStatistic(const char* name, Typecode typecode) -{ - impl = new SchemaStatisticImpl(this, name, typecode); -} - +SchemaProperty::~SchemaProperty() { delete impl; } +void SchemaProperty::setAccess(Access access) { impl->setAccess(access); } +void SchemaProperty::setIndex(bool val) { impl->setIndex(val); } +void SchemaProperty::setOptional(bool val) { impl->setOptional(val); } +void SchemaProperty::setUnit(const char* val) { impl->setUnit(val); } +void SchemaProperty::setDesc(const char* desc) { impl->setDesc(desc); } +const char* SchemaProperty::getName() const { return impl->getName().c_str(); } +Typecode SchemaProperty::getType() const { return impl->getType(); } +Access SchemaProperty::getAccess() const { return impl->getAccess(); } +bool SchemaProperty::isIndex() const { return impl->isIndex(); } +bool SchemaProperty::isOptional() const { return impl->isOptional(); } +const char* SchemaProperty::getUnit() const { return impl->getUnit().c_str(); } +const char* SchemaProperty::getDesc() const { return impl->getDesc().c_str(); } +SchemaStatistic::SchemaStatistic(const char* name, Typecode typecode) { impl = new SchemaStatisticImpl(this, name, typecode); } SchemaStatistic::SchemaStatistic(SchemaStatisticImpl* i) : impl(i) {} - -SchemaStatistic::~SchemaStatistic() -{ - delete impl; -} - -void SchemaStatistic::setUnit(const char* val) -{ - impl->setUnit(val); -} - -void SchemaStatistic::setDesc(const char* desc) -{ - impl->setDesc(desc); -} - -const char* SchemaStatistic::getName() const -{ - return impl->getName().c_str(); -} - -Typecode SchemaStatistic::getType() const -{ - return impl->getType(); -} - -const char* SchemaStatistic::getUnit() const -{ - return impl->getUnit().c_str(); -} - -const char* SchemaStatistic::getDesc() const -{ - return impl->getDesc().c_str(); -} - -SchemaObjectClass::SchemaObjectClass(const char* package, const char* name) -{ - impl = new SchemaObjectClassImpl(this, package, name); -} - +SchemaStatistic::~SchemaStatistic() { delete impl; } +void SchemaStatistic::setUnit(const char* val) { impl->setUnit(val); } +void SchemaStatistic::setDesc(const char* desc) { impl->setDesc(desc); } +const char* SchemaStatistic::getName() const { return impl->getName().c_str(); } +Typecode SchemaStatistic::getType() const { return impl->getType(); } +const char* SchemaStatistic::getUnit() const { return impl->getUnit().c_str(); } +const char* SchemaStatistic::getDesc() const { return impl->getDesc().c_str(); } +SchemaClassKey::SchemaClassKey(SchemaClassKeyImpl* i) : impl(i) {} +SchemaClassKey::~SchemaClassKey() { delete impl; } +const char* SchemaClassKey::getPackageName() const { return impl->getPackageName().c_str(); } +const char* SchemaClassKey::getClassName() const { return impl->getClassName().c_str(); } +const uint8_t* SchemaClassKey::getHash() const { return impl->getHash(); } +SchemaObjectClass::SchemaObjectClass(const char* package, const char* name) { impl = new SchemaObjectClassImpl(this, package, name); } SchemaObjectClass::SchemaObjectClass(SchemaObjectClassImpl* i) : impl(i) {} - -SchemaObjectClass::~SchemaObjectClass() -{ - delete impl; -} - -void SchemaObjectClass::addProperty(const SchemaProperty& property) -{ - impl->addProperty(property); -} - -void SchemaObjectClass::addStatistic(const SchemaStatistic& statistic) -{ - impl->addStatistic(statistic); -} - -void SchemaObjectClass::addMethod(const SchemaMethod& method) -{ - impl->addMethod(method); -} - -const char* SchemaObjectClass::getPackage() const -{ - return impl->getPackage().c_str(); -} - -const char* SchemaObjectClass::getName() const -{ - return impl->getName().c_str(); -} - -const uint8_t* SchemaObjectClass::getHash() const -{ - return impl->getHash(); -} - -int SchemaObjectClass::getPropertyCount() const -{ - return impl->getPropertyCount(); -} - -int SchemaObjectClass::getStatisticCount() const -{ - return impl->getStatisticCount(); -} - -int SchemaObjectClass::getMethodCount() const -{ - return impl->getMethodCount(); -} - -const SchemaProperty* SchemaObjectClass::getProperty(int idx) const -{ - return impl->getProperty(idx); -} - -const SchemaStatistic* SchemaObjectClass::getStatistic(int idx) const -{ - return impl->getStatistic(idx); -} - -const SchemaMethod* SchemaObjectClass::getMethod(int idx) const -{ - return impl->getMethod(idx); -} - -SchemaEventClass::SchemaEventClass(const char* package, const char* name) -{ - impl = new SchemaEventClassImpl(this, package, name); -} - +SchemaObjectClass::~SchemaObjectClass() { delete impl; } +void SchemaObjectClass::addProperty(const SchemaProperty& property) { impl->addProperty(property); } +void SchemaObjectClass::addStatistic(const SchemaStatistic& statistic) { impl->addStatistic(statistic); } +void SchemaObjectClass::addMethod(const SchemaMethod& method) { impl->addMethod(method); } +const SchemaClassKey* SchemaObjectClass::getClassKey() const { return impl->getClassKey(); } +int SchemaObjectClass::getPropertyCount() const { return impl->getPropertyCount(); } +int SchemaObjectClass::getStatisticCount() const { return impl->getStatisticCount(); } +int SchemaObjectClass::getMethodCount() const { return impl->getMethodCount(); } +const SchemaProperty* SchemaObjectClass::getProperty(int idx) const { return impl->getProperty(idx); } +const SchemaStatistic* SchemaObjectClass::getStatistic(int idx) const { return impl->getStatistic(idx); } +const SchemaMethod* SchemaObjectClass::getMethod(int idx) const { return impl->getMethod(idx); } +SchemaEventClass::SchemaEventClass(const char* package, const char* name) { impl = new SchemaEventClassImpl(this, package, name); } SchemaEventClass::SchemaEventClass(SchemaEventClassImpl* i) : impl(i) {} - -SchemaEventClass::~SchemaEventClass() -{ - delete impl; -} - -void SchemaEventClass::addArgument(const SchemaArgument& argument) -{ - impl->addArgument(argument); -} - -void SchemaEventClass::setDesc(const char* desc) -{ - impl->setDesc(desc); -} - -const char* SchemaEventClass::getPackage() const -{ - return impl->getPackage().c_str(); -} - -const char* SchemaEventClass::getName() const -{ - return impl->getName().c_str(); -} - -const uint8_t* SchemaEventClass::getHash() const -{ - return impl->getHash(); -} - -int SchemaEventClass::getArgumentCount() const -{ - return impl->getArgumentCount(); -} - -const SchemaArgument* SchemaEventClass::getArgument(int idx) const -{ - return impl->getArgument(idx); -} +SchemaEventClass::~SchemaEventClass() { delete impl; } +void SchemaEventClass::addArgument(const SchemaArgument& argument) { impl->addArgument(argument); } +void SchemaEventClass::setDesc(const char* desc) { impl->setDesc(desc); } +const SchemaClassKey* SchemaEventClass::getClassKey() const { return impl->getClassKey(); } +int SchemaEventClass::getArgumentCount() const { return impl->getArgumentCount(); } +const SchemaArgument* SchemaEventClass::getArgument(int idx) const { return impl->getArgument(idx); } diff --git a/qpid/cpp/src/qmf/SchemaImpl.h b/qpid/cpp/src/qmf/SchemaImpl.h index 2c30a8851f..3e9677d1fa 100644 --- a/qpid/cpp/src/qmf/SchemaImpl.h +++ b/qpid/cpp/src/qmf/SchemaImpl.h @@ -21,6 +21,7 @@ */ #include "qmf/Schema.h" +#include <boost/shared_ptr.hpp> #include <string> #include <vector> #include <qpid/framing/Buffer.h> @@ -35,7 +36,7 @@ namespace qmf { uint8_t hash[16]; public: SchemaHash(); - void encode(qpid::framing::Buffer& buffer); + void encode(qpid::framing::Buffer& buffer) const; void decode(qpid::framing::Buffer& buffer); void update(const char* data, uint32_t len); void update(uint8_t data); @@ -45,6 +46,9 @@ namespace qmf { void update(Access a) { update((uint8_t) a); } void update(bool b) { update((uint8_t) (b ? 1 : 0)); } const uint8_t* get() const { return hash; } + bool operator==(const SchemaHash& other) const; + bool operator<(const SchemaHash& other) const; + bool operator>(const SchemaHash& other) const; }; struct SchemaArgumentImpl { @@ -138,27 +142,45 @@ namespace qmf { void updateHash(SchemaHash& hash) const; }; + struct SchemaClassKeyImpl { + const SchemaClassKey* envelope; + const std::string& package; + const std::string& name; + const SchemaHash& hash; + + SchemaClassKeyImpl(const std::string& package, const std::string& name, const SchemaHash& hash); + + const std::string& getPackageName() const { return package; } + const std::string& getClassName() const { return name; } + const uint8_t* getHash() const { return hash.get(); } + + void encode(qpid::framing::Buffer& buffer) const; + bool operator==(const SchemaClassKeyImpl& other) const; + bool operator<(const SchemaClassKeyImpl& other) const; + std::string str() const; + }; + struct SchemaObjectClassImpl { + typedef boost::shared_ptr<SchemaObjectClassImpl> Ptr; SchemaObjectClass* envelope; std::string package; std::string name; mutable SchemaHash hash; mutable bool hasHash; + SchemaClassKeyImpl classKey; std::vector<SchemaPropertyImpl*> properties; std::vector<SchemaStatisticImpl*> statistics; std::vector<SchemaMethodImpl*> methods; SchemaObjectClassImpl(SchemaObjectClass* e, const char* p, const char* n) : - envelope(e), package(p), name(n), hasHash(false) {} + envelope(e), package(p), name(n), hasHash(false), classKey(package, name, hash) {} SchemaObjectClassImpl(qpid::framing::Buffer& buffer); void encode(qpid::framing::Buffer& buffer) const; void addProperty(const SchemaProperty& property); void addStatistic(const SchemaStatistic& statistic); void addMethod(const SchemaMethod& method); - const std::string& getPackage() const { return package; } - const std::string& getName() const { return name; } - const uint8_t* getHash() const; + const SchemaClassKey* getClassKey() const; int getPropertyCount() const { return properties.size(); } int getStatisticCount() const { return statistics.size(); } int getMethodCount() const { return methods.size(); } @@ -168,24 +190,24 @@ namespace qmf { }; struct SchemaEventClassImpl { + typedef boost::shared_ptr<SchemaEventClassImpl> Ptr; SchemaEventClass* envelope; std::string package; std::string name; mutable SchemaHash hash; mutable bool hasHash; + SchemaClassKeyImpl classKey; std::string description; std::vector<SchemaArgumentImpl*> arguments; SchemaEventClassImpl(SchemaEventClass* e, const char* p, const char* n) : - envelope(e), package(p), name(n), hasHash(false) {} + envelope(e), package(p), name(n), hasHash(false), classKey(package, name, hash) {} SchemaEventClassImpl(qpid::framing::Buffer& buffer); void encode(qpid::framing::Buffer& buffer) const; void addArgument(const SchemaArgument& argument); void setDesc(const char* desc) { description = desc; } - const std::string& getPackage() const { return package; } - const std::string& getName() const { return name; } - const uint8_t* getHash() const; + const SchemaClassKey* getClassKey() const; int getArgumentCount() const { return arguments.size(); } const SchemaArgument* getArgument(int idx) const; }; diff --git a/qpid/cpp/src/qmf/SequenceManager.cpp b/qpid/cpp/src/qmf/SequenceManager.cpp new file mode 100644 index 0000000000..f51ce9d8b8 --- /dev/null +++ b/qpid/cpp/src/qmf/SequenceManager.cpp @@ -0,0 +1,50 @@ +/* + * 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 "qmf/SequenceManager.h" + +using namespace std; +using namespace qmf; +using namespace qpid::sys; + +SequenceManager::SequenceManager() : nextSequence(1) {} + +uint32_t SequenceManager::reserve(SequenceContext* ctx) +{ + Mutex::ScopedLock _lock(lock); + uint32_t seq = nextSequence; + while (contextMap.find(seq) != contextMap.end()) + seq = seq < 0xFFFFFFFF ? seq + 1 : 1; + nextSequence = seq < 0xFFFFFFFF ? seq + 1 : 1; + contextMap[seq] = ctx; + return seq; +} + +void SequenceManager::release(uint32_t sequence) +{ + Mutex::ScopedLock _lock(lock); + map<uint32_t, SequenceContext*>::iterator iter = contextMap.find(sequence); + if (iter != contextMap.end()) { + if (iter->second != 0) + iter->second->complete(); + contextMap.erase(iter); + } +} + + diff --git a/qpid/cpp/src/qmf/SequenceManager.h b/qpid/cpp/src/qmf/SequenceManager.h new file mode 100644 index 0000000000..c027872313 --- /dev/null +++ b/qpid/cpp/src/qmf/SequenceManager.h @@ -0,0 +1,52 @@ +#ifndef _QmfSequenceManager_ +#define _QmfSequenceManager_ + +/* + * 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 "qpid/sys/Mutex.h" +#include <map> + +namespace qmf { + + class SequenceContext { + public: + SequenceContext() {} + virtual ~SequenceContext() {} + + virtual void complete() = 0; + }; + + class SequenceManager { + public: + SequenceManager(); + + uint32_t reserve(SequenceContext* ctx); + void release(uint32_t sequence); + + private: + mutable qpid::sys::Mutex lock; + uint32_t nextSequence; + std::map<uint32_t, SequenceContext*> contextMap; + }; + +} + +#endif + diff --git a/qpid/cpp/src/qmf/Value.h b/qpid/cpp/src/qmf/Value.h index a45df14ea9..bb946d31d3 100644 --- a/qpid/cpp/src/qmf/Value.h +++ b/qpid/cpp/src/qmf/Value.h @@ -30,7 +30,7 @@ namespace qmf { class Value { public: - Value(); + // Value(); Value(Typecode t, Typecode arrayType = TYPE_UINT8); Value(ValueImpl* impl); ~Value(); |