From 4e9025e343b611268d8164d255282b2198332597 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Fri, 25 Sep 2009 20:24:22 +0000 Subject: QMF updates: - Refactored into two namespaces: 1) 'qmf' for the public QMF api (c++) 2) 'qmf::engine' for the public engine API (used for language bindings) - Added object and first_object calls to Console (in Ruby) - Made objects call compatible with the kwarg arguments used in the older API - Added to_s functions to classes that needed them git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@818994 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/Agent.cpp | 857 +++++++++++++++++++++ qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp | 704 +++++++++++++++++ qpid/cpp/src/qmf/engine/BrokerProxyImpl.h | 228 ++++++ qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp | 270 +++++++ qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h | 63 ++ qpid/cpp/src/qmf/engine/ConsoleImpl.cpp | 360 +++++++++ qpid/cpp/src/qmf/engine/ConsoleImpl.h | 135 ++++ qpid/cpp/src/qmf/engine/MessageImpl.cpp | 43 ++ qpid/cpp/src/qmf/engine/MessageImpl.h | 44 ++ qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp | 162 ++++ qpid/cpp/src/qmf/engine/ObjectIdImpl.h | 71 ++ qpid/cpp/src/qmf/engine/ObjectImpl.cpp | 232 ++++++ qpid/cpp/src/qmf/engine/ObjectImpl.h | 76 ++ qpid/cpp/src/qmf/engine/Protocol.cpp | 52 ++ qpid/cpp/src/qmf/engine/Protocol.h | 69 ++ qpid/cpp/src/qmf/engine/QueryImpl.cpp | 103 +++ qpid/cpp/src/qmf/engine/QueryImpl.h | 100 +++ qpid/cpp/src/qmf/engine/ResilientConnection.cpp | 484 ++++++++++++ qpid/cpp/src/qmf/engine/SchemaImpl.cpp | 609 +++++++++++++++ qpid/cpp/src/qmf/engine/SchemaImpl.h | 222 ++++++ qpid/cpp/src/qmf/engine/SequenceManager.cpp | 96 +++ qpid/cpp/src/qmf/engine/SequenceManager.h | 68 ++ qpid/cpp/src/qmf/engine/ValueImpl.cpp | 266 +++++++ qpid/cpp/src/qmf/engine/ValueImpl.h | 150 ++++ 24 files changed, 5464 insertions(+) create mode 100644 qpid/cpp/src/qmf/engine/Agent.cpp create mode 100644 qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/BrokerProxyImpl.h create mode 100644 qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h create mode 100644 qpid/cpp/src/qmf/engine/ConsoleImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/ConsoleImpl.h create mode 100644 qpid/cpp/src/qmf/engine/MessageImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/MessageImpl.h create mode 100644 qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/ObjectIdImpl.h create mode 100644 qpid/cpp/src/qmf/engine/ObjectImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/ObjectImpl.h create mode 100644 qpid/cpp/src/qmf/engine/Protocol.cpp create mode 100644 qpid/cpp/src/qmf/engine/Protocol.h create mode 100644 qpid/cpp/src/qmf/engine/QueryImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/QueryImpl.h create mode 100644 qpid/cpp/src/qmf/engine/ResilientConnection.cpp create mode 100644 qpid/cpp/src/qmf/engine/SchemaImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/SchemaImpl.h create mode 100644 qpid/cpp/src/qmf/engine/SequenceManager.cpp create mode 100644 qpid/cpp/src/qmf/engine/SequenceManager.h create mode 100644 qpid/cpp/src/qmf/engine/ValueImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/ValueImpl.h (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/Agent.cpp b/qpid/cpp/src/qmf/engine/Agent.cpp new file mode 100644 index 0000000000..c5d1bff2e0 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/Agent.cpp @@ -0,0 +1,857 @@ +/* + * 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/engine/Agent.h" +#include "qmf/engine/MessageImpl.h" +#include "qmf/engine/SchemaImpl.h" +#include "qmf/engine/Typecode.h" +#include "qmf/engine/ObjectImpl.h" +#include "qmf/engine/ObjectIdImpl.h" +#include "qmf/engine/QueryImpl.h" +#include "qmf/engine/ValueImpl.h" +#include "qmf/engine/Protocol.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std; +using namespace qmf::engine; +using namespace qpid::framing; +using namespace qpid::sys; + +namespace qmf { +namespace engine { + + struct AgentEventImpl { + typedef boost::shared_ptr Ptr; + AgentEvent::EventKind kind; + uint32_t sequence; + string authUserId; + string authToken; + string name; + Object* object; + boost::shared_ptr objectId; + boost::shared_ptr query; + boost::shared_ptr arguments; + string exchange; + string bindingKey; + const SchemaObjectClass* objectClass; + + AgentEventImpl(AgentEvent::EventKind k) : + kind(k), sequence(0), object(0), objectClass(0) {} + ~AgentEventImpl() {} + AgentEvent copy(); + }; + + struct AgentQueryContext { + typedef boost::shared_ptr Ptr; + uint32_t sequence; + string exchange; + string key; + const SchemaMethod* schemaMethod; + AgentQueryContext() : schemaMethod(0) {} + }; + + class AgentImpl : public boost::noncopyable { + public: + AgentImpl(char* label, bool internalStore); + ~AgentImpl(); + + void setStoreDir(const char* path); + void setTransferDir(const char* path); + void handleRcvMessage(Message& message); + bool getXmtMessage(Message& item) const; + void popXmt(); + bool getEvent(AgentEvent& event) const; + void popEvent(); + void newSession(); + void startProtocol(); + void heartbeat(); + void methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments); + void queryResponse(uint32_t sequence, Object& object, bool prop, bool stat); + void queryComplete(uint32_t sequence); + void registerClass(SchemaObjectClass* cls); + void registerClass(SchemaEventClass* cls); + const ObjectId* addObject(Object& obj, uint64_t persistId); + const ObjectId* allocObjectId(uint64_t persistId); + const ObjectId* allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi); + void raiseEvent(Event& event); + + private: + mutable Mutex lock; + Mutex addLock; + string label; + string queueName; + string storeDir; + string transferDir; + bool internalStore; + uint64_t nextTransientId; + Uuid systemId; + uint32_t requestedBrokerBank; + uint32_t requestedAgentBank; + uint32_t assignedBrokerBank; + uint32_t assignedAgentBank; + AgentAttachment attachment; + uint16_t bootSequence; + uint64_t nextObjectId; + uint32_t nextContextNum; + deque eventQueue; + deque xmtQueue; + map contextMap; + + static const char* QMF_EXCHANGE; + static const char* DIR_EXCHANGE; + static const char* BROKER_KEY; + static const uint32_t MERR_UNKNOWN_METHOD = 2; + static const uint32_t MERR_UNKNOWN_PACKAGE = 8; + static const uint32_t MERR_UNKNOWN_CLASS = 9; + static const uint32_t MERR_INTERNAL_ERROR = 10; +# define MA_BUFFER_SIZE 65536 + char outputBuffer[MA_BUFFER_SIZE]; + + struct AgentClassKey { + string name; + uint8_t hash[16]; + AgentClassKey(const string& n, const uint8_t* h) : name(n) { + memcpy(hash, h, 16); + } + AgentClassKey(Buffer& buffer) { + buffer.getShortString(name); + buffer.getBin128(hash); + } + string repr() { + return name; + } + }; + + struct AgentClassKeyComp { + bool operator() (const AgentClassKey& lhs, const AgentClassKey& rhs) const + { + if (lhs.name != rhs.name) + return lhs.name < rhs.name; + else + for (int i = 0; i < 16; i++) + if (lhs.hash[i] != rhs.hash[i]) + return lhs.hash[i] < rhs.hash[i]; + return false; + } + }; + + typedef map ObjectClassMap; + typedef map EventClassMap; + + struct ClassMaps { + ObjectClassMap objectClasses; + EventClassMap eventClasses; + }; + + map packages; + + AgentEventImpl::Ptr eventDeclareQueue(const string& queueName); + AgentEventImpl::Ptr eventBind(const string& exchange, const string& queue, const string& key); + AgentEventImpl::Ptr eventSetupComplete(); + AgentEventImpl::Ptr eventQuery(uint32_t num, const string& userId, const string& package, const string& cls, + boost::shared_ptr oid); + AgentEventImpl::Ptr eventMethod(uint32_t num, const string& userId, const string& method, + boost::shared_ptr oid, boost::shared_ptr argMap, + const SchemaObjectClass* objectClass); + void sendBufferLH(Buffer& buf, const string& destination, const string& routingKey); + + void sendPackageIndicationLH(const string& packageName); + 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=""); + void handleAttachResponse(Buffer& inBuffer); + void handlePackageRequest(Buffer& inBuffer); + void handleClassQuery(Buffer& inBuffer); + void handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, + const string& replyToExchange, const string& replyToKey); + void handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId); + void handleMethodRequest(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId); + void handleConsoleAddedIndication(); + }; +} +} + +const char* AgentImpl::QMF_EXCHANGE = "qpid.management"; +const char* AgentImpl::DIR_EXCHANGE = "amq.direct"; +const char* AgentImpl::BROKER_KEY = "broker"; + +#define STRING_REF(s) {if (!s.empty()) item.s = const_cast(s.c_str());} + +AgentEvent AgentEventImpl::copy() +{ + AgentEvent item; + + ::memset(&item, 0, sizeof(AgentEvent)); + item.kind = kind; + item.sequence = sequence; + item.object = object; + item.objectId = objectId.get(); + item.query = query.get(); + item.arguments = arguments.get(); + item.objectClass = objectClass; + + STRING_REF(authUserId); + STRING_REF(authToken); + STRING_REF(name); + STRING_REF(exchange); + STRING_REF(bindingKey); + + return item; +} + +AgentImpl::AgentImpl(char* _label, bool i) : + label(_label), queueName("qmfa-"), internalStore(i), nextTransientId(1), + requestedBrokerBank(0), requestedAgentBank(0), + assignedBrokerBank(0), assignedAgentBank(0), + bootSequence(1), nextObjectId(1), nextContextNum(1) +{ + queueName += label; +} + +AgentImpl::~AgentImpl() +{ +} + +void AgentImpl::setStoreDir(const char* path) +{ + Mutex::ScopedLock _lock(lock); + if (path) + storeDir = path; + else + storeDir.clear(); +} + +void AgentImpl::setTransferDir(const char* path) +{ + Mutex::ScopedLock _lock(lock); + if (path) + transferDir = path; + else + transferDir.clear(); +} + +void AgentImpl::handleRcvMessage(Message& message) +{ + Buffer inBuffer(message.body, message.length); + uint8_t opcode; + uint32_t sequence; + string replyToExchange(message.replyExchange ? message.replyExchange : ""); + string replyToKey(message.replyKey ? message.replyKey : ""); + string userId(message.userId ? message.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, "AgentImpl::handleRcvMessage invalid opcode=" << opcode); + break; + } + } +} + +bool AgentImpl::getXmtMessage(Message& item) const +{ + Mutex::ScopedLock _lock(lock); + if (xmtQueue.empty()) + return false; + item = xmtQueue.front()->copy(); + return true; +} + +void AgentImpl::popXmt() +{ + Mutex::ScopedLock _lock(lock); + if (!xmtQueue.empty()) + xmtQueue.pop_front(); +} + +bool AgentImpl::getEvent(AgentEvent& event) const +{ + Mutex::ScopedLock _lock(lock); + if (eventQueue.empty()) + return false; + event = eventQueue.front()->copy(); + return true; +} + +void AgentImpl::popEvent() +{ + Mutex::ScopedLock _lock(lock); + if (!eventQueue.empty()) + eventQueue.pop_front(); +} + +void AgentImpl::newSession() +{ + Mutex::ScopedLock _lock(lock); + eventQueue.clear(); + xmtQueue.clear(); + eventQueue.push_back(eventDeclareQueue(queueName)); + eventQueue.push_back(eventBind("amq.direct", queueName, queueName)); + eventQueue.push_back(eventSetupComplete()); +} + +void AgentImpl::startProtocol() +{ + Mutex::ScopedLock _lock(lock); + char rawbuffer[512]; + Buffer buffer(rawbuffer, 512); + + Protocol::encodeHeader(buffer, Protocol::OP_ATTACH_REQUEST); + buffer.putShortString("qmfa"); + systemId.encode(buffer); + buffer.putLong(requestedBrokerBank); + buffer.putLong(requestedAgentBank); + sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT AttachRequest: reqBroker=" << requestedBrokerBank << + " reqAgent=" << requestedAgentBank); +} + +void AgentImpl::heartbeat() +{ + Mutex::ScopedLock _lock(lock); + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + + Protocol::encodeHeader(buffer, Protocol::OP_HEARTBEAT_INDICATION); + buffer.putLongLong(uint64_t(Duration(now()))); + stringstream key; + key << "console.heartbeat." << assignedBrokerBank << "." << assignedAgentBank; + sendBufferLH(buffer, QMF_EXCHANGE, key.str()); + QPID_LOG(trace, "SENT HeartbeatIndication"); +} + +void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, + const Value& argMap) +{ + Mutex::ScopedLock _lock(lock); + map::iterator iter = contextMap.find(sequence); + if (iter == contextMap.end()) + return; + AgentQueryContext::Ptr context = iter->second; + contextMap.erase(iter); + + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + Protocol::encodeHeader(buffer, Protocol::OP_METHOD_RESPONSE, context->sequence); + buffer.putLong(status); + buffer.putMediumString(text); + if (status == 0) { + for (vector::const_iterator aIter = context->schemaMethod->impl->arguments.begin(); + aIter != context->schemaMethod->impl->arguments.end(); aIter++) { + const SchemaArgument* schemaArg = *aIter; + if (schemaArg->getDirection() == DIR_OUT || schemaArg->getDirection() == DIR_IN_OUT) { + if (argMap.keyInMap(schemaArg->getName())) { + const Value* val = argMap.byKey(schemaArg->getName()); + val->impl->encode(buffer); + } else { + Value val(schemaArg->getType()); + val.impl->encode(buffer); + } + } + } + } + sendBufferLH(buffer, context->exchange, context->key); + QPID_LOG(trace, "SENT MethodResponse seq=" << context->sequence << " status=" << status << " text=" << text); +} + +void AgentImpl::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) +{ + Mutex::ScopedLock _lock(lock); + map::iterator iter = contextMap.find(sequence); + if (iter == contextMap.end()) + return; + AgentQueryContext::Ptr context = iter->second; + + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + Protocol::encodeHeader(buffer, Protocol::OP_OBJECT_INDICATION, context->sequence); + + object.impl->encodeSchemaKey(buffer); + object.impl->encodeManagedObjectData(buffer); + if (prop) + object.impl->encodeProperties(buffer); + if (stat) + object.impl->encodeStatistics(buffer); + + sendBufferLH(buffer, context->exchange, context->key); + QPID_LOG(trace, "SENT ContentIndication seq=" << context->sequence); +} + +void AgentImpl::queryComplete(uint32_t sequence) +{ + Mutex::ScopedLock _lock(lock); + map::iterator iter = contextMap.find(sequence); + if (iter == contextMap.end()) + return; + + AgentQueryContext::Ptr context = iter->second; + contextMap.erase(iter); + sendCommandCompleteLH(context->exchange, context->key, context->sequence, 0, "OK"); +} + +void AgentImpl::registerClass(SchemaObjectClass* cls) +{ + Mutex::ScopedLock _lock(lock); + + map::iterator iter = packages.find(cls->getClassKey()->getPackageName()); + if (iter == packages.end()) { + packages[cls->getClassKey()->getPackageName()] = ClassMaps(); + iter = packages.find(cls->getClassKey()->getPackageName()); + // TODO: Indicate this package if connected + } + + AgentClassKey key(cls->getClassKey()->getClassName(), cls->getClassKey()->getHash()); + iter->second.objectClasses[key] = cls; + + // TODO: Indicate this schema if connected. +} + +void AgentImpl::registerClass(SchemaEventClass* cls) +{ + Mutex::ScopedLock _lock(lock); + + map::iterator iter = packages.find(cls->getClassKey()->getPackageName()); + if (iter == packages.end()) { + packages[cls->getClassKey()->getPackageName()] = ClassMaps(); + iter = packages.find(cls->getClassKey()->getPackageName()); + // TODO: Indicate this package if connected + } + + AgentClassKey key(cls->getClassKey()->getClassName(), cls->getClassKey()->getHash()); + iter->second.eventClasses[key] = cls; + + // TODO: Indicate this schema if connected. +} + +const ObjectId* AgentImpl::addObject(Object&, uint64_t) +{ + Mutex::ScopedLock _lock(lock); + return 0; +} + +const ObjectId* AgentImpl::allocObjectId(uint64_t persistId) +{ + Mutex::ScopedLock _lock(lock); + uint16_t sequence = persistId ? 0 : bootSequence; + uint64_t objectNum = persistId ? persistId : nextObjectId++; + + ObjectId* oid = ObjectIdImpl::factory(&attachment, 0, sequence, objectNum); + return oid; +} + +const ObjectId* AgentImpl::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) +{ + return allocObjectId(((uint64_t) persistIdHi) << 32 | (uint64_t) persistIdLo); +} + +void AgentImpl::raiseEvent(Event&) +{ + Mutex::ScopedLock _lock(lock); +} + +AgentEventImpl::Ptr AgentImpl::eventDeclareQueue(const string& name) +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::DECLARE_QUEUE)); + event->name = name; + + return event; +} + +AgentEventImpl::Ptr AgentImpl::eventBind(const string& exchange, const string& queue, + const string& key) +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::BIND)); + event->name = queue; + event->exchange = exchange; + event->bindingKey = key; + + return event; +} + +AgentEventImpl::Ptr AgentImpl::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 oid) +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::GET_QUERY)); + event->sequence = num; + event->authUserId = userId; + if (oid.get()) + event->query.reset(new Query(oid.get())); + else + event->query.reset(new Query(cls.c_str(), package.c_str())); + return event; +} + +AgentEventImpl::Ptr AgentImpl::eventMethod(uint32_t num, const string& userId, const string& method, + boost::shared_ptr oid, boost::shared_ptr argMap, + const SchemaObjectClass* objectClass) +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::METHOD_CALL)); + event->sequence = num; + event->authUserId = userId; + event->name = method; + event->objectId = oid; + event->arguments = argMap; + event->objectClass = objectClass; + return event; +} + +void AgentImpl::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 = "amq.direct"; + message->replyKey = queueName; + + xmtQueue.push_back(message); +} + +void AgentImpl::sendPackageIndicationLH(const string& packageName) +{ + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + 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 AgentClassKey& key) +{ + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + Protocol::encodeHeader(buffer, Protocol::OP_CLASS_INDICATION); + buffer.putOctet((int) kind); + buffer.putShortString(packageName); + buffer.putShortString(key.name); + buffer.putBin128(const_cast(key.hash)); // const_cast needed for older Qpid libraries + sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); + 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) +{ + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + 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) +{ + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + Protocol::encodeHeader(buffer, Protocol::OP_METHOD_RESPONSE, sequence); + buffer.putLong(code); + + string fulltext; + switch (code) { + case MERR_UNKNOWN_PACKAGE: fulltext = "Unknown Package"; break; + case MERR_UNKNOWN_CLASS: fulltext = "Unknown Class"; break; + case MERR_UNKNOWN_METHOD: fulltext = "Unknown Method"; break; + case MERR_INTERNAL_ERROR: fulltext = "Internal Error"; break; + default: fulltext = "Unspecified Error"; break; + } + + if (!text.empty()) { + fulltext += " ("; + fulltext += text; + fulltext += ")"; + } + + buffer.putMediumString(fulltext); + sendBufferLH(buffer, DIR_EXCHANGE, key); + QPID_LOG(trace, "SENT MethodResponse: errorCode=" << code << " text=" << fulltext); +} + +void AgentImpl::handleAttachResponse(Buffer& inBuffer) +{ + Mutex::ScopedLock _lock(lock); + + assignedBrokerBank = inBuffer.getLong(); + assignedAgentBank = inBuffer.getLong(); + + QPID_LOG(trace, "RCVD AttachResponse: broker=" << assignedBrokerBank << " agent=" << assignedAgentBank); + + if ((assignedBrokerBank != requestedBrokerBank) || + (assignedAgentBank != requestedAgentBank)) { + if (requestedAgentBank == 0) { + QPID_LOG(notice, "Initial object-id bank assigned: " << assignedBrokerBank << "." << + assignedAgentBank); + } else { + QPID_LOG(warning, "Collision in object-id! New bank assigned: " << assignedBrokerBank << + "." << assignedAgentBank); + } + //storeData(); // TODO + requestedBrokerBank = assignedBrokerBank; + requestedAgentBank = assignedAgentBank; + } + + attachment.setBanks(assignedBrokerBank, assignedAgentBank); + + // Bind to qpid.management to receive commands + stringstream key; + key << "agent." << assignedBrokerBank << "." << assignedAgentBank; + eventQueue.push_back(eventBind(QMF_EXCHANGE, queueName, key.str())); + + // Send package indications for all local packages + for (map::iterator pIter = packages.begin(); + pIter != packages.end(); + pIter++) { + sendPackageIndicationLH(pIter->first); + + // Send class indications for all local classes + ClassMaps cMap = pIter->second; + for (ObjectClassMap::iterator cIter = cMap.objectClasses.begin(); + cIter != cMap.objectClasses.end(); cIter++) + sendClassIndicationLH(CLASS_OBJECT, pIter->first, cIter->first); + for (EventClassMap::iterator cIter = cMap.eventClasses.begin(); + cIter != cMap.eventClasses.end(); cIter++) + sendClassIndicationLH(CLASS_EVENT, pIter->first, cIter->first); + } +} + +void AgentImpl::handlePackageRequest(Buffer&) +{ + Mutex::ScopedLock _lock(lock); +} + +void AgentImpl::handleClassQuery(Buffer&) +{ + Mutex::ScopedLock _lock(lock); +} + +void AgentImpl::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); + AgentClassKey key(inBuffer); + + if (rExchange.empty()) + rExchange = QMF_EXCHANGE; + if (rKey.empty()) + rKey = BROKER_KEY; + + QPID_LOG(trace, "RCVD SchemaRequest: package=" << packageName << " class=" << key.name); + + map::iterator pIter = packages.find(packageName); + if (pIter == packages.end()) { + sendCommandCompleteLH(rExchange, rKey, sequence, 1, "package not found"); + return; + } + + ClassMaps cMap = pIter->second; + ObjectClassMap::iterator ocIter = cMap.objectClasses.find(key); + if (ocIter != cMap.objectClasses.end()) { + SchemaObjectClass* oImpl = ocIter->second; + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + Protocol::encodeHeader(buffer, Protocol::OP_SCHEMA_RESPONSE, sequence); + oImpl->impl->encode(buffer); + sendBufferLH(buffer, rExchange, rKey); + QPID_LOG(trace, "SENT SchemaResponse: (object) package=" << packageName << " class=" << key.name); + return; + } + + EventClassMap::iterator ecIter = cMap.eventClasses.find(key); + if (ecIter != cMap.eventClasses.end()) { + SchemaEventClass* eImpl = ecIter->second; + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + Protocol::encodeHeader(buffer, Protocol::OP_SCHEMA_RESPONSE, sequence); + eImpl->impl->encode(buffer); + sendBufferLH(buffer, rExchange, rKey); + QPID_LOG(trace, "SENT SchemaResponse: (event) package=" << packageName << " class=" << key.name); + return; + } + + sendCommandCompleteLH(rExchange, rKey, sequence, 1, "class not found"); +} + +void AgentImpl::handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId) +{ + Mutex::ScopedLock _lock(lock); + FieldTable ft; + FieldTable::ValuePtr value; + map::const_iterator pIter = packages.end(); + string pname; + string cname; + string oidRepr; + boost::shared_ptr oid; + + ft.decode(inBuffer); + + QPID_LOG(trace, "RCVD GetQuery: seq=" << sequence << " map=" << ft); + + value = ft.get("_package"); + if (value.get() && value->convertsTo()) { + pname = value->get(); + pIter = packages.find(pname); + if (pIter == packages.end()) { + sendCommandCompleteLH(DIR_EXCHANGE, replyTo, sequence); + return; + } + } + + value = ft.get("_class"); + if (value.get() && value->convertsTo()) { + cname = value->get(); + // TODO - check for validity of class (in package or any package) + if (pIter == packages.end()) { + } else { + + } + } + + value = ft.get("_objectid"); + if (value.get() && value->convertsTo()) { + oidRepr = value->get(); + oid.reset(new ObjectId()); + oid->impl->fromString(oidRepr); + } + + AgentQueryContext::Ptr context(new AgentQueryContext); + uint32_t contextNum = nextContextNum++; + context->sequence = sequence; + context->exchange = DIR_EXCHANGE; + context->key = replyTo; + contextMap[contextNum] = context; + + eventQueue.push_back(eventQuery(contextNum, userId, pname, cname, oid)); +} + +void AgentImpl::handleMethodRequest(Buffer& buffer, uint32_t sequence, const string& replyTo, const string& userId) +{ + Mutex::ScopedLock _lock(lock); + string pname; + string method; + boost::shared_ptr oid(ObjectIdImpl::factory(buffer)); + buffer.getShortString(pname); + AgentClassKey classKey(buffer); + buffer.getShortString(method); + + QPID_LOG(trace, "RCVD MethodRequest seq=" << sequence << " method=" << method); + + map::const_iterator pIter = packages.find(pname); + if (pIter == packages.end()) { + sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_PACKAGE, pname); + return; + } + + ObjectClassMap::const_iterator cIter = pIter->second.objectClasses.find(classKey); + if (cIter == pIter->second.objectClasses.end()) { + sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_CLASS, classKey.repr()); + return; + } + + const SchemaObjectClass* schema = cIter->second; + vector::const_iterator mIter = schema->impl->methods.begin(); + for (; mIter != schema->impl->methods.end(); mIter++) { + if ((*mIter)->getName() == method) + break; + } + + if (mIter == schema->impl->methods.end()) { + sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_METHOD, method); + return; + } + + const SchemaMethod* schemaMethod = *mIter; + boost::shared_ptr argMap(new Value(TYPE_MAP)); + Value* value; + for (vector::const_iterator aIter = schemaMethod->impl->arguments.begin(); + aIter != schemaMethod->impl->arguments.end(); aIter++) { + const SchemaArgument* schemaArg = *aIter; + if (schemaArg->getDirection() == DIR_IN || schemaArg->getDirection() == DIR_IN_OUT) + value = ValueImpl::factory(schemaArg->getType(), buffer); + else + value = ValueImpl::factory(schemaArg->getType()); + argMap->insert(schemaArg->getName(), value); + } + + AgentQueryContext::Ptr context(new AgentQueryContext); + uint32_t contextNum = nextContextNum++; + context->sequence = sequence; + context->exchange = DIR_EXCHANGE; + context->key = replyTo; + context->schemaMethod = schemaMethod; + contextMap[contextNum] = context; + + eventQueue.push_back(eventMethod(contextNum, userId, method, oid, argMap, schema)); +} + +void AgentImpl::handleConsoleAddedIndication() +{ + Mutex::ScopedLock _lock(lock); +} + +//================================================================== +// Wrappers +//================================================================== + +Agent::Agent(char* label, bool internalStore) { impl = new AgentImpl(label, internalStore); } +Agent::~Agent() { delete impl; } +void Agent::setStoreDir(const char* path) { impl->setStoreDir(path); } +void Agent::setTransferDir(const char* path) { impl->setTransferDir(path); } +void Agent::handleRcvMessage(Message& message) { impl->handleRcvMessage(message); } +bool Agent::getXmtMessage(Message& item) const { return impl->getXmtMessage(item); } +void Agent::popXmt() { impl->popXmt(); } +bool Agent::getEvent(AgentEvent& event) const { 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); } + diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp new file mode 100644 index 0000000000..36d3ffe361 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp @@ -0,0 +1,704 @@ +/* + * 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/engine/BrokerProxyImpl.h" +#include "qmf/engine/ConsoleImpl.h" +#include "qmf/engine/Protocol.h" +#include "qpid/Address.h" +#include "qpid/sys/SystemInfo.h" +#include +#include +#include +#include + +using namespace std; +using namespace qmf::engine; +using namespace qpid::framing; +using namespace qpid::sys; + +namespace { + const char* QMF_EXCHANGE = "qpid.management"; + const char* DIR_EXCHANGE = "amq.direct"; + const char* BROKER_KEY = "broker"; + const char* BROKER_PACKAGE = "org.apache.qpid.broker"; + const char* AGENT_CLASS = "agent"; + const char* BROKER_AGENT_KEY = "agent.1.0"; +} + +const Object* QueryResponseImpl::getObject(uint32_t idx) const +{ + vector::const_iterator iter = results.begin(); + + while (idx > 0) { + if (iter == results.end()) + return 0; + iter++; + idx--; + } + + return iter->get(); +} + +#define STRING_REF(s) {if (!s.empty()) item.s = const_cast(s.c_str());} + +BrokerEvent BrokerEventImpl::copy() +{ + BrokerEvent item; + + ::memset(&item, 0, sizeof(BrokerEvent)); + item.kind = kind; + + STRING_REF(name); + STRING_REF(exchange); + STRING_REF(bindingKey); + item.context = context; + item.queryResponse = queryResponse.get(); + item.methodResponse = methodResponse.get(); + + return item; +} + +BrokerProxyImpl::BrokerProxyImpl(BrokerProxy& pub, Console& _console) : publicObject(pub), console(_console) +{ + stringstream qn; + qpid::TcpAddress addr; + + SystemInfo::getLocalHostname(addr); + qn << "qmfc-" << SystemInfo::getProcessName() << "-" << addr << "-" << SystemInfo::getProcessId(); + queueName = qn.str(); + + seqMgr.setUnsolicitedContext(SequenceContext::Ptr(new StaticContext(*this))); +} + +void BrokerProxyImpl::sessionOpened(SessionHandle& /*sh*/) +{ + Mutex::ScopedLock _lock(lock); + agentList.clear(); + 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); + agentList.clear(); + eventQueue.clear(); + xmtQueue.clear(); +} + +void BrokerProxyImpl::startProtocol() +{ + Mutex::ScopedLock _lock(lock); + char rawbuffer[512]; + Buffer buffer(rawbuffer, 512); + + agentList[0] = AgentProxyPtr(AgentProxyImpl::factory(console, publicObject, 0, "Agent embedded in broker")); + + requestsOutstanding = 1; + topicBound = false; + uint32_t sequence(seqMgr.reserve()); + Protocol::encodeHeader(buffer, Protocol::OP_BROKER_REQUEST, sequence); + sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT BrokerRequest seq=" << sequence); +} + +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)) + seqMgr.dispatch(opcode, sequence, inBuffer); +} + +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(); +} + +uint32_t BrokerProxyImpl::agentCount() const +{ + Mutex::ScopedLock _lock(lock); + return agentList.size(); +} + +const AgentProxy* BrokerProxyImpl::getAgent(uint32_t idx) const +{ + Mutex::ScopedLock _lock(lock); + for (map::const_iterator iter = agentList.begin(); + iter != agentList.end(); iter++) + if (idx-- == 0) + return iter->second.get(); + return 0; +} + +void BrokerProxyImpl::sendQuery(const Query& query, void* context, const AgentProxy* agent) +{ + SequenceContext::Ptr queryContext(new QueryContext(*this, context)); + Mutex::ScopedLock _lock(lock); + if (agent != 0) { + sendGetRequestLH(queryContext, query, agent); + } else { + // TODO (optimization) only send queries to agents that have the requested class+package + for (map::const_iterator iter = agentList.begin(); + iter != agentList.end(); iter++) { + sendGetRequestLH(queryContext, query, iter->second.get()); + } + } +} + +void BrokerProxyImpl::sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent) +{ + stringstream key; + Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + uint32_t sequence(seqMgr.reserve(queryContext)); + + Protocol::encodeHeader(outBuffer, Protocol::OP_GET_QUERY, sequence); + query.impl->encode(outBuffer); + key << "agent.1." << agent->impl->agentBank; + sendBufferLH(outBuffer, QMF_EXCHANGE, key.str()); + QPID_LOG(trace, "SENT GetQuery seq=" << sequence << " key=" << key.str()); +} + +string BrokerProxyImpl::encodeMethodArguments(const SchemaMethod* schema, const Value* argmap, Buffer& buffer) +{ + int argCount = schema->getArgumentCount(); + + if (argmap == 0 || !argmap->isMap()) + return string("Arguments must be in a map value"); + + for (int aIdx = 0; aIdx < argCount; aIdx++) { + const SchemaArgument* arg(schema->getArgument(aIdx)); + if (arg->getDirection() == DIR_IN || arg->getDirection() == DIR_IN_OUT) { + if (argmap->keyInMap(arg->getName())) { + const Value* argVal(argmap->byKey(arg->getName())); + if (argVal->getType() != arg->getType()) + return string("Argument is the wrong type: ") + arg->getName(); + argVal->impl->encode(buffer); + } else { + Value defaultValue(arg->getType()); + defaultValue.impl->encode(buffer); + } + } + } + + return string(); +} + +void BrokerProxyImpl::sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls, + const string& methodName, const Value* args, void* userContext) +{ + int methodCount = cls->getMethodCount(); + int idx; + for (idx = 0; idx < methodCount; idx++) { + const SchemaMethod* method = cls->getMethod(idx); + if (string(method->getName()) == methodName) { + Mutex::ScopedLock _lock(lock); + SequenceContext::Ptr methodContext(new MethodContext(*this, userContext, method)); + stringstream key; + Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + uint32_t sequence(seqMgr.reserve(methodContext)); + + Protocol::encodeHeader(outBuffer, Protocol::OP_METHOD_REQUEST, sequence); + oid->impl->encode(outBuffer); + cls->getClassKey()->impl->encode(outBuffer); + outBuffer.putShortString(methodName); + + string argErrorString = encodeMethodArguments(method, args, outBuffer); + if (argErrorString.empty()) { + key << "agent.1." << oid->impl->getAgentBank(); + sendBufferLH(outBuffer, QMF_EXCHANGE, key.str()); + QPID_LOG(trace, "SENT MethodRequest seq=" << sequence << " method=" << methodName << " key=" << key.str()); + } else { + MethodResponsePtr argError(MethodResponseImpl::factory(1, argErrorString)); + eventQueue.push_back(eventMethodResponse(userContext, argError)); + } + return; + } + } + + MethodResponsePtr error(MethodResponseImpl::factory(1, string("Unknown method: ") + methodName)); + Mutex::ScopedLock _lock(lock); + eventQueue.push_back(eventMethodResponse(userContext, error)); +} + +void BrokerProxyImpl::addBinding(const string& exchange, const string& key) +{ + Mutex::ScopedLock _lock(lock); + 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() +{ + QPID_LOG(trace, "Console Link to Broker Stable"); + BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::STABLE)); + return event; +} + +BrokerEventImpl::Ptr BrokerProxyImpl::eventQueryComplete(void* context, QueryResponsePtr response) +{ + BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::QUERY_COMPLETE)); + event->context = context; + event->queryResponse = response; + return event; +} + +BrokerEventImpl::Ptr BrokerProxyImpl::eventMethodResponse(void* context, MethodResponsePtr response) +{ + BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::METHOD_RESPONSE)); + event->context = context; + event->methodResponse = response; + return event; +} + +void BrokerProxyImpl::handleBrokerResponse(Buffer& inBuffer, uint32_t seq) +{ + 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()); + incOutstandingLH(); + 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.impl->learnPackage(package); + + Mutex::ScopedLock _lock(lock); + Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + uint32_t sequence(seqMgr.reserve()); + 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); +} + +void BrokerProxyImpl::handleClassIndication(Buffer& inBuffer, uint32_t seq) +{ + uint8_t kind = inBuffer.getOctet(); + auto_ptr classKey(SchemaClassKeyImpl::factory(inBuffer)); + + QPID_LOG(trace, "RCVD ClassIndication seq=" << seq << " kind=" << (int) kind << " key=" << classKey->impl->str()); + + if (!console.impl->haveClass(classKey.get())) { + Mutex::ScopedLock _lock(lock); + incOutstandingLH(); + Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + uint32_t sequence(seqMgr.reserve()); + Protocol::encodeHeader(outBuffer, Protocol::OP_SCHEMA_REQUEST, sequence); + classKey->impl->encode(outBuffer); + sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT SchemaRequest seq=" << sequence <<" key=" << classKey->impl->str()); + } +} + +MethodResponsePtr BrokerProxyImpl::handleMethodResponse(Buffer& inBuffer, uint32_t seq, const SchemaMethod* schema) +{ + MethodResponsePtr response(MethodResponseImpl::factory(inBuffer, schema)); + + QPID_LOG(trace, "RCVD MethodResponse seq=" << seq << " status=" << response->getStatus() << " text=" << + response->getException()->asString()); + + return response; +} + +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) +{ + SchemaObjectClass* oClassPtr; + SchemaEventClass* eClassPtr; + uint8_t kind = inBuffer.getOctet(); + const SchemaClassKey* key; + if (kind == CLASS_OBJECT) { + oClassPtr = SchemaObjectClassImpl::factory(inBuffer); + console.impl->learnClass(oClassPtr); + key = oClassPtr->getClassKey(); + QPID_LOG(trace, "RCVD SchemaResponse seq=" << seq << " kind=object key=" << key->impl->str()); + + // + // If we have just learned about the org.apache.qpid.broker:agent class, send a get + // request for the current list of agents so we can have it on-hand before we declare + // this session "stable". + // + if (key->impl->getClassName() == AGENT_CLASS && key->impl->getPackageName() == BROKER_PACKAGE) { + Mutex::ScopedLock _lock(lock); + incOutstandingLH(); + Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + uint32_t sequence(seqMgr.reserve()); + Protocol::encodeHeader(outBuffer, Protocol::OP_GET_QUERY, sequence); + FieldTable ft; + ft.setString("_class", AGENT_CLASS); + ft.setString("_package", BROKER_PACKAGE); + ft.encode(outBuffer); + sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_AGENT_KEY); + QPID_LOG(trace, "SENT GetQuery seq=" << sequence << " key=" << BROKER_AGENT_KEY); + } + } else if (kind == CLASS_EVENT) { + eClassPtr = SchemaEventClassImpl::factory(inBuffer); + console.impl->learnClass(eClassPtr); + key = eClassPtr->getClassKey(); + QPID_LOG(trace, "RCVD SchemaResponse seq=" << seq << " kind=event key=" << key->impl->str()); + } + else { + QPID_LOG(error, "BrokerProxyImpl::handleSchemaResponse received unknown class kind: " << (int) kind); + } +} + +ObjectPtr BrokerProxyImpl::handleObjectIndication(Buffer& inBuffer, uint32_t seq, bool prop, bool stat) +{ + auto_ptr classKey(SchemaClassKeyImpl::factory(inBuffer)); + QPID_LOG(trace, "RCVD ObjectIndication seq=" << seq << " key=" << classKey->impl->str()); + + SchemaObjectClass* schema = console.impl->getSchema(classKey.get()); + if (schema == 0) { + QPID_LOG(trace, "No Schema Found for ObjectIndication. seq=" << seq << " key=" << classKey->impl->str()); + return ObjectPtr(); + } + + ObjectPtr optr(ObjectImpl::factory(schema, this, inBuffer, prop, stat, true)); + if (prop && classKey->impl->getPackageName() == BROKER_PACKAGE && classKey->impl->getClassName() == AGENT_CLASS) { + // + // We've intercepted information about a remote agent... update the agent list accordingly + // + updateAgentList(optr); + } + return optr; +} + +void BrokerProxyImpl::updateAgentList(ObjectPtr obj) +{ + Value* value = obj->getValue("agentBank"); + if (value != 0 && value->isUint()) { + uint32_t agentBank = value->asUint(); + if (obj->isDeleted()) { + agentList.erase(agentBank); + QPID_LOG(trace, "Agent at bank " << agentBank << " removed from agent list"); + } else { + Value* str = obj->getValue("label"); + string label; + if (str != 0 && str->isString()) + label = str->asString(); + map::const_iterator iter = agentList.find(agentBank); + if (iter == agentList.end()) { + agentList[agentBank] = AgentProxyPtr(AgentProxyImpl::factory(console, publicObject, agentBank, label)); + QPID_LOG(trace, "Agent '" << label << "' found at bank " << agentBank); + } + } + } +} + +void BrokerProxyImpl::incOutstandingLH() +{ + requestsOutstanding++; +} + +void BrokerProxyImpl::decOutstanding() +{ + Mutex::ScopedLock _lock(lock); + requestsOutstanding--; + if (requestsOutstanding == 0 && !topicBound) { + topicBound = true; + for (vector >::const_iterator iter = console.impl->bindingList.begin(); + iter != console.impl->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(const MethodResponseImpl& from) : + status(from.status), schema(from.schema) +{ + if (from.exception.get()) + exception.reset(new Value(*(from.exception))); + if (from.arguments.get()) + arguments.reset(new Value(*(from.arguments))); +} + +MethodResponseImpl::MethodResponseImpl(Buffer& buf, const SchemaMethod* s) : schema(s) +{ + string text; + + status = buf.getLong(); + buf.getMediumString(text); + exception.reset(new Value(TYPE_LSTR)); + exception->setString(text.c_str()); + + if (status != 0) + return; + + arguments.reset(new Value(TYPE_MAP)); + int argCount(schema->getArgumentCount()); + for (int idx = 0; idx < argCount; idx++) { + const SchemaArgument* arg = schema->getArgument(idx); + if (arg->getDirection() == DIR_OUT || arg->getDirection() == DIR_IN_OUT) { + Value* value(ValueImpl::factory(arg->getType(), buf)); + arguments->insert(arg->getName(), value); + } + } +} + +MethodResponseImpl::MethodResponseImpl(uint32_t s, const string& text) : schema(0) +{ + status = s; + exception.reset(new Value(TYPE_LSTR)); + exception->setString(text.c_str()); +} + +MethodResponse* MethodResponseImpl::factory(Buffer& buf, const SchemaMethod* schema) +{ + MethodResponseImpl* impl(new MethodResponseImpl(buf, schema)); + return new MethodResponse(impl); +} + +MethodResponse* MethodResponseImpl::factory(uint32_t status, const std::string& text) +{ + MethodResponseImpl* impl(new MethodResponseImpl(status, text)); + return new MethodResponse(impl); +} + +bool StaticContext::handleMessage(uint8_t opcode, uint32_t sequence, Buffer& buffer) +{ + bool completeContext = false; + if (opcode == Protocol::OP_BROKER_RESPONSE) { + broker.handleBrokerResponse(buffer, sequence); + completeContext = true; + } + else if (opcode == Protocol::OP_COMMAND_COMPLETE) { + broker.handleCommandComplete(buffer, sequence); + completeContext = true; + } + else if (opcode == Protocol::OP_SCHEMA_RESPONSE) { + broker.handleSchemaResponse(buffer, sequence); + completeContext = true; + } + else if (opcode == Protocol::OP_PACKAGE_INDICATION) + broker.handlePackageIndication(buffer, sequence); + else if (opcode == Protocol::OP_CLASS_INDICATION) + broker.handleClassIndication(buffer, sequence); + else if (opcode == Protocol::OP_HEARTBEAT_INDICATION) + broker.handleHeartbeatIndication(buffer, sequence); + else if (opcode == Protocol::OP_EVENT_INDICATION) + broker.handleEventIndication(buffer, sequence); + else if (opcode == Protocol::OP_PROPERTY_INDICATION) + broker.handleObjectIndication(buffer, sequence, true, false); + else if (opcode == Protocol::OP_STATISTIC_INDICATION) + broker.handleObjectIndication(buffer, sequence, false, true); + else if (opcode == Protocol::OP_OBJECT_INDICATION) + broker.handleObjectIndication(buffer, sequence, true, true); + else { + QPID_LOG(trace, "StaticContext::handleMessage invalid opcode: " << opcode); + completeContext = true; + } + + return completeContext; +} + +void QueryContext::reserve() +{ + Mutex::ScopedLock _lock(lock); + requestsOutstanding++; +} + +void QueryContext::release() +{ + { + Mutex::ScopedLock _lock(lock); + if (--requestsOutstanding > 0) + return; + } + + Mutex::ScopedLock _block(broker.lock); + broker.eventQueue.push_back(broker.eventQueryComplete(userContext, queryResponse)); +} + +bool QueryContext::handleMessage(uint8_t opcode, uint32_t sequence, Buffer& buffer) +{ + bool completeContext = false; + ObjectPtr object; + + if (opcode == Protocol::OP_COMMAND_COMPLETE) { + broker.handleCommandComplete(buffer, sequence); + completeContext = true; + } + else if (opcode == Protocol::OP_OBJECT_INDICATION) { + object = broker.handleObjectIndication(buffer, sequence, true, true); + if (object.get() != 0) + queryResponse->impl->results.push_back(object); + } + else { + QPID_LOG(trace, "QueryContext::handleMessage invalid opcode: " << opcode); + completeContext = true; + } + + return completeContext; +} + +void MethodContext::release() +{ + Mutex::ScopedLock _block(broker.lock); + broker.eventQueue.push_back(broker.eventMethodResponse(userContext, methodResponse)); +} + +bool MethodContext::handleMessage(uint8_t opcode, uint32_t sequence, Buffer& buffer) +{ + if (opcode == Protocol::OP_METHOD_RESPONSE) + methodResponse = broker.handleMethodResponse(buffer, sequence, schema); + else + QPID_LOG(trace, "QueryContext::handleMessage invalid opcode: " << opcode); + + return true; +} + + +//================================================================== +// Wrappers +//================================================================== + +AgentProxy::AgentProxy(AgentProxyImpl* i) : impl(i) {} +AgentProxy::~AgentProxy() { delete impl; } +const char* AgentProxy::getLabel() const { return impl->getLabel().c_str(); } + +BrokerProxy::BrokerProxy(Console& 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(); } +uint32_t BrokerProxy::agentCount() const { return impl->agentCount(); } +const AgentProxy* BrokerProxy::getAgent(uint32_t idx) const { return impl->getAgent(idx); } +void BrokerProxy::sendQuery(const Query& query, void* context, const AgentProxy* agent) { impl->sendQuery(query, context, agent); } + +MethodResponse::MethodResponse(const MethodResponse& from) : impl(new MethodResponseImpl(*(from.impl))) {} +MethodResponse::MethodResponse(MethodResponseImpl* i) : impl(i) {} +MethodResponse::~MethodResponse() {} +uint32_t MethodResponse::getStatus() const { return impl->getStatus(); } +const Value* MethodResponse::getException() const { return impl->getException(); } +const Value* MethodResponse::getArgs() const { return impl->getArgs(); } + +QueryResponse::QueryResponse(QueryResponseImpl* i) : impl(i) {} +QueryResponse::~QueryResponse() {} +uint32_t QueryResponse::getStatus() const { return impl->getStatus(); } +const Value* QueryResponse::getException() const { return impl->getException(); } +uint32_t QueryResponse::getObjectCount() const { return impl->getObjectCount(); } +const Object* QueryResponse::getObject(uint32_t idx) const { return impl->getObject(idx); } + diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h new file mode 100644 index 0000000000..660cb86c61 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h @@ -0,0 +1,228 @@ +#ifndef _QmfEngineBrokerProxyImpl_ +#define _QmfEngineBrokerProxyImpl_ + +/* + * 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/engine/Console.h" +#include "qmf/engine/ObjectImpl.h" +#include "qmf/engine/SchemaImpl.h" +#include "qmf/engine/ValueImpl.h" +#include "qmf/engine/QueryImpl.h" +#include "qmf/engine/SequenceManager.h" +#include "qmf/engine/MessageImpl.h" +#include "qpid/framing/Buffer.h" +#include "qpid/framing/Uuid.h" +#include "qpid/sys/Mutex.h" +#include "boost/shared_ptr.hpp" +#include "boost/noncopyable.hpp" +#include +#include +#include +#include +#include + +namespace qmf { +namespace engine { + + typedef boost::shared_ptr MethodResponsePtr; + struct MethodResponseImpl { + uint32_t status; + const SchemaMethod* schema; + std::auto_ptr exception; + std::auto_ptr arguments; + + MethodResponseImpl(const MethodResponseImpl& from); + MethodResponseImpl(qpid::framing::Buffer& buf, const SchemaMethod* schema); + MethodResponseImpl(uint32_t status, const std::string& text); + static MethodResponse* factory(qpid::framing::Buffer& buf, const SchemaMethod* schema); + static MethodResponse* factory(uint32_t status, const std::string& text); + ~MethodResponseImpl() {} + uint32_t getStatus() const { return status; } + const Value* getException() const { return exception.get(); } + const Value* getArgs() const { return arguments.get(); } + }; + + typedef boost::shared_ptr QueryResponsePtr; + struct QueryResponseImpl { + uint32_t status; + std::auto_ptr exception; + std::vector results; + + QueryResponseImpl() : status(0) {} + static QueryResponse* factory() { + QueryResponseImpl* impl(new QueryResponseImpl()); + return new QueryResponse(impl); + } + ~QueryResponseImpl() {} + uint32_t getStatus() const { return status; } + const Value* getException() const { return exception.get(); } + uint32_t getObjectCount() const { return results.size(); } + const Object* getObject(uint32_t idx) const; + }; + + struct BrokerEventImpl { + typedef boost::shared_ptr Ptr; + BrokerEvent::EventKind kind; + std::string name; + std::string exchange; + std::string bindingKey; + void* context; + QueryResponsePtr queryResponse; + MethodResponsePtr methodResponse; + + BrokerEventImpl(BrokerEvent::EventKind k) : kind(k), context(0) {} + ~BrokerEventImpl() {} + BrokerEvent copy(); + }; + + typedef boost::shared_ptr AgentProxyPtr; + struct AgentProxyImpl { + Console& console; + BrokerProxy& broker; + uint32_t agentBank; + std::string label; + + AgentProxyImpl(Console& c, BrokerProxy& b, uint32_t ab, const std::string& l) : console(c), broker(b), agentBank(ab), label(l) {} + static AgentProxy* factory(Console& c, BrokerProxy& b, uint32_t ab, const std::string& l) { + AgentProxyImpl* impl(new AgentProxyImpl(c, b, ab, l)); + return new AgentProxy(impl); + } + ~AgentProxyImpl() {} + const std::string& getLabel() const { return label; } + }; + + class BrokerProxyImpl : public boost::noncopyable { + public: + BrokerProxyImpl(BrokerProxy& pub, Console& _console); + ~BrokerProxyImpl() {} + + void sessionOpened(SessionHandle& sh); + void sessionClosed(); + void startProtocol(); + + void sendBufferLH(qpid::framing::Buffer& buf, const std::string& destination, const std::string& routingKey); + void handleRcvMessage(Message& message); + bool getXmtMessage(Message& item) const; + void popXmt(); + + bool getEvent(BrokerEvent& event) const; + void popEvent(); + + uint32_t agentCount() const; + const AgentProxy* getAgent(uint32_t idx) const; + void sendQuery(const Query& query, void* context, const AgentProxy* agent); + void sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent); + std::string encodeMethodArguments(const SchemaMethod* schema, const Value* args, qpid::framing::Buffer& buffer); + void sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls, const std::string& method, const Value* args, void* context); + + void addBinding(const std::string& exchange, const std::string& key); + void staticRelease() { decOutstanding(); } + + private: + friend struct StaticContext; + friend struct QueryContext; + friend struct MethodContext; + BrokerProxy& publicObject; + mutable qpid::sys::Mutex lock; + Console& console; + std::string queueName; + qpid::framing::Uuid brokerId; + SequenceManager seqMgr; + uint32_t requestsOutstanding; + bool topicBound; + std::map agentList; + std::deque xmtQueue; + std::deque eventQueue; + +# define MA_BUFFER_SIZE 65536 + char outputBuffer[MA_BUFFER_SIZE]; + + BrokerEventImpl::Ptr eventDeclareQueue(const std::string& queueName); + BrokerEventImpl::Ptr eventBind(const std::string& exchange, const std::string& queue, const std::string& key); + BrokerEventImpl::Ptr eventSetupComplete(); + BrokerEventImpl::Ptr eventStable(); + BrokerEventImpl::Ptr eventQueryComplete(void* context, QueryResponsePtr response); + BrokerEventImpl::Ptr eventMethodResponse(void* context, MethodResponsePtr response); + + void handleBrokerResponse(qpid::framing::Buffer& inBuffer, uint32_t seq); + void handlePackageIndication(qpid::framing::Buffer& inBuffer, uint32_t seq); + void handleCommandComplete(qpid::framing::Buffer& inBuffer, uint32_t seq); + void handleClassIndication(qpid::framing::Buffer& inBuffer, uint32_t seq); + MethodResponsePtr handleMethodResponse(qpid::framing::Buffer& inBuffer, uint32_t seq, const SchemaMethod* schema); + void handleHeartbeatIndication(qpid::framing::Buffer& inBuffer, uint32_t seq); + void handleEventIndication(qpid::framing::Buffer& inBuffer, uint32_t seq); + void handleSchemaResponse(qpid::framing::Buffer& inBuffer, uint32_t seq); + ObjectPtr handleObjectIndication(qpid::framing::Buffer& inBuffer, uint32_t seq, bool prop, bool stat); + void updateAgentList(ObjectPtr obj); + void incOutstandingLH(); + void decOutstanding(); + }; + + // + // StaticContext is used to handle: + // + // 1) Responses to console-level requests (for schema info, etc.) + // 2) Unsolicited messages from agents (events, published updates, etc.) + // + struct StaticContext : public SequenceContext { + StaticContext(BrokerProxyImpl& b) : broker(b) {} + virtual ~StaticContext() {} + void reserve() {} + void release() { broker.staticRelease(); } + bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer); + BrokerProxyImpl& broker; + }; + + // + // QueryContext is used to track and handle responses associated with a single Get Query + // + struct QueryContext : public SequenceContext { + QueryContext(BrokerProxyImpl& b, void* u) : + broker(b), userContext(u), requestsOutstanding(0), queryResponse(QueryResponseImpl::factory()) {} + virtual ~QueryContext() {} + void reserve(); + void release(); + bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer); + + mutable qpid::sys::Mutex lock; + BrokerProxyImpl& broker; + void* userContext; + uint32_t requestsOutstanding; + QueryResponsePtr queryResponse; + }; + + struct MethodContext : public SequenceContext { + MethodContext(BrokerProxyImpl& b, void* u, const SchemaMethod* s) : broker(b), userContext(u), schema(s) {} + virtual ~MethodContext() {} + void reserve() {} + void release(); + bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer); + + BrokerProxyImpl& broker; + void* userContext; + const SchemaMethod* schema; + MethodResponsePtr methodResponse; + }; + +} +} + +#endif + diff --git a/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp new file mode 100644 index 0000000000..8c918eec20 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp @@ -0,0 +1,270 @@ +/* + * 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/engine/ConnectionSettingsImpl.h" +#include "qmf/engine/Typecode.h" + +using namespace std; +using namespace qmf::engine; +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"); +const string attrSendUserId("sendUserId"); + +ConnectionSettingsImpl::ConnectionSettingsImpl() : + retryDelayMin(1), retryDelayMax(64), retryDelayFactor(2), sendUserId(true) +{ +} + +ConnectionSettingsImpl::ConnectionSettingsImpl(const string& /*url*/) : + retryDelayMin(1), retryDelayMax(64), retryDelayFactor(2), sendUserId(true) +{ + // 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(); + else if (key == attrSendUserId) sendUserId = value.asBool(); +} + +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(); } +ConnectionSettings::ConnectionSettings(const char* url) { impl = new ConnectionSettingsImpl(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/engine/ConnectionSettingsImpl.h b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h new file mode 100644 index 0000000000..c47c8bee2e --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h @@ -0,0 +1,63 @@ +#ifndef _QmfEngineConnectionSettingsImpl_ +#define _QmfEngineConnectionSettingsImpl_ + +/* + * 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/engine/ConnectionSettings.h" +#include "qmf/engine/Value.h" +#include "qpid/client/ConnectionSettings.h" +#include +#include + +namespace qmf { +namespace engine { + + class ConnectionSettingsImpl { + qpid::client::ConnectionSettings clientSettings; + mutable std::string attrString; + int retryDelayMin; + int retryDelayMax; + int retryDelayFactor; + bool sendUserId; + + public: + ConnectionSettingsImpl(); + ConnectionSettingsImpl(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; + bool getSendUserId() const { return sendUserId; } + }; + +} +} + +#endif diff --git a/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp b/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp new file mode 100644 index 0000000000..c856f04c51 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp @@ -0,0 +1,360 @@ +/* + * 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/engine/ConsoleImpl.h" +#include "qmf/engine/MessageImpl.h" +#include "qmf/engine/SchemaImpl.h" +#include "qmf/engine/Typecode.h" +#include "qmf/engine/ObjectImpl.h" +#include "qmf/engine/ObjectIdImpl.h" +#include "qmf/engine/QueryImpl.h" +#include "qmf/engine/ValueImpl.h" +#include "qmf/engine/Protocol.h" +#include "qmf/engine/SequenceManager.h" +#include "qmf/engine/BrokerProxyImpl.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std; +using namespace qmf::engine; +using namespace qpid::framing; +using namespace qpid::sys; + +namespace { + const char* QMF_EXCHANGE = "qpid.management"; +} + +#define STRING_REF(s) {if (!s.empty()) item.s = const_cast(s.c_str());} + +ConsoleEvent ConsoleEventImpl::copy() +{ + ConsoleEvent item; + + ::memset(&item, 0, sizeof(ConsoleEvent)); + item.kind = kind; + item.agent = agent.get(); + item.classKey = classKey.get(); + item.object = object; + item.context = context; + item.event = event; + item.timestamp = timestamp; + + STRING_REF(name); + + return item; +} + +ConsoleImpl::ConsoleImpl(const ConsoleSettings& s) : settings(s) +{ + bindingList.push_back(pair(string(), "schema.#")); + if (settings.rcvObjects && settings.rcvEvents && settings.rcvHeartbeats && !settings.userBindings) { + bindingList.push_back(pair(string(), "console.#")); + } else { + if (settings.rcvObjects && !settings.userBindings) + bindingList.push_back(pair(string(), "console.obj.#")); + else + bindingList.push_back(pair(string(), "console.obj.*.*.org.apache.qpid.broker.agent")); + if (settings.rcvEvents) + bindingList.push_back(pair(string(), "console.event.#")); + if (settings.rcvHeartbeats) + bindingList.push_back(pair(string(), "console.heartbeat.#")); + } +} + +ConsoleImpl::~ConsoleImpl() +{ + // This function intentionally left blank. +} + +bool ConsoleImpl::getEvent(ConsoleEvent& event) const +{ + Mutex::ScopedLock _lock(lock); + if (eventQueue.empty()) + return false; + event = eventQueue.front()->copy(); + return true; +} + +void ConsoleImpl::popEvent() +{ + Mutex::ScopedLock _lock(lock); + if (!eventQueue.empty()) + eventQueue.pop_front(); +} + +void ConsoleImpl::addConnection(BrokerProxy& broker, void* /*context*/) +{ + Mutex::ScopedLock _lock(lock); + brokerList.push_back(broker.impl); +} + +void ConsoleImpl::delConnection(BrokerProxy& broker) +{ + Mutex::ScopedLock _lock(lock); + for (vector::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + if (*iter == broker.impl) { + brokerList.erase(iter); + break; + } +} + +uint32_t ConsoleImpl::packageCount() const +{ + Mutex::ScopedLock _lock(lock); + return packages.size(); +} + +const string& ConsoleImpl::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 ConsoleImpl::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* ConsoleImpl::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 ConsoleImpl::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) != eList.end()) + return CLASS_EVENT; + return CLASS_OBJECT; +} + +const SchemaObjectClass* ConsoleImpl::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); + if (iter == oList.end()) + return 0; + return iter->second; +} + +const SchemaEventClass* ConsoleImpl::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); + if (iter == eList.end()) + return 0; + return iter->second; +} + +void ConsoleImpl::bindPackage(const char* packageName) +{ + stringstream key; + key << "console.obj.*.*." << packageName << ".#"; + Mutex::ScopedLock _lock(lock); + bindingList.push_back(pair(string(), key.str())); + for (vector::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + (*iter)->addBinding(QMF_EXCHANGE, key.str()); +} + +void ConsoleImpl::bindClass(const SchemaClassKey* classKey) +{ + stringstream key; + key << "console.obj.*.*." << classKey->getPackageName() << "." << classKey->getClassName() << ".#"; + Mutex::ScopedLock _lock(lock); + bindingList.push_back(pair(string(), key.str())); + for (vector::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + (*iter)->addBinding(QMF_EXCHANGE, key.str()); +} + +void ConsoleImpl::bindClass(const char* packageName, const char* className) +{ + stringstream key; + key << "console.obj.*.*." << packageName << "." << className << ".#"; + Mutex::ScopedLock _lock(lock); + bindingList.push_back(pair(string(), key.str())); + for (vector::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + (*iter)->addBinding(QMF_EXCHANGE, key.str()); +} + +/* +void ConsoleImpl::startSync(const Query& query, void* context, SyncQuery& sync) +{ +} + +void ConsoleImpl::touchSync(SyncQuery& sync) +{ +} + +void ConsoleImpl::endSync(SyncQuery& sync) +{ +} +*/ + +void ConsoleImpl::learnPackage(const string& packageName) +{ + Mutex::ScopedLock _lock(lock); + if (packages.find(packageName) == packages.end()) + packages.insert(pair > + (packageName, pair(ObjectClassList(), EventClassList()))); +} + +void ConsoleImpl::learnClass(SchemaObjectClass* 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) == list.end()) + list[key] = cls; +} + +void ConsoleImpl::learnClass(SchemaEventClass* 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) == list.end()) + list[key] = cls; +} + +bool ConsoleImpl::haveClass(const SchemaClassKey* 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(); +} + +SchemaObjectClass* ConsoleImpl::getSchema(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); + if (iter == oList.end()) + return 0; + + return iter->second; +} + +//================================================================== +// Wrappers +//================================================================== + +Console::Console(const ConsoleSettings& settings) : impl(new ConsoleImpl(settings)) {} +Console::~Console() { delete impl; } +bool Console::getEvent(ConsoleEvent& event) const { return impl->getEvent(event); } +void Console::popEvent() { impl->popEvent(); } +void Console::addConnection(BrokerProxy& broker, void* context) { impl->addConnection(broker, context); } +void Console::delConnection(BrokerProxy& broker) { impl->delConnection(broker); } +uint32_t Console::packageCount() const { return impl->packageCount(); } +const char* Console::getPackageName(uint32_t idx) const { return impl->getPackageName(idx).c_str(); } +uint32_t Console::classCount(const char* packageName) const { return impl->classCount(packageName); } +const SchemaClassKey* Console::getClass(const char* packageName, uint32_t idx) const { return impl->getClass(packageName, idx); } +ClassKind Console::getClassKind(const SchemaClassKey* key) const { return impl->getClassKind(key); } +const SchemaObjectClass* Console::getObjectClass(const SchemaClassKey* key) const { return impl->getObjectClass(key); } +const SchemaEventClass* Console::getEventClass(const SchemaClassKey* key) const { return impl->getEventClass(key); } +void Console::bindPackage(const char* packageName) { impl->bindPackage(packageName); } +void Console::bindClass(const SchemaClassKey* key) { impl->bindClass(key); } +void Console::bindClass(const char* packageName, const char* className) { impl->bindClass(packageName, className); } +//void Console::startSync(const Query& query, void* context, SyncQuery& sync) { impl->startSync(query, context, sync); } +//void Console::touchSync(SyncQuery& sync) { impl->touchSync(sync); } +//void Console::endSync(SyncQuery& sync) { impl->endSync(sync); } + + diff --git a/qpid/cpp/src/qmf/engine/ConsoleImpl.h b/qpid/cpp/src/qmf/engine/ConsoleImpl.h new file mode 100644 index 0000000000..2c4ee48a02 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ConsoleImpl.h @@ -0,0 +1,135 @@ +#ifndef _QmfEngineConsoleEngineImpl_ +#define _QmfEngineConsoleEngineImpl_ + +/* + * 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/engine/Console.h" +#include "qmf/engine/MessageImpl.h" +#include "qmf/engine/SchemaImpl.h" +#include "qmf/engine/Typecode.h" +#include "qmf/engine/ObjectImpl.h" +#include "qmf/engine/ObjectIdImpl.h" +#include "qmf/engine/QueryImpl.h" +#include "qmf/engine/ValueImpl.h" +#include "qmf/engine/Protocol.h" +#include "qmf/engine/SequenceManager.h" +#include "qmf/engine/BrokerProxyImpl.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace qmf { +namespace engine { + + struct ConsoleEventImpl { + typedef boost::shared_ptr Ptr; + ConsoleEvent::EventKind kind; + boost::shared_ptr agent; + std::string name; + boost::shared_ptr classKey; + Object* object; + void* context; + Event* event; + uint64_t timestamp; + + ConsoleEventImpl(ConsoleEvent::EventKind k) : + kind(k), object(0), context(0), event(0), timestamp(0) {} + ~ConsoleEventImpl() {} + ConsoleEvent copy(); + }; + + class ConsoleImpl : public boost::noncopyable { + public: + ConsoleImpl(const ConsoleSettings& settings = ConsoleSettings()); + ~ConsoleImpl(); + + bool getEvent(ConsoleEvent& event) const; + void popEvent(); + + void addConnection(BrokerProxy& broker, void* context); + void delConnection(BrokerProxy& broker); + + uint32_t packageCount() const; + const std::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); + + /* + void startSync(const Query& query, void* context, SyncQuery& sync); + void touchSync(SyncQuery& sync); + void endSync(SyncQuery& sync); + */ + + private: + friend class BrokerProxyImpl; + const ConsoleSettings& settings; + mutable qpid::sys::Mutex lock; + std::deque eventQueue; + std::vector brokerList; + std::vector > 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 SchemaClassKey* left, const SchemaClassKey* right) const { + return *left < *right; + } + }; + + typedef std::map ObjectClassList; + typedef std::map EventClassList; + typedef std::map > PackageList; + + PackageList packages; + + void learnPackage(const std::string& packageName); + void learnClass(SchemaObjectClass* cls); + void learnClass(SchemaEventClass* cls); + bool haveClass(const SchemaClassKey* key) const; + SchemaObjectClass* getSchema(const SchemaClassKey* key) const; + }; +} +} + +#endif + diff --git a/qpid/cpp/src/qmf/engine/MessageImpl.cpp b/qpid/cpp/src/qmf/engine/MessageImpl.cpp new file mode 100644 index 0000000000..0047d3eb9d --- /dev/null +++ b/qpid/cpp/src/qmf/engine/MessageImpl.cpp @@ -0,0 +1,43 @@ +/* + * 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/engine/MessageImpl.h" +#include + +using namespace std; +using namespace qmf::engine; + +#define STRING_REF(s) {if (!s.empty()) item.s = const_cast(s.c_str());} + +Message MessageImpl::copy() +{ + Message item; + + ::memset(&item, 0, sizeof(Message)); + item.body = const_cast(body.c_str()); + item.length = body.length(); + STRING_REF(destination); + STRING_REF(routingKey); + STRING_REF(replyExchange); + STRING_REF(replyKey); + STRING_REF(userId); + + return item; +} + diff --git a/qpid/cpp/src/qmf/engine/MessageImpl.h b/qpid/cpp/src/qmf/engine/MessageImpl.h new file mode 100644 index 0000000000..b91291d2e4 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/MessageImpl.h @@ -0,0 +1,44 @@ +#ifndef _QmfEngineMessageImpl_ +#define _QmfEngineMessageImpl_ + +/* + * 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/engine/Message.h" +#include +#include + +namespace qmf { +namespace engine { + + struct MessageImpl { + typedef boost::shared_ptr Ptr; + std::string body; + std::string destination; + std::string routingKey; + std::string replyExchange; + std::string replyKey; + std::string userId; + + Message copy(); + }; +} +} + +#endif diff --git a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp new file mode 100644 index 0000000000..032bc557c0 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp @@ -0,0 +1,162 @@ +/* + * 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/engine/ObjectIdImpl.h" +#include + +using namespace std; +using namespace qmf::engine; +using qpid::framing::Buffer; + + +void AgentAttachment::setBanks(uint32_t broker, uint32_t agent) +{ + first = + ((uint64_t) (broker & 0x000fffff)) << 28 | + ((uint64_t) (agent & 0x0fffffff)); +} + +ObjectIdImpl::ObjectIdImpl(Buffer& buffer) : agent(0) +{ + decode(buffer); +} + +ObjectIdImpl::ObjectIdImpl(AgentAttachment* a, uint8_t flags, uint16_t seq, uint64_t object) : agent(a) +{ + first = + ((uint64_t) (flags & 0x0f)) << 60 | + ((uint64_t) (seq & 0x0fff)) << 48; + second = object; +} + +ObjectId* ObjectIdImpl::factory(Buffer& buffer) +{ + ObjectIdImpl* impl(new ObjectIdImpl(buffer)); + return new ObjectId(impl); +} + +ObjectId* ObjectIdImpl::factory(AgentAttachment* agent, uint8_t flags, uint16_t seq, uint64_t object) +{ + ObjectIdImpl* impl(new ObjectIdImpl(agent, flags, seq, object)); + return new ObjectId(impl); +} + +void ObjectIdImpl::decode(Buffer& buffer) +{ + first = buffer.getLongLong(); + second = buffer.getLongLong(); +} + +void ObjectIdImpl::encode(Buffer& buffer) const +{ + if (agent == 0) + buffer.putLongLong(first); + else + buffer.putLongLong(first | agent->first); + buffer.putLongLong(second); +} + +void ObjectIdImpl::fromString(const std::string& repr) +{ +#define FIELDS 5 +#if defined (_WIN32) && !defined (atoll) +# define atoll(X) _atoi64(X) +#endif + + std::string copy(repr.c_str()); + char* cText; + char* field[FIELDS]; + bool atFieldStart = true; + int idx = 0; + + cText = const_cast(copy.c_str()); + for (char* cursor = cText; *cursor; cursor++) { + if (atFieldStart) { + if (idx >= FIELDS) + return; // TODO error + field[idx++] = cursor; + atFieldStart = false; + } else { + if (*cursor == '-') { + *cursor = '\0'; + atFieldStart = true; + } + } + } + + if (idx != FIELDS) + return; // TODO error + + first = (atoll(field[0]) << 60) + + (atoll(field[1]) << 48) + + (atoll(field[2]) << 28) + + atoll(field[3]); + second = atoll(field[4]); + agent = 0; +} + +std::string ObjectIdImpl::asString() const +{ + stringstream val; + + val << (int) getFlags() << "-" << getSequence() << "-" << getBrokerBank() << "-" << + getAgentBank() << "-" << getObjectNum(); + return val.str(); +} + +bool ObjectIdImpl::operator==(const ObjectIdImpl& other) const +{ + uint64_t otherFirst = agent == 0 ? other.first : other.first & 0xffff000000000000LL; + + return first == otherFirst && second == other.second; +} + +bool ObjectIdImpl::operator<(const ObjectIdImpl& other) const +{ + uint64_t otherFirst = agent == 0 ? other.first : other.first & 0xffff000000000000LL; + + return (first < otherFirst) || ((first == otherFirst) && (second < other.second)); +} + +bool ObjectIdImpl::operator>(const ObjectIdImpl& other) const +{ + uint64_t otherFirst = agent == 0 ? other.first : other.first & 0xffff000000000000LL; + + return (first > otherFirst) || ((first == otherFirst) && (second > other.second)); +} + + +//================================================================== +// Wrappers +//================================================================== + +ObjectId::ObjectId() : impl(new ObjectIdImpl()) {} +ObjectId::ObjectId(const ObjectId& from) : impl(new ObjectIdImpl(*(from.impl))) {} +ObjectId::ObjectId(ObjectIdImpl* i) : impl(i) {} +ObjectId::~ObjectId() { delete impl; } +uint64_t ObjectId::getObjectNum() const { return impl->getObjectNum(); } +uint32_t ObjectId::getObjectNumHi() const { return impl->getObjectNumHi(); } +uint32_t ObjectId::getObjectNumLo() const { return impl->getObjectNumLo(); } +bool ObjectId::isDurable() const { return impl->isDurable(); } +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; } +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/engine/ObjectIdImpl.h b/qpid/cpp/src/qmf/engine/ObjectIdImpl.h new file mode 100644 index 0000000000..44fa8adffc --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.h @@ -0,0 +1,71 @@ +#ifndef _QmfEngineObjectIdImpl_ +#define _QmfEngineObjectIdImpl_ + +/* + * 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 +#include + +namespace qmf { +namespace engine { + + struct AgentAttachment { + uint64_t first; + + AgentAttachment() : first(0) {} + void setBanks(uint32_t broker, uint32_t bank); + uint64_t getFirst() const { return first; } + }; + + struct ObjectIdImpl { + AgentAttachment* agent; + uint64_t first; + uint64_t second; + + ObjectIdImpl() : agent(0), first(0), second(0) {} + ObjectIdImpl(qpid::framing::Buffer& buffer); + ObjectIdImpl(AgentAttachment* agent, uint8_t flags, uint16_t seq, uint64_t object); + + static ObjectId* factory(qpid::framing::Buffer& buffer); + static ObjectId* factory(AgentAttachment* agent, uint8_t flags, uint16_t seq, uint64_t object); + + void decode(qpid::framing::Buffer& buffer); + void encode(qpid::framing::Buffer& buffer) const; + void fromString(const std::string& repr); + std::string asString() const; + uint8_t getFlags() const { return (first & 0xF000000000000000LL) >> 60; } + uint16_t getSequence() const { return (first & 0x0FFF000000000000LL) >> 48; } + uint32_t getBrokerBank() const { return (first & 0x0000FFFFF0000000LL) >> 28; } + uint32_t getAgentBank() const { return first & 0x000000000FFFFFFFLL; } + uint64_t getObjectNum() const { return second; } + uint32_t getObjectNumHi() const { return (uint32_t) (second >> 32); } + uint32_t getObjectNumLo() const { return (uint32_t) (second & 0x00000000FFFFFFFFLL); } + bool isDurable() const { return getSequence() == 0; } + void setValue(uint64_t f, uint64_t s) { first = f; second = s; } + + bool operator==(const ObjectIdImpl& other) const; + bool operator<(const ObjectIdImpl& other) const; + bool operator>(const ObjectIdImpl& other) const; + }; +} +} + +#endif + diff --git a/qpid/cpp/src/qmf/engine/ObjectImpl.cpp b/qpid/cpp/src/qmf/engine/ObjectImpl.cpp new file mode 100644 index 0000000000..cae0e0da68 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ObjectImpl.cpp @@ -0,0 +1,232 @@ +/* + * 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/engine/ObjectImpl.h" +#include "qmf/engine/ValueImpl.h" +#include "qmf/engine/BrokerProxyImpl.h" +#include + +using namespace std; +using namespace qmf::engine; +using namespace qpid::sys; +using qpid::framing::Buffer; + +ObjectImpl::ObjectImpl(const SchemaObjectClass* type) : objectClass(type), broker(0), createTime(uint64_t(Duration(now()))), destroyTime(0), lastUpdatedTime(createTime) +{ + int propCount = objectClass->getPropertyCount(); + int statCount = objectClass->getStatisticCount(); + int idx; + + for (idx = 0; idx < propCount; idx++) { + const SchemaProperty* prop = objectClass->getProperty(idx); + properties[prop->getName()] = ValuePtr(new Value(prop->getType())); + } + + for (idx = 0; idx < statCount; idx++) { + const SchemaStatistic* stat = objectClass->getStatistic(idx); + statistics[stat->getName()] = ValuePtr(new Value(stat->getType())); + } +} + +ObjectImpl::ObjectImpl(const SchemaObjectClass* type, BrokerProxyImpl* b, Buffer& buffer, bool prop, bool stat, bool managed) : + objectClass(type), broker(b), createTime(0), destroyTime(0), lastUpdatedTime(0) +{ + int idx; + + if (managed) { + lastUpdatedTime = buffer.getLongLong(); + createTime = buffer.getLongLong(); + destroyTime = buffer.getLongLong(); + objectId.reset(ObjectIdImpl::factory(buffer)); + } + + if (prop) { + int propCount = objectClass->getPropertyCount(); + set excludes; + parsePresenceMasks(buffer, excludes); + for (idx = 0; idx < propCount; idx++) { + const SchemaProperty* prop = objectClass->getProperty(idx); + if (excludes.count(prop->getName()) != 0) { + properties[prop->getName()] = ValuePtr(new Value(prop->getType())); + } else { + Value* pval = ValueImpl::factory(prop->getType(), buffer); + properties[prop->getName()] = ValuePtr(pval); + } + } + } + + if (stat) { + int statCount = objectClass->getStatisticCount(); + for (idx = 0; idx < statCount; idx++) { + const SchemaStatistic* stat = objectClass->getStatistic(idx); + Value* sval = ValueImpl::factory(stat->getType(), buffer); + statistics[stat->getName()] = ValuePtr(sval); + } + } +} + +Object* ObjectImpl::factory(const SchemaObjectClass* type, BrokerProxyImpl* b, Buffer& buffer, bool prop, bool stat, bool managed) +{ + ObjectImpl* impl(new ObjectImpl(type, b, buffer, prop, stat, managed)); + return new Object(impl); +} + +ObjectImpl::~ObjectImpl() +{ +} + +void ObjectImpl::destroy() +{ + destroyTime = uint64_t(Duration(now())); + // TODO - flag deletion +} + +Value* ObjectImpl::getValue(const string& key) const +{ + map::const_iterator iter; + + iter = properties.find(key); + if (iter != properties.end()) + return iter->second.get(); + + iter = statistics.find(key); + if (iter != statistics.end()) + return iter->second.get(); + + return 0; +} + +void ObjectImpl::invokeMethod(const string& methodName, const Value* inArgs, void* context) const +{ + if (broker != 0 && objectId.get() != 0) + broker->sendMethodRequest(objectId.get(), objectClass, methodName, inArgs, context); +} + +void ObjectImpl::merge(const Object& from) +{ + for (map::const_iterator piter = from.impl->properties.begin(); + piter != from.impl->properties.end(); piter++) + properties[piter->first] = piter->second; + for (map::const_iterator siter = from.impl->statistics.begin(); + siter != from.impl->statistics.end(); siter++) + statistics[siter->first] = siter->second; +} + +void ObjectImpl::parsePresenceMasks(Buffer& buffer, set& excludeList) +{ + int propCount = objectClass->getPropertyCount(); + excludeList.clear(); + uint8_t bit = 0; + uint8_t mask = 0; + + for (int idx = 0; idx < propCount; idx++) { + const SchemaProperty* prop = objectClass->getProperty(idx); + if (prop->isOptional()) { + if (bit == 0) { + mask = buffer.getOctet(); + bit = 1; + } + if ((mask & bit) == 0) + excludeList.insert(string(prop->getName())); + if (bit == 0x80) + bit = 0; + else + bit = bit << 1; + } + } +} + +void ObjectImpl::encodeSchemaKey(qpid::framing::Buffer& buffer) const +{ + buffer.putShortString(objectClass->getClassKey()->getPackageName()); + buffer.putShortString(objectClass->getClassKey()->getClassName()); + buffer.putBin128(const_cast(objectClass->getClassKey()->getHash())); +} + +void ObjectImpl::encodeManagedObjectData(qpid::framing::Buffer& buffer) const +{ + buffer.putLongLong(lastUpdatedTime); + buffer.putLongLong(createTime); + buffer.putLongLong(destroyTime); + objectId->impl->encode(buffer); +} + +void ObjectImpl::encodeProperties(qpid::framing::Buffer& buffer) const +{ + int propCount = objectClass->getPropertyCount(); + uint8_t bit = 0; + uint8_t mask = 0; + ValuePtr value; + + for (int idx = 0; idx < propCount; idx++) { + const SchemaProperty* prop = objectClass->getProperty(idx); + if (prop->isOptional()) { + value = properties[prop->getName()]; + if (bit == 0) + bit = 1; + if (!value->isNull()) + mask |= bit; + if (bit == 0x80) { + buffer.putOctet(mask); + bit = 0; + mask = 0; + } else + bit = bit << 1; + } + } + if (bit != 0) { + buffer.putOctet(mask); + } + + for (int idx = 0; idx < propCount; idx++) { + const SchemaProperty* prop = objectClass->getProperty(idx); + value = properties[prop->getName()]; + if (!prop->isOptional() || !value->isNull()) { + value->impl->encode(buffer); + } + } +} + +void ObjectImpl::encodeStatistics(qpid::framing::Buffer& buffer) const +{ + int statCount = objectClass->getStatisticCount(); + for (int idx = 0; idx < statCount; idx++) { + const SchemaStatistic* stat = objectClass->getStatistic(idx); + ValuePtr value = statistics[stat->getName()]; + value->impl->encode(buffer); + } +} + +//================================================================== +// Wrappers +//================================================================== + +Object::Object(const SchemaObjectClass* type) : impl(new ObjectImpl(type)) {} +Object::Object(ObjectImpl* i) : impl(i) {} +Object::Object(const Object& from) : impl(new ObjectImpl(*(from.impl))) {} +Object::~Object() { delete impl; } +void Object::destroy() { impl->destroy(); } +const ObjectId* Object::getObjectId() const { return impl->getObjectId(); } +void Object::setObjectId(ObjectId* oid) { impl->setObjectId(oid); } +const SchemaObjectClass* Object::getClass() const { return impl->getClass(); } +Value* Object::getValue(const char* key) const { return impl->getValue(key); } +void Object::invokeMethod(const char* m, const Value* a, void* c) const { impl->invokeMethod(m, a, c); } +bool Object::isDeleted() const { return impl->isDeleted(); } +void Object::merge(const Object& from) { impl->merge(from); } + diff --git a/qpid/cpp/src/qmf/engine/ObjectImpl.h b/qpid/cpp/src/qmf/engine/ObjectImpl.h new file mode 100644 index 0000000000..ddd20bfea2 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ObjectImpl.h @@ -0,0 +1,76 @@ +#ifndef _QmfEngineObjectImpl_ +#define _QmfEngineObjectImpl_ + +/* + * 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 +#include +#include +#include +#include +#include +#include +#include + +namespace qmf { +namespace engine { + + class BrokerProxyImpl; + + typedef boost::shared_ptr ObjectPtr; + + struct ObjectImpl { + typedef boost::shared_ptr ValuePtr; + const SchemaObjectClass* objectClass; + BrokerProxyImpl* broker; + boost::shared_ptr objectId; + uint64_t createTime; + uint64_t destroyTime; + uint64_t lastUpdatedTime; + mutable std::map properties; + mutable std::map statistics; + + ObjectImpl(const SchemaObjectClass* type); + ObjectImpl(const SchemaObjectClass* type, BrokerProxyImpl* b, qpid::framing::Buffer& buffer, + bool prop, bool stat, bool managed); + static Object* factory(const SchemaObjectClass* type, BrokerProxyImpl* b, qpid::framing::Buffer& buffer, + bool prop, bool stat, bool managed); + ~ObjectImpl(); + + void destroy(); + const ObjectId* getObjectId() const { return objectId.get(); } + void setObjectId(ObjectId* oid) { objectId.reset(oid); } + const SchemaObjectClass* getClass() const { return objectClass; } + Value* getValue(const std::string& key) const; + void invokeMethod(const std::string& methodName, const Value* inArgs, void* context) const; + bool isDeleted() const { return destroyTime != 0; } + void merge(const Object& from); + + void parsePresenceMasks(qpid::framing::Buffer& buffer, std::set& excludeList); + void encodeSchemaKey(qpid::framing::Buffer& buffer) const; + void encodeManagedObjectData(qpid::framing::Buffer& buffer) const; + void encodeProperties(qpid::framing::Buffer& buffer) const; + void encodeStatistics(qpid::framing::Buffer& buffer) const; + }; +} +} + +#endif + diff --git a/qpid/cpp/src/qmf/engine/Protocol.cpp b/qpid/cpp/src/qmf/engine/Protocol.cpp new file mode 100644 index 0000000000..6061b70a8d --- /dev/null +++ b/qpid/cpp/src/qmf/engine/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/engine/Protocol.h" +#include "qpid/framing/Buffer.h" + +using namespace std; +using namespace qmf::engine; +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/engine/Protocol.h b/qpid/cpp/src/qmf/engine/Protocol.h new file mode 100644 index 0000000000..1cdfa60c84 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/Protocol.h @@ -0,0 +1,69 @@ +#ifndef _QmfEngineProtocol_ +#define _QmfEngineProtocol_ + +/* + * 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 + +namespace qpid { + namespace framing { + class Buffer; + } +} + +namespace qmf { +namespace engine { + + 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/engine/QueryImpl.cpp b/qpid/cpp/src/qmf/engine/QueryImpl.cpp new file mode 100644 index 0000000000..6f2beeee87 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/QueryImpl.cpp @@ -0,0 +1,103 @@ +/* + * 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/engine/QueryImpl.h" +#include "qmf/engine/ObjectIdImpl.h" +#include "qpid/framing/Buffer.h" +#include "qpid/framing/FieldTable.h" + +using namespace std; +using namespace qmf::engine; +using namespace qpid::framing; + +bool QueryElementImpl::evaluate(const Object* /*object*/) const +{ + // TODO: Implement this + return false; +} + +bool QueryExpressionImpl::evaluate(const Object* /*object*/) const +{ + // TODO: Implement this + return false; +} + +QueryImpl::QueryImpl(Buffer& buffer) +{ + FieldTable ft; + ft.decode(buffer); + // TODO +} + +Query* QueryImpl::factory(Buffer& buffer) +{ + QueryImpl* impl(new QueryImpl(buffer)); + return new Query(impl); +} + +void QueryImpl::encode(Buffer& buffer) const +{ + FieldTable ft; + + if (oid.get() != 0) { + ft.setString("_objectid", oid->impl->asString()); + } else { + if (!packageName.empty()) + ft.setString("_package", packageName); + ft.setString("_class", className); + } + + ft.encode(buffer); +} + + +//================================================================== +// Wrappers +//================================================================== + +QueryElement::QueryElement(const char* attrName, const Value* value, ValueOper oper) : impl(new QueryElementImpl(attrName, value, oper)) {} +QueryElement::QueryElement(QueryElementImpl* i) : impl(i) {} +QueryElement::~QueryElement() { delete impl; } +bool QueryElement::evaluate(const Object* object) const { return impl->evaluate(object); } + +QueryExpression::QueryExpression(ExprOper oper, const QueryOperand* operand1, const QueryOperand* operand2) : impl(new QueryExpressionImpl(oper, operand1, operand2)) {} +QueryExpression::QueryExpression(QueryExpressionImpl* i) : impl(i) {} +QueryExpression::~QueryExpression() { delete impl; } +bool QueryExpression::evaluate(const Object* object) const { return impl->evaluate(object); } + +Query::Query(const char* className, const char* packageName) : impl(new QueryImpl(className, packageName)) {} +Query::Query(const SchemaClassKey* key) : impl(new QueryImpl(key)) {} +Query::Query(const ObjectId* oid) : impl(new QueryImpl(oid)) {} +Query::Query(QueryImpl* i) : impl(i) {} +Query::Query(const Query& from) : impl(new QueryImpl(*(from.impl))) {} +Query::~Query() { delete impl; } +void Query::setSelect(const QueryOperand* criterion) { impl->setSelect(criterion); } +void Query::setLimit(uint32_t maxResults) { impl->setLimit(maxResults); } +void Query::setOrderBy(const char* attrName, bool decreasing) { impl->setOrderBy(attrName, decreasing); } +const char* Query::getPackage() const { return impl->getPackage().c_str(); } +const char* Query::getClass() const { return impl->getClass().c_str(); } +const ObjectId* Query::getObjectId() const { return impl->getObjectId(); } +bool Query::haveSelect() const { return impl->haveSelect(); } +bool Query::haveLimit() const { return impl->haveLimit(); } +bool Query::haveOrderBy() const { return impl->haveOrderBy(); } +const QueryOperand* Query::getSelect() const { return impl->getSelect(); } +uint32_t Query::getLimit() const { return impl->getLimit(); } +const char* Query::getOrderBy() const { return impl->getOrderBy().c_str(); } +bool Query::getDecreasing() const { return impl->getDecreasing(); } + diff --git a/qpid/cpp/src/qmf/engine/QueryImpl.h b/qpid/cpp/src/qmf/engine/QueryImpl.h new file mode 100644 index 0000000000..2c64c6739c --- /dev/null +++ b/qpid/cpp/src/qmf/engine/QueryImpl.h @@ -0,0 +1,100 @@ +#ifndef _QmfEngineQueryImpl_ +#define _QmfEngineQueryImpl_ + +/* + * 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/engine/Query.h" +#include "qmf/engine/Schema.h" +#include +#include + +namespace qpid { + namespace framing { + class Buffer; + } +} + +namespace qmf { +namespace engine { + + struct QueryElementImpl { + QueryElementImpl(const std::string& a, const Value* v, ValueOper o) : attrName(a), value(v), oper(o) {} + ~QueryElementImpl() {} + bool evaluate(const Object* object) const; + + std::string attrName; + const Value* value; + ValueOper oper; + }; + + struct QueryExpressionImpl { + QueryExpressionImpl(ExprOper o, const QueryOperand* operand1, const QueryOperand* operand2) : oper(o), left(operand1), right(operand2) {} + ~QueryExpressionImpl() {} + bool evaluate(const Object* object) const; + + ExprOper oper; + const QueryOperand* left; + const QueryOperand* right; + }; + + struct QueryImpl { + // Constructors mapped to public + QueryImpl(const std::string& c, const std::string& p) : packageName(p), className(c), select(0), resultLimit(0) {} + QueryImpl(const SchemaClassKey* key) : packageName(key->getPackageName()), className(key->getClassName()), select(0), resultLimit(0) {} + QueryImpl(const ObjectId* oid) : oid(new ObjectId(*oid)), select(0), resultLimit(0) {} + + // Factory constructors + QueryImpl(qpid::framing::Buffer& buffer); + + ~QueryImpl() {}; + static Query* factory(qpid::framing::Buffer& buffer); + + void setSelect(const QueryOperand* criterion) { select = criterion; } + void setLimit(uint32_t maxResults) { resultLimit = maxResults; } + void setOrderBy(const std::string& attrName, bool decreasing) { + orderBy = attrName; orderDecreasing = decreasing; + } + + const std::string& getPackage() const { return packageName; } + const std::string& getClass() const { return className; } + const ObjectId* getObjectId() const { return oid.get(); } + + bool haveSelect() const { return select != 0; } + bool haveLimit() const { return resultLimit > 0; } + bool haveOrderBy() const { return !orderBy.empty(); } + const QueryOperand* getSelect() const { return select; } + uint32_t getLimit() const { return resultLimit; } + const std::string& getOrderBy() const { return orderBy; } + bool getDecreasing() const { return orderDecreasing; } + + void encode(qpid::framing::Buffer& buffer) const; + + std::string packageName; + std::string className; + boost::shared_ptr oid; + const QueryOperand* select; + uint32_t resultLimit; + std::string orderBy; + bool orderDecreasing; + }; +} +} + +#endif diff --git a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp new file mode 100644 index 0000000000..709cfd1236 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp @@ -0,0 +1,484 @@ +/* + * 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/engine/ResilientConnection.h" +#include "qmf/engine/MessageImpl.h" +#include "qmf/engine/ConnectionSettingsImpl.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std; +using namespace qmf::engine; +using namespace qpid; +using qpid::sys::Mutex; + +namespace qmf { +namespace engine { + struct ResilientConnectionEventImpl { + ResilientConnectionEvent::EventKind kind; + void* sessionContext; + string errorText; + MessageImpl message; + + ResilientConnectionEventImpl(ResilientConnectionEvent::EventKind k, + const MessageImpl& m = MessageImpl()) : + kind(k), sessionContext(0), message(m) {} + ResilientConnectionEvent copy(); + }; + + struct RCSession : public client::MessageListener, public qpid::sys::Runnable, public qpid::RefCounted { + typedef boost::intrusive_ptr Ptr; + ResilientConnectionImpl& connImpl; + string name; + client::Connection& connection; + client::Session session; + client::SubscriptionManager* subscriptions; + string userId; + void* userContext; + vector dests; + qpid::sys::Thread thread; + + RCSession(ResilientConnectionImpl& ci, const string& n, client::Connection& c, void* uc); + ~RCSession(); + void received(client::Message& msg); + void run(); + void stop(); + }; + + class ResilientConnectionImpl : public qpid::sys::Runnable, public boost::noncopyable { + public: + ResilientConnectionImpl(const ConnectionSettings& settings); + ~ResilientConnectionImpl(); + + bool isConnected() const; + bool getEvent(ResilientConnectionEvent& event); + void popEvent(); + bool createSession(const char* name, void* sessionContext, SessionHandle& handle); + void destroySession(SessionHandle handle); + void sendMessage(SessionHandle handle, qmf::engine::Message& message); + void declareQueue(SessionHandle handle, char* queue); + void deleteQueue(SessionHandle handle, char* queue); + void bind(SessionHandle handle, char* exchange, char* queue, char* key); + void unbind(SessionHandle handle, char* exchange, char* queue, char* key); + void setNotifyFd(int fd); + + void run(); + void failure(); + void sessionClosed(RCSession* sess); + + void EnqueueEvent(ResilientConnectionEvent::EventKind kind, + void* sessionContext = 0, + const MessageImpl& message = MessageImpl(), + const string& errorText = ""); + + private: + int notifyFd; + bool connected; + bool shutdown; + string lastError; + const ConnectionSettings settings; + client::Connection connection; + mutable qpid::sys::Mutex lock; + int delayMin; + int delayMax; + int delayFactor; + qpid::sys::Condition cond; + qpid::sys::Thread connThread; + deque eventQueue; + set sessions; + }; +} +} + +ResilientConnectionEvent ResilientConnectionEventImpl::copy() +{ + ResilientConnectionEvent item; + + ::memset(&item, 0, sizeof(ResilientConnectionEvent)); + item.kind = kind; + item.sessionContext = sessionContext; + item.message = message.copy(); + item.errorText = const_cast(errorText.c_str()); + + return item; +} + +RCSession::RCSession(ResilientConnectionImpl& ci, const string& n, client::Connection& c, void* uc) : + connImpl(ci), name(n), connection(c), session(connection.newSession(name)), + subscriptions(new client::SubscriptionManager(session)), userContext(uc), thread(*this) +{ + const qpid::client::ConnectionSettings& operSettings = connection.getNegotiatedSettings(); + userId = operSettings.username; +} + +RCSession::~RCSession() +{ + subscriptions->stop(); + thread.join(); + session.close(); + delete subscriptions; +} + +void RCSession::run() +{ + try { + subscriptions->run(); + } catch (exception& /*e*/) { + connImpl.sessionClosed(this); + } +} + +void RCSession::stop() +{ + subscriptions->stop(); +} + +void RCSession::received(client::Message& msg) +{ + MessageImpl qmsg; + qmsg.body = msg.getData(); + + qpid::framing::MessageProperties p = msg.getMessageProperties(); + if (p.hasReplyTo()) { + const qpid::framing::ReplyTo& rt = p.getReplyTo(); + qmsg.replyExchange = rt.getExchange(); + qmsg.replyKey = rt.getRoutingKey(); + } + + if (p.hasUserId()) { + qmsg.userId = p.getUserId(); + } + + connImpl.EnqueueEvent(ResilientConnectionEvent::RECV, userContext, qmsg); +} + +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() +{ + shutdown = true; + connected = false; + cond.notify(); + connThread.join(); + connection.close(); +} + +bool ResilientConnectionImpl::isConnected() const +{ + Mutex::ScopedLock _lock(lock); + return connected; +} + +bool ResilientConnectionImpl::getEvent(ResilientConnectionEvent& event) +{ + Mutex::ScopedLock _lock(lock); + if (eventQueue.empty()) + return false; + event = eventQueue.front().copy(); + return true; +} + +void ResilientConnectionImpl::popEvent() +{ + Mutex::ScopedLock _lock(lock); + if (!eventQueue.empty()) + eventQueue.pop_front(); +} + +bool ResilientConnectionImpl::createSession(const char* name, void* sessionContext, + SessionHandle& handle) +{ + Mutex::ScopedLock _lock(lock); + if (!connected) + return false; + + RCSession::Ptr sess = RCSession::Ptr(new RCSession(*this, name, connection, sessionContext)); + + handle.impl = (void*) sess.get(); + sessions.insert(sess); + + return true; +} + +void ResilientConnectionImpl::destroySession(SessionHandle handle) +{ + Mutex::ScopedLock _lock(lock); + RCSession::Ptr sess = RCSession::Ptr((RCSession*) handle.impl); + set::iterator iter = sessions.find(sess); + if (iter != sessions.end()) { + for (vector::iterator dIter = sess->dests.begin(); dIter != sess->dests.end(); dIter++) + sess->subscriptions->cancel(dIter->c_str()); + sess->subscriptions->stop(); + sess->subscriptions->wait(); + + sessions.erase(iter); + return; + } +} + +void ResilientConnectionImpl::sendMessage(SessionHandle handle, qmf::engine::Message& message) +{ + Mutex::ScopedLock _lock(lock); + RCSession::Ptr sess = RCSession::Ptr((RCSession*) handle.impl); + set::iterator iter = sessions.find(sess); + qpid::client::Message msg; + string data(message.body, message.length); + msg.getDeliveryProperties().setRoutingKey(message.routingKey); + msg.getMessageProperties().setReplyTo(qpid::framing::ReplyTo(message.replyExchange, message.replyKey)); + if (settings.impl->getSendUserId()) + msg.getMessageProperties().setUserId(sess->userId); + msg.setData(data); + + try { + 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); + EnqueueEvent(ResilientConnectionEvent::SESSION_CLOSED, (*iter)->userContext); + } +} + +void ResilientConnectionImpl::declareQueue(SessionHandle handle, char* queue) +{ + Mutex::ScopedLock _lock(lock); + RCSession* sess = (RCSession*) handle.impl; + + 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.impl; + + sess->session.queueDelete(client::arg::queue=queue); + for (vector::iterator iter = sess->dests.begin(); + iter != sess->dests.end(); iter++) + if (*iter == queue) { + sess->subscriptions->cancel(queue); + sess->dests.erase(iter); + break; + } +} + +void ResilientConnectionImpl::bind(SessionHandle handle, + char* exchange, char* queue, char* key) +{ + Mutex::ScopedLock _lock(lock); + RCSession* sess = (RCSession*) handle.impl; + + 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.impl; + + sess->session.exchangeUnbind(client::arg::exchange=exchange, client::arg::queue=queue, client::arg::bindingKey=key); +} + +void ResilientConnectionImpl::setNotifyFd(int fd) +{ + notifyFd = fd; +} + +void ResilientConnectionImpl::run() +{ + int delay(delayMin); + + while (true) { + try { + QPID_LOG(trace, "Trying to open connection..."); + connection.open(settings.impl->getClientSettings()); + { + Mutex::ScopedLock _lock(lock); + connected = true; + EnqueueEvent(ResilientConnectionEvent::CONNECTED); + + while (connected) + cond.wait(lock); + delay = delayMin; + + while (!sessions.empty()) { + set::iterator iter = sessions.begin(); + RCSession::Ptr sess = *iter; + sessions.erase(iter); + 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); + + if (shutdown) + return; + } + connection.close(); + } catch (exception &e) { + QPID_LOG(debug, "connection.open exception: " << e.what()); + Mutex::ScopedLock _lock(lock); + lastError = e.what(); + if (delay < delayMax) + delay *= delayFactor; + } + + ::qpid::sys::sleep(delay); + } +} + +void ResilientConnectionImpl::failure() +{ + Mutex::ScopedLock _lock(lock); + + connected = false; + lastError = "Closed by Peer"; + cond.notify(); +} + +void ResilientConnectionImpl::sessionClosed(RCSession*) +{ + Mutex::ScopedLock _lock(lock); + connected = false; + lastError = "Closed due to Session failure"; + cond.notify(); +} + +void ResilientConnectionImpl::EnqueueEvent(ResilientConnectionEvent::EventKind kind, + void* sessionContext, + const MessageImpl& message, + const string& errorText) +{ + Mutex::ScopedLock _lock(lock); + ResilientConnectionEventImpl event(kind, message); + + event.sessionContext = sessionContext; + event.errorText = errorText; + + eventQueue.push_back(event); + if (notifyFd != -1) + { + int unused_ret; //Suppress warnings about ignoring return value. + unused_ret = ::write(notifyFd, ".", 1); + } +} + + +//================================================================== +// Wrappers +//================================================================== + +ResilientConnection::ResilientConnection(const ConnectionSettings& settings) +{ + impl = new ResilientConnectionImpl(settings); +} + +ResilientConnection::~ResilientConnection() +{ + delete impl; +} + +bool ResilientConnection::isConnected() const +{ + return impl->isConnected(); +} + +bool ResilientConnection::getEvent(ResilientConnectionEvent& event) +{ + return impl->getEvent(event); +} + +void ResilientConnection::popEvent() +{ + impl->popEvent(); +} + +bool ResilientConnection::createSession(const char* name, void* sessionContext, SessionHandle& handle) +{ + return impl->createSession(name, sessionContext, handle); +} + +void ResilientConnection::destroySession(SessionHandle handle) +{ + impl->destroySession(handle); +} + +void ResilientConnection::sendMessage(SessionHandle handle, qmf::engine::Message& message) +{ + impl->sendMessage(handle, message); +} + +void ResilientConnection::declareQueue(SessionHandle handle, char* queue) +{ + impl->declareQueue(handle, queue); +} + +void ResilientConnection::deleteQueue(SessionHandle handle, char* queue) +{ + impl->deleteQueue(handle, queue); +} + +void ResilientConnection::bind(SessionHandle handle, char* exchange, char* queue, char* key) +{ + impl->bind(handle, exchange, queue, key); +} + +void ResilientConnection::unbind(SessionHandle handle, char* exchange, char* queue, char* key) +{ + impl->unbind(handle, exchange, queue, key); +} + +void ResilientConnection::setNotifyFd(int fd) +{ + impl->setNotifyFd(fd); +} + diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp new file mode 100644 index 0000000000..fb09980680 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp @@ -0,0 +1,609 @@ +/* + * 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/engine/SchemaImpl.h" +#include +#include +#include +#include +#include +#include + +using namespace std; +using namespace qmf::engine; +using qpid::framing::Buffer; +using qpid::framing::FieldTable; +using qpid::framing::Uuid; + +SchemaHash::SchemaHash() +{ + for (int idx = 0; idx < 16; idx++) + hash[idx] = 0x5A; +} + +void SchemaHash::encode(Buffer& buffer) const +{ + buffer.putBin128(hash); +} + +void SchemaHash::decode(Buffer& buffer) +{ + buffer.getBin128(hash); +} + +void SchemaHash::update(uint8_t data) +{ + update((char*) &data, 1); +} + +void SchemaHash::update(const char* data, uint32_t len) +{ + uint64_t* first = (uint64_t*) hash; + uint64_t* second = (uint64_t*) hash + 1; + + for (uint32_t idx = 0; idx < len; idx++) { + *first = *first ^ (uint64_t) data[idx]; + *second = *second << 1; + *second |= ((*first & 0x8000000000000000LL) >> 63); + *first = *first << 1; + *first = *first ^ *second; + } +} + +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) +{ + FieldTable map; + map.decode(buffer); + + name = map.getAsString("name"); + typecode = (Typecode) map.getAsInt("type"); + unit = map.getAsString("unit"); + description = map.getAsString("desc"); + + dir = DIR_IN; + string dstr(map.getAsString("dir")); + if (dstr == "O") + dir = DIR_OUT; + else if (dstr == "IO") + dir = DIR_IN_OUT; +} + +SchemaArgument* SchemaArgumentImpl::factory(Buffer& buffer) +{ + SchemaArgumentImpl* impl(new SchemaArgumentImpl(buffer)); + return new SchemaArgument(impl); +} + +void SchemaArgumentImpl::encode(Buffer& buffer) const +{ + FieldTable map; + + map.setString("name", name); + map.setInt("type", (int) typecode); + if (dir == DIR_IN) + map.setString("dir", "I"); + else if (dir == DIR_OUT) + map.setString("dir", "O"); + else + map.setString("dir", "IO"); + if (!unit.empty()) + map.setString("unit", unit); + if (!description.empty()) + map.setString("desc", description); + + map.encode(buffer); +} + +void SchemaArgumentImpl::updateHash(SchemaHash& hash) const +{ + hash.update(name); + hash.update(typecode); + hash.update(dir); + hash.update(unit); + hash.update(description); +} + +SchemaMethodImpl::SchemaMethodImpl(Buffer& buffer) +{ + FieldTable map; + int argCount; + + map.decode(buffer); + name = map.getAsString("name"); + argCount = map.getAsInt("argCount"); + description = map.getAsString("desc"); + + for (int idx = 0; idx < argCount; idx++) { + SchemaArgument* arg = SchemaArgumentImpl::factory(buffer); + addArgument(arg); + } +} + +SchemaMethod* SchemaMethodImpl::factory(Buffer& buffer) +{ + SchemaMethodImpl* impl(new SchemaMethodImpl(buffer)); + return new SchemaMethod(impl); +} + +void SchemaMethodImpl::encode(Buffer& buffer) const +{ + FieldTable map; + + map.setString("name", name); + map.setInt("argCount", arguments.size()); + if (!description.empty()) + map.setString("desc", description); + map.encode(buffer); + + for (vector::const_iterator iter = arguments.begin(); + iter != arguments.end(); iter++) + (*iter)->impl->encode(buffer); +} + +void SchemaMethodImpl::addArgument(const SchemaArgument* argument) +{ + arguments.push_back(argument); +} + +const SchemaArgument* SchemaMethodImpl::getArgument(int idx) const +{ + int count = 0; + for (vector::const_iterator iter = arguments.begin(); + iter != arguments.end(); iter++, count++) + if (idx == count) + return (*iter); + return 0; +} + +void SchemaMethodImpl::updateHash(SchemaHash& hash) const +{ + hash.update(name); + hash.update(description); + for (vector::const_iterator iter = arguments.begin(); + iter != arguments.end(); iter++) + (*iter)->impl->updateHash(hash); +} + +SchemaPropertyImpl::SchemaPropertyImpl(Buffer& buffer) +{ + FieldTable map; + map.decode(buffer); + + name = map.getAsString("name"); + typecode = (Typecode) map.getAsInt("type"); + access = (Access) map.getAsInt("access"); + index = map.getAsInt("index") != 0; + optional = map.getAsInt("optional") != 0; + unit = map.getAsString("unit"); + description = map.getAsString("desc"); +} + +SchemaProperty* SchemaPropertyImpl::factory(Buffer& buffer) +{ + SchemaPropertyImpl* impl(new SchemaPropertyImpl(buffer)); + return new SchemaProperty(impl); +} + +void SchemaPropertyImpl::encode(Buffer& buffer) const +{ + FieldTable map; + + map.setString("name", name); + map.setInt("type", (int) typecode); + map.setInt("access", (int) access); + map.setInt("index", index ? 1 : 0); + map.setInt("optional", optional ? 1 : 0); + if (!unit.empty()) + map.setString("unit", unit); + if (!description.empty()) + map.setString("desc", description); + + map.encode(buffer); +} + +void SchemaPropertyImpl::updateHash(SchemaHash& hash) const +{ + hash.update(name); + hash.update(typecode); + hash.update(access); + hash.update(index); + hash.update(optional); + hash.update(unit); + hash.update(description); +} + +SchemaStatisticImpl::SchemaStatisticImpl(Buffer& buffer) +{ + FieldTable map; + map.decode(buffer); + + name = map.getAsString("name"); + typecode = (Typecode) map.getAsInt("type"); + unit = map.getAsString("unit"); + description = map.getAsString("desc"); +} + +SchemaStatistic* SchemaStatisticImpl::factory(Buffer& buffer) +{ + SchemaStatisticImpl* impl(new SchemaStatisticImpl(buffer)); + return new SchemaStatistic(impl); +} + +void SchemaStatisticImpl::encode(Buffer& buffer) const +{ + FieldTable map; + + map.setString("name", name); + map.setInt("type", (int) typecode); + if (!unit.empty()) + map.setString("unit", unit); + if (!description.empty()) + map.setString("desc", description); + + map.encode(buffer); +} + +void SchemaStatisticImpl::updateHash(SchemaHash& hash) const +{ + hash.update(name); + hash.update(typecode); + hash.update(unit); + hash.update(description); +} + +SchemaClassKeyImpl::SchemaClassKeyImpl(const string& p, const string& n, const SchemaHash& h) : package(p), name(n), hash(h) {} + +SchemaClassKeyImpl::SchemaClassKeyImpl(Buffer& buffer) : package(packageContainer), name(nameContainer), hash(hashContainer) +{ + buffer.getShortString(packageContainer); + buffer.getShortString(nameContainer); + hashContainer.decode(buffer); +} + +SchemaClassKey* SchemaClassKeyImpl::factory(const string& package, const string& name, const SchemaHash& hash) +{ + SchemaClassKeyImpl* impl(new SchemaClassKeyImpl(package, name, hash)); + return new SchemaClassKey(impl); +} + +SchemaClassKey* SchemaClassKeyImpl::factory(Buffer& buffer) +{ + SchemaClassKeyImpl* impl(new SchemaClassKeyImpl(buffer)); + return new SchemaClassKey(impl); +} + +void SchemaClassKeyImpl::encode(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) : hasHash(true), classKey(SchemaClassKeyImpl::factory(package, name, hash)) +{ + buffer.getShortString(package); + buffer.getShortString(name); + hash.decode(buffer); + + /*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++) { + const SchemaProperty* property = SchemaPropertyImpl::factory(buffer); + addProperty(property); + } + + for (uint16_t idx = 0; idx < statCount; idx++) { + const SchemaStatistic* statistic = SchemaStatisticImpl::factory(buffer); + addStatistic(statistic); + } + + for (uint16_t idx = 0; idx < methodCount; idx++) { + SchemaMethod* method = SchemaMethodImpl::factory(buffer); + addMethod(method); + } +} + +SchemaObjectClass* SchemaObjectClassImpl::factory(Buffer& buffer) +{ + SchemaObjectClassImpl* impl(new SchemaObjectClassImpl(buffer)); + return new SchemaObjectClass(impl); +} + +void SchemaObjectClassImpl::encode(Buffer& buffer) const +{ + buffer.putOctet((uint8_t) CLASS_OBJECT); + buffer.putShortString(package); + buffer.putShortString(name); + hash.encode(buffer); + buffer.putOctet(0); // No parent class + buffer.putShort((uint16_t) properties.size()); + buffer.putShort((uint16_t) statistics.size()); + buffer.putShort((uint16_t) methods.size()); + + for (vector::const_iterator iter = properties.begin(); + iter != properties.end(); iter++) + (*iter)->impl->encode(buffer); + for (vector::const_iterator iter = statistics.begin(); + iter != statistics.end(); iter++) + (*iter)->impl->encode(buffer); + for (vector::const_iterator iter = methods.begin(); + iter != methods.end(); iter++) + (*iter)->impl->encode(buffer); +} + +const SchemaClassKey* SchemaObjectClassImpl::getClassKey() const +{ + if (!hasHash) { + hasHash = true; + hash.update(package); + hash.update(name); + for (vector::const_iterator iter = properties.begin(); + iter != properties.end(); iter++) + (*iter)->impl->updateHash(hash); + for (vector::const_iterator iter = statistics.begin(); + iter != statistics.end(); iter++) + (*iter)->impl->updateHash(hash); + for (vector::const_iterator iter = methods.begin(); + iter != methods.end(); iter++) + (*iter)->impl->updateHash(hash); + } + + return classKey.get(); +} + +void SchemaObjectClassImpl::addProperty(const SchemaProperty* property) +{ + properties.push_back(property); +} + +void SchemaObjectClassImpl::addStatistic(const SchemaStatistic* statistic) +{ + statistics.push_back(statistic); +} + +void SchemaObjectClassImpl::addMethod(const SchemaMethod* method) +{ + methods.push_back(method); +} + +const SchemaProperty* SchemaObjectClassImpl::getProperty(int idx) const +{ + int count = 0; + for (vector::const_iterator iter = properties.begin(); + iter != properties.end(); iter++, count++) + if (idx == count) + return *iter; + return 0; +} + +const SchemaStatistic* SchemaObjectClassImpl::getStatistic(int idx) const +{ + int count = 0; + for (vector::const_iterator iter = statistics.begin(); + iter != statistics.end(); iter++, count++) + if (idx == count) + return *iter; + return 0; +} + +const SchemaMethod* SchemaObjectClassImpl::getMethod(int idx) const +{ + int count = 0; + for (vector::const_iterator iter = methods.begin(); + iter != methods.end(); iter++, count++) + if (idx == count) + return *iter; + return 0; +} + +SchemaEventClassImpl::SchemaEventClassImpl(Buffer& buffer) : hasHash(true), classKey(SchemaClassKeyImpl::factory(package, name, hash)) +{ + buffer.getShortString(package); + buffer.getShortString(name); + hash.decode(buffer); + buffer.putOctet(0); // No parent class + + uint16_t argCount = buffer.getShort(); + + for (uint16_t idx = 0; idx < argCount; idx++) { + SchemaArgument* argument = SchemaArgumentImpl::factory(buffer); + addArgument(argument); + } +} + +SchemaEventClass* SchemaEventClassImpl::factory(Buffer& buffer) +{ + SchemaEventClassImpl* impl(new SchemaEventClassImpl(buffer)); + return new SchemaEventClass(impl); +} + +void SchemaEventClassImpl::encode(Buffer& buffer) const +{ + buffer.putOctet((uint8_t) CLASS_EVENT); + buffer.putShortString(package); + buffer.putShortString(name); + hash.encode(buffer); + buffer.putShort((uint16_t) arguments.size()); + + for (vector::const_iterator iter = arguments.begin(); + iter != arguments.end(); iter++) + (*iter)->impl->encode(buffer); +} + +const SchemaClassKey* SchemaEventClassImpl::getClassKey() const +{ + if (!hasHash) { + hasHash = true; + hash.update(package); + hash.update(name); + for (vector::const_iterator iter = arguments.begin(); + iter != arguments.end(); iter++) + (*iter)->impl->updateHash(hash); + } + return classKey.get(); +} + +void SchemaEventClassImpl::addArgument(const SchemaArgument* argument) +{ + arguments.push_back(argument); +} + +const SchemaArgument* SchemaEventClassImpl::getArgument(int idx) const +{ + int count = 0; + for (vector::const_iterator iter = arguments.begin(); + iter != arguments.end(); iter++, count++) + if (idx == count) + return (*iter); + return 0; +} + + +//================================================================== +// Wrappers +//================================================================== + +SchemaArgument::SchemaArgument(const char* name, Typecode typecode) { impl = new SchemaArgumentImpl(name, typecode); } +SchemaArgument::SchemaArgument(SchemaArgumentImpl* i) : impl(i) {} +SchemaArgument::SchemaArgument(const SchemaArgument& from) : impl(new SchemaArgumentImpl(*(from.impl))) {} +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(name)) {} +SchemaMethod::SchemaMethod(SchemaMethodImpl* i) : impl(i) {} +SchemaMethod::SchemaMethod(const SchemaMethod& from) : impl(new SchemaMethodImpl(*(from.impl))) {} +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(name, typecode)) {} +SchemaProperty::SchemaProperty(SchemaPropertyImpl* i) : impl(i) {} +SchemaProperty::SchemaProperty(const SchemaProperty& from) : impl(new SchemaPropertyImpl(*(from.impl))) {} +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(name, typecode)) {} +SchemaStatistic::SchemaStatistic(SchemaStatisticImpl* i) : impl(i) {} +SchemaStatistic::SchemaStatistic(const SchemaStatistic& from) : impl(new SchemaStatisticImpl(*(from.impl))) {} +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(const SchemaClassKey& from) : impl(new SchemaClassKeyImpl(*(from.impl))) {} +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(); } +bool SchemaClassKey::operator==(const SchemaClassKey& other) const { return *impl == *(other.impl); } +bool SchemaClassKey::operator<(const SchemaClassKey& other) const { return *impl < *(other.impl); } + +SchemaObjectClass::SchemaObjectClass(const char* package, const char* name) : impl(new SchemaObjectClassImpl(package, name)) {} +SchemaObjectClass::SchemaObjectClass(SchemaObjectClassImpl* i) : impl(i) {} +SchemaObjectClass::SchemaObjectClass(const SchemaObjectClass& from) : impl(new SchemaObjectClassImpl(*(from.impl))) {} +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(package, name)) {} +SchemaEventClass::SchemaEventClass(SchemaEventClassImpl* i) : impl(i) {} +SchemaEventClass::SchemaEventClass(const SchemaEventClass& from) : impl(new SchemaEventClassImpl(*(from.impl))) {} +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/engine/SchemaImpl.h b/qpid/cpp/src/qmf/engine/SchemaImpl.h new file mode 100644 index 0000000000..865556f076 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.h @@ -0,0 +1,222 @@ +#ifndef _QmfEngineSchemaImpl_ +#define _QmfEngineSchemaImpl_ + +/* + * 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/engine/Schema.h" +#include +#include +#include + +namespace qmf { +namespace engine { + + // TODO: Destructors for schema classes + // TODO: Add "frozen" attribute for schema classes so they can't be modified after + // they've been registered. + + class SchemaHash { + uint8_t hash[16]; + public: + SchemaHash(); + 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); + void update(const std::string& data) { update(data.c_str(), data.size()); } + void update(Typecode t) { update((uint8_t) t); } + void update(Direction d) { update((uint8_t) d); } + 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 { + std::string name; + Typecode typecode; + Direction dir; + std::string unit; + std::string description; + + SchemaArgumentImpl(const char* n, Typecode t) : name(n), typecode(t), dir(DIR_IN) {} + SchemaArgumentImpl(qpid::framing::Buffer& buffer); + static SchemaArgument* factory(qpid::framing::Buffer& buffer); + void encode(qpid::framing::Buffer& buffer) const; + void setDirection(Direction d) { dir = d; } + void setUnit(const char* val) { unit = val; } + void setDesc(const char* desc) { description = desc; } + const std::string& getName() const { return name; } + Typecode getType() const { return typecode; } + Direction getDirection() const { return dir; } + const std::string& getUnit() const { return unit; } + const std::string& getDesc() const { return description; } + void updateHash(SchemaHash& hash) const; + }; + + struct SchemaMethodImpl { + std::string name; + std::string description; + std::vector arguments; + + SchemaMethodImpl(const char* n) : name(n) {} + SchemaMethodImpl(qpid::framing::Buffer& buffer); + static SchemaMethod* factory(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& getName() const { return name; } + const std::string& getDesc() const { return description; } + int getArgumentCount() const { return arguments.size(); } + const SchemaArgument* getArgument(int idx) const; + void updateHash(SchemaHash& hash) const; + }; + + struct SchemaPropertyImpl { + std::string name; + Typecode typecode; + Access access; + bool index; + bool optional; + std::string unit; + std::string description; + + SchemaPropertyImpl(const char* n, Typecode t) : name(n), typecode(t), access(ACCESS_READ_ONLY), index(false), optional(false) {} + SchemaPropertyImpl(qpid::framing::Buffer& buffer); + static SchemaProperty* factory(qpid::framing::Buffer& buffer); + void encode(qpid::framing::Buffer& buffer) const; + void setAccess(Access a) { access = a; } + void setIndex(bool val) { index = val; } + void setOptional(bool val) { optional = val; } + void setUnit(const char* val) { unit = val; } + void setDesc(const char* desc) { description = desc; } + const std::string& getName() const { return name; } + Typecode getType() const { return typecode; } + Access getAccess() const { return access; } + bool isIndex() const { return index; } + bool isOptional() const { return optional; } + const std::string& getUnit() const { return unit; } + const std::string& getDesc() const { return description; } + void updateHash(SchemaHash& hash) const; + }; + + struct SchemaStatisticImpl { + std::string name; + Typecode typecode; + std::string unit; + std::string description; + + SchemaStatisticImpl(const char* n, Typecode t) : name(n), typecode(t) {} + SchemaStatisticImpl(qpid::framing::Buffer& buffer); + static SchemaStatistic* factory(qpid::framing::Buffer& buffer); + void encode(qpid::framing::Buffer& buffer) const; + void setUnit(const char* val) { unit = val; } + void setDesc(const char* desc) { description = desc; } + const std::string& getName() const { return name; } + Typecode getType() const { return typecode; } + const std::string& getUnit() const { return unit; } + const std::string& getDesc() const { return description; } + void updateHash(SchemaHash& hash) const; + }; + + struct SchemaClassKeyImpl { + const std::string& package; + const std::string& name; + const SchemaHash& hash; + + // The *Container elements are only used if there isn't an external place to + // store these values. + std::string packageContainer; + std::string nameContainer; + SchemaHash hashContainer; + + SchemaClassKeyImpl(const std::string& package, const std::string& name, const SchemaHash& hash); + SchemaClassKeyImpl(qpid::framing::Buffer& buffer); + static SchemaClassKey* factory(const std::string& package, const std::string& name, const SchemaHash& hash); + static SchemaClassKey* factory(qpid::framing::Buffer& buffer); + + 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 { + std::string package; + std::string name; + mutable SchemaHash hash; + mutable bool hasHash; + std::auto_ptr classKey; + std::vector properties; + std::vector statistics; + std::vector methods; + + SchemaObjectClassImpl(const char* p, const char* n) : + package(p), name(n), hasHash(false), classKey(SchemaClassKeyImpl::factory(package, name, hash)) {} + SchemaObjectClassImpl(qpid::framing::Buffer& buffer); + static SchemaObjectClass* factory(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 SchemaClassKey* getClassKey() const; + int getPropertyCount() const { return properties.size(); } + int getStatisticCount() const { return statistics.size(); } + int getMethodCount() const { return methods.size(); } + const SchemaProperty* getProperty(int idx) const; + const SchemaStatistic* getStatistic(int idx) const; + const SchemaMethod* getMethod(int idx) const; + }; + + struct SchemaEventClassImpl { + std::string package; + std::string name; + mutable SchemaHash hash; + mutable bool hasHash; + std::auto_ptr classKey; + std::string description; + std::vector arguments; + + SchemaEventClassImpl(const char* p, const char* n) : + package(p), name(n), hasHash(false), classKey(SchemaClassKeyImpl::factory(package, name, hash)) {} + SchemaEventClassImpl(qpid::framing::Buffer& buffer); + static SchemaEventClass* factory(qpid::framing::Buffer& buffer); + + void encode(qpid::framing::Buffer& buffer) const; + void addArgument(const SchemaArgument* argument); + void setDesc(const char* desc) { description = desc; } + + const SchemaClassKey* getClassKey() const; + int getArgumentCount() const { return arguments.size(); } + const SchemaArgument* getArgument(int idx) const; + }; +} +} + +#endif + diff --git a/qpid/cpp/src/qmf/engine/SequenceManager.cpp b/qpid/cpp/src/qmf/engine/SequenceManager.cpp new file mode 100644 index 0000000000..3708105b46 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/SequenceManager.cpp @@ -0,0 +1,96 @@ +/* + * 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/engine/SequenceManager.h" + +using namespace std; +using namespace qmf::engine; +using namespace qpid::sys; + +SequenceManager::SequenceManager() : nextSequence(1) {} + +void SequenceManager::setUnsolicitedContext(SequenceContext::Ptr ctx) +{ + unsolicitedContext = ctx; +} + +uint32_t SequenceManager::reserve(SequenceContext::Ptr ctx) +{ + Mutex::ScopedLock _lock(lock); + if (ctx.get() == 0) + ctx = unsolicitedContext; + 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; + ctx->reserve(); + return seq; +} + +void SequenceManager::release(uint32_t sequence) +{ + Mutex::ScopedLock _lock(lock); + + if (sequence == 0) { + if (unsolicitedContext.get() != 0) + unsolicitedContext->release(); + return; + } + + map::iterator iter = contextMap.find(sequence); + if (iter != contextMap.end()) { + if (iter->second != 0) + iter->second->release(); + contextMap.erase(iter); + } +} + +void SequenceManager::releaseAll() +{ + Mutex::ScopedLock _lock(lock); + contextMap.clear(); +} + +void SequenceManager::dispatch(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer) +{ + Mutex::ScopedLock _lock(lock); + bool done; + + if (sequence == 0) { + if (unsolicitedContext.get() != 0) { + done = unsolicitedContext->handleMessage(opcode, sequence, buffer); + if (done) + unsolicitedContext->release(); + } + return; + } + + map::iterator iter = contextMap.find(sequence); + if (iter != contextMap.end()) { + if (iter->second != 0) { + done = iter->second->handleMessage(opcode, sequence, buffer); + if (done) { + iter->second->release(); + contextMap.erase(iter); + } + } + } +} + diff --git a/qpid/cpp/src/qmf/engine/SequenceManager.h b/qpid/cpp/src/qmf/engine/SequenceManager.h new file mode 100644 index 0000000000..5f7db8bdb3 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/SequenceManager.h @@ -0,0 +1,68 @@ +#ifndef _QmfEngineSequenceManager_ +#define _QmfEngineSequenceManager_ + +/* + * 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 +#include + +namespace qpid { + namespace framing { + class Buffer; + } +} + +namespace qmf { +namespace engine { + + class SequenceContext { + public: + typedef boost::shared_ptr Ptr; + SequenceContext() {} + virtual ~SequenceContext() {} + + virtual void reserve() = 0; + virtual bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer) = 0; + virtual void release() = 0; + }; + + class SequenceManager { + public: + SequenceManager(); + + void setUnsolicitedContext(SequenceContext::Ptr ctx); + uint32_t reserve(SequenceContext::Ptr ctx = SequenceContext::Ptr()); + void release(uint32_t sequence); + void releaseAll(); + void dispatch(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer); + + private: + mutable qpid::sys::Mutex lock; + uint32_t nextSequence; + SequenceContext::Ptr unsolicitedContext; + std::map contextMap; + }; + +} +} + +#endif + diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.cpp b/qpid/cpp/src/qmf/engine/ValueImpl.cpp new file mode 100644 index 0000000000..f80bdab866 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ValueImpl.cpp @@ -0,0 +1,266 @@ +/* + * 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/engine/ValueImpl.h" +#include + +using namespace std; +using namespace qmf::engine; +using qpid::framing::Buffer; + +ValueImpl::ValueImpl(Typecode t, Buffer& buf) : typecode(t) +{ + uint64_t first; + uint64_t second; + qpid::framing::FieldTable ft; + + switch (typecode) { + case TYPE_UINT8 : value.u32 = (uint32_t) buf.getOctet(); break; + case TYPE_UINT16 : value.u32 = (uint32_t) buf.getShort(); break; + case TYPE_UINT32 : value.u32 = (uint32_t) buf.getLong(); break; + case TYPE_UINT64 : value.u64 = buf.getLongLong(); break; + case TYPE_SSTR : buf.getShortString(stringVal); break; + case TYPE_LSTR : buf.getMediumString(stringVal); break; + case TYPE_ABSTIME : value.s64 = buf.getLongLong(); break; + case TYPE_DELTATIME : value.u64 = buf.getLongLong(); break; + case TYPE_BOOL : value.boolVal = (buf.getOctet() != 0); break; + case TYPE_FLOAT : value.floatVal = buf.getFloat(); break; + case TYPE_DOUBLE : value.doubleVal = buf.getDouble(); break; + case TYPE_INT8 : value.s32 = (int32_t) ((int8_t) buf.getOctet()); break; + case TYPE_INT16 : value.s32 = (int32_t) ((int16_t) buf.getShort()); break; + case TYPE_INT32 : value.s32 = (int32_t) buf.getLong(); break; + case TYPE_INT64 : value.s64 = buf.getLongLong(); break; + case TYPE_UUID : buf.getBin128(value.uuidVal); break; + case TYPE_REF: + first = buf.getLongLong(); + second = buf.getLongLong(); + refVal.impl->setValue(first, second); + break; + + case TYPE_MAP: + ft.decode(buf); + // TODO: either update to recursively use QMF types or reduce to int/string/... + // (maybe use another ctor with a FieldValue argument) + break; + + case TYPE_LIST: + case TYPE_ARRAY: + case TYPE_OBJECT: + default: + break; + } +} + +ValueImpl::ValueImpl(Typecode t, Typecode at) : typecode(t), valid(false), arrayTypecode(at) +{ +} + +ValueImpl::ValueImpl(Typecode t) : typecode(t) +{ + ::memset(&value, 0, sizeof(value)); +} + +Value* ValueImpl::factory(Typecode t, Buffer& b) +{ + ValueImpl* impl(new ValueImpl(t, b)); + return new Value(impl); +} + +Value* ValueImpl::factory(Typecode t) +{ + ValueImpl* impl(new ValueImpl(t)); + return new Value(impl); +} + +ValueImpl::~ValueImpl() +{ +} + +void ValueImpl::encode(Buffer& buf) const +{ + switch (typecode) { + case TYPE_UINT8 : buf.putOctet((uint8_t) value.u32); break; + case TYPE_UINT16 : buf.putShort((uint16_t) value.u32); break; + case TYPE_UINT32 : buf.putLong(value.u32); break; + case TYPE_UINT64 : buf.putLongLong(value.u64); break; + case TYPE_SSTR : buf.putShortString(stringVal); break; + case TYPE_LSTR : buf.putMediumString(stringVal); break; + case TYPE_ABSTIME : buf.putLongLong(value.s64); break; + case TYPE_DELTATIME : buf.putLongLong(value.u64); break; + case TYPE_BOOL : buf.putOctet(value.boolVal ? 1 : 0); break; + case TYPE_FLOAT : buf.putFloat(value.floatVal); break; + case TYPE_DOUBLE : buf.putDouble(value.doubleVal); break; + case TYPE_INT8 : buf.putOctet((uint8_t) value.s32); break; + case TYPE_INT16 : buf.putShort((uint16_t) value.s32); break; + case TYPE_INT32 : buf.putLong(value.s32); break; + case TYPE_INT64 : buf.putLongLong(value.s64); break; + case TYPE_UUID : buf.putBin128(value.uuidVal); break; + case TYPE_REF : refVal.impl->encode(buf); break; + case TYPE_MAP: // TODO + case TYPE_LIST: + case TYPE_ARRAY: + case TYPE_OBJECT: + default: + break; + } +} + +bool ValueImpl::keyInMap(const char* key) const +{ + return typecode == TYPE_MAP && mapVal.count(key) > 0; +} + +Value* ValueImpl::byKey(const char* key) +{ + if (keyInMap(key)) { + map::iterator iter = mapVal.find(key); + if (iter != mapVal.end()) + return &iter->second; + } + return 0; +} + +const Value* ValueImpl::byKey(const char* key) const +{ + if (keyInMap(key)) { + map::const_iterator iter = mapVal.find(key); + if (iter != mapVal.end()) + return &iter->second; + } + return 0; +} + +void ValueImpl::deleteKey(const char* key) +{ + mapVal.erase(key); +} + +void ValueImpl::insert(const char* key, Value* val) +{ + pair entry(key, *val); + mapVal.insert(entry); +} + +const char* ValueImpl::key(uint32_t idx) const +{ + map::const_iterator iter = mapVal.begin(); + for (uint32_t i = 0; i < idx; i++) { + if (iter == mapVal.end()) + break; + iter++; + } + + if (iter == mapVal.end()) + return 0; + else + return iter->first.c_str(); +} + +Value* ValueImpl::listItem(uint32_t) +{ + return 0; +} + +void ValueImpl::appendToList(Value*) +{ +} + +void ValueImpl::deleteListItem(uint32_t) +{ +} + +Value* ValueImpl::arrayItem(uint32_t) +{ + return 0; +} + +void ValueImpl::appendToArray(Value*) +{ +} + +void ValueImpl::deleteArrayItem(uint32_t) +{ +} + + +//================================================================== +// Wrappers +//================================================================== + +Value::Value(const Value& from) : impl(new ValueImpl(*(from.impl))) {} +Value::Value(Typecode t, Typecode at) : impl(new ValueImpl(t, at)) {} +Value::Value(ValueImpl* i) : impl(i) {} +Value::~Value() { delete impl;} + +Typecode Value::getType() const { return impl->getType(); } +bool Value::isNull() const { return impl->isNull(); } +void Value::setNull() { impl->setNull(); } +bool Value::isObjectId() const { return impl->isObjectId(); } +const ObjectId& Value::asObjectId() const { return impl->asObjectId(); } +void Value::setObjectId(const ObjectId& oid) { impl->setObjectId(oid); } +bool Value::isUint() const { return impl->isUint(); } +uint32_t Value::asUint() const { return impl->asUint(); } +void Value::setUint(uint32_t val) { impl->setUint(val); } +bool Value::isInt() const { return impl->isInt(); } +int32_t Value::asInt() const { return impl->asInt(); } +void Value::setInt(int32_t val) { impl->setInt(val); } +bool Value::isUint64() const { return impl->isUint64(); } +uint64_t Value::asUint64() const { return impl->asUint64(); } +void Value::setUint64(uint64_t val) { impl->setUint64(val); } +bool Value::isInt64() const { return impl->isInt64(); } +int64_t Value::asInt64() const { return impl->asInt64(); } +void Value::setInt64(int64_t val) { impl->setInt64(val); } +bool Value::isString() const { return impl->isString(); } +const char* Value::asString() const { return impl->asString(); } +void Value::setString(const char* val) { impl->setString(val); } +bool Value::isBool() const { return impl->isBool(); } +bool Value::asBool() const { return impl->asBool(); } +void Value::setBool(bool val) { impl->setBool(val); } +bool Value::isFloat() const { return impl->isFloat(); } +float Value::asFloat() const { return impl->asFloat(); } +void Value::setFloat(float val) { impl->setFloat(val); } +bool Value::isDouble() const { return impl->isDouble(); } +double Value::asDouble() const { return impl->asDouble(); } +void Value::setDouble(double val) { impl->setDouble(val); } +bool Value::isUuid() const { return impl->isUuid(); } +const uint8_t* Value::asUuid() const { return impl->asUuid(); } +void Value::setUuid(const uint8_t* val) { impl->setUuid(val); } +bool Value::isObject() const { return impl->isObject(); } +const Object* Value::asObject() const { return impl->asObject(); } +void Value::setObject(Object* val) { impl->setObject(val); } +bool Value::isMap() const { return impl->isMap(); } +bool Value::keyInMap(const char* key) const { return impl->keyInMap(key); } +Value* Value::byKey(const char* key) { return impl->byKey(key); } +const Value* Value::byKey(const char* key) const { return impl->byKey(key); } +void Value::deleteKey(const char* key) { impl->deleteKey(key); } +void Value::insert(const char* key, Value* val) { impl->insert(key, val); } +uint32_t Value::keyCount() const { return impl->keyCount(); } +const char* Value::key(uint32_t idx) const { return impl->key(idx); } +bool Value::isList() const { return impl->isList(); } +uint32_t Value::listItemCount() const { return impl->listItemCount(); } +Value* Value::listItem(uint32_t idx) { return impl->listItem(idx); } +void Value::appendToList(Value* val) { impl->appendToList(val); } +void Value::deleteListItem(uint32_t idx) { impl->deleteListItem(idx); } +bool Value::isArray() const { return impl->isArray(); } +Typecode Value::arrayType() const { return impl->arrayType(); } +uint32_t Value::arrayItemCount() const { return impl->arrayItemCount(); } +Value* Value::arrayItem(uint32_t idx) { return impl->arrayItem(idx); } +void Value::appendToArray(Value* val) { impl->appendToArray(val); } +void Value::deleteArrayItem(uint32_t idx) { impl->deleteArrayItem(idx); } + diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.h b/qpid/cpp/src/qmf/engine/ValueImpl.h new file mode 100644 index 0000000000..b6adae5d93 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/ValueImpl.h @@ -0,0 +1,150 @@ +#ifndef _QmfEngineValueImpl_ +#define _QmfEngineValueImpl_ + +/* + * 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 +#include +#include +#include +#include +#include +#include +#include +#include + +namespace qmf { +namespace engine { + + // TODO: set valid flag on all value settors + // TODO: add a modified flag and accessors + + struct ValueImpl { + const Typecode typecode; + bool valid; + + ObjectId refVal; + std::string stringVal; + std::auto_ptr objectVal; + std::map mapVal; + std::vector vectorVal; + Typecode arrayTypecode; + + union { + uint32_t u32; + uint64_t u64; + int32_t s32; + int64_t s64; + bool boolVal; + float floatVal; + double doubleVal; + uint8_t uuidVal[16]; + } value; + + ValueImpl(const ValueImpl& from) : + typecode(from.typecode), valid(from.valid), refVal(from.refVal), stringVal(from.stringVal), + objectVal(from.objectVal.get() ? new Object(*(from.objectVal)) : 0), + mapVal(from.mapVal), vectorVal(from.vectorVal), arrayTypecode(from.arrayTypecode), + value(from.value) {} + + ValueImpl(Typecode t, Typecode at); + ValueImpl(Typecode t, qpid::framing::Buffer& b); + ValueImpl(Typecode t); + static Value* factory(Typecode t, qpid::framing::Buffer& b); + static Value* factory(Typecode t); + ~ValueImpl(); + + void encode(qpid::framing::Buffer& b) const; + + Typecode getType() const { return typecode; } + bool isNull() const { return !valid; } + void setNull() { valid = false; } + + bool isObjectId() const { return typecode == TYPE_REF; } + const ObjectId& asObjectId() const { return refVal; } + void setObjectId(const ObjectId& o) { refVal = o; } // TODO + + bool isUint() const { return typecode >= TYPE_UINT8 && typecode <= TYPE_UINT32; } + uint32_t asUint() const { return value.u32; } + void setUint(uint32_t val) { value.u32 = val; } + + bool isInt() const { return typecode >= TYPE_INT8 && typecode <= TYPE_INT32; } + int32_t asInt() const { return value.s32; } + void setInt(int32_t val) { value.s32 = val; } + + bool isUint64() const { return typecode == TYPE_UINT64 || typecode == TYPE_DELTATIME; } + uint64_t asUint64() const { return value.u64; } + void setUint64(uint64_t val) { value.u64 = val; } + + bool isInt64() const { return typecode == TYPE_INT64 || typecode == TYPE_ABSTIME; } + int64_t asInt64() const { return value.s64; } + void setInt64(int64_t val) { value.s64 = val; } + + bool isString() const { return typecode == TYPE_SSTR || typecode == TYPE_LSTR; } + const char* asString() const { return stringVal.c_str(); } + void setString(const char* val) { stringVal = val; } + + bool isBool() const { return typecode == TYPE_BOOL; } + bool asBool() const { return value.boolVal; } + void setBool(bool val) { value.boolVal = val; } + + bool isFloat() const { return typecode == TYPE_FLOAT; } + float asFloat() const { return value.floatVal; } + void setFloat(float val) { value.floatVal = val; } + + bool isDouble() const { return typecode == TYPE_DOUBLE; } + double asDouble() const { return value.doubleVal; } + void setDouble(double val) { value.doubleVal = val; } + + bool isUuid() const { return typecode == TYPE_UUID; } + const uint8_t* asUuid() const { return value.uuidVal; } + void setUuid(const uint8_t* val) { ::memcpy(value.uuidVal, val, 16); } + + bool isObject() const { return typecode == TYPE_OBJECT; } + Object* asObject() const { return objectVal.get(); } + void setObject(Object* val) { objectVal.reset(val); } + + bool isMap() const { return typecode == TYPE_MAP; } + bool keyInMap(const char* key) const; + Value* byKey(const char* key); + const Value* byKey(const char* key) const; + void deleteKey(const char* key); + void insert(const char* key, Value* val); + uint32_t keyCount() const { return mapVal.size(); } + const char* key(uint32_t idx) const; + + bool isList() const { return typecode == TYPE_LIST; } + uint32_t listItemCount() const { return vectorVal.size(); } + Value* listItem(uint32_t idx); + void appendToList(Value* val); + void deleteListItem(uint32_t idx); + + bool isArray() const { return typecode == TYPE_ARRAY; } + Typecode arrayType() const { return arrayTypecode; } + uint32_t arrayItemCount() const { return vectorVal.size(); } + Value* arrayItem(uint32_t idx); + void appendToArray(Value* val); + void deleteArrayItem(uint32_t idx); + }; +} +} + +#endif + -- cgit v1.2.1 From 74fecbf7759d549824bf92266c51a3d5180b6b33 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Tue, 29 Sep 2009 03:21:49 +0000 Subject: QMF Engine updates: - Connected console handler callbacks - Added string representations for a number of object classes - Added a feature that completes query requests sent to disconnected agents git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@819819 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp | 109 ++++++++++++++++++------ qpid/cpp/src/qmf/engine/BrokerProxyImpl.h | 17 +++- qpid/cpp/src/qmf/engine/ConsoleImpl.cpp | 69 +++++++++++++-- qpid/cpp/src/qmf/engine/ConsoleImpl.h | 16 +++- qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp | 6 +- qpid/cpp/src/qmf/engine/ObjectIdImpl.h | 3 +- qpid/cpp/src/qmf/engine/ResilientConnection.cpp | 15 ++-- qpid/cpp/src/qmf/engine/SchemaImpl.cpp | 6 +- qpid/cpp/src/qmf/engine/SchemaImpl.h | 3 +- qpid/cpp/src/qmf/engine/SequenceManager.cpp | 6 +- qpid/cpp/src/qmf/engine/SequenceManager.h | 4 +- 11 files changed, 200 insertions(+), 54 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp index 36d3ffe361..e296254bf8 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp @@ -23,6 +23,7 @@ #include "qpid/Address.h" #include "qpid/sys/SystemInfo.h" #include +#include #include #include #include @@ -109,18 +110,23 @@ void BrokerProxyImpl::sessionClosed() void BrokerProxyImpl::startProtocol() { - Mutex::ScopedLock _lock(lock); - char rawbuffer[512]; - Buffer buffer(rawbuffer, 512); + AgentProxyPtr agent(AgentProxyImpl::factory(console, publicObject, 0, "Agent embedded in broker")); + { + Mutex::ScopedLock _lock(lock); + char rawbuffer[512]; + Buffer buffer(rawbuffer, 512); - agentList[0] = AgentProxyPtr(AgentProxyImpl::factory(console, publicObject, 0, "Agent embedded in broker")); + agentList[0] = agent; - requestsOutstanding = 1; - topicBound = false; - uint32_t sequence(seqMgr.reserve()); - Protocol::encodeHeader(buffer, Protocol::OP_BROKER_REQUEST, sequence); - sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); - QPID_LOG(trace, "SENT BrokerRequest seq=" << sequence); + requestsOutstanding = 1; + topicBound = false; + uint32_t sequence(seqMgr.reserve()); + Protocol::encodeHeader(buffer, Protocol::OP_BROKER_REQUEST, sequence); + sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT BrokerRequest seq=" << sequence); + } + + console.impl->eventAgentAdded(agent); } void BrokerProxyImpl::sendBufferLH(Buffer& buf, const string& destination, const string& routingKey) @@ -145,7 +151,7 @@ void BrokerProxyImpl::handleRcvMessage(Message& message) uint32_t sequence; while (Protocol::checkHeader(inBuffer, &opcode, &sequence)) - seqMgr.dispatch(opcode, sequence, inBuffer); + seqMgr.dispatch(opcode, sequence, message.routingKey ? string(message.routingKey) : string(), inBuffer); } bool BrokerProxyImpl::getXmtMessage(Message& item) const @@ -216,6 +222,7 @@ void BrokerProxyImpl::sendGetRequestLH(SequenceContext::Ptr queryContext, const stringstream key; Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); uint32_t sequence(seqMgr.reserve(queryContext)); + agent->impl->addSequence(sequence); Protocol::encodeHeader(outBuffer, Protocol::OP_GET_QUERY, sequence); query.impl->encode(outBuffer); @@ -406,9 +413,23 @@ MethodResponsePtr BrokerProxyImpl::handleMethodResponse(Buffer& inBuffer, uint32 return response; } -void BrokerProxyImpl::handleHeartbeatIndication(Buffer& /*inBuffer*/, uint32_t /*seq*/) +void BrokerProxyImpl::handleHeartbeatIndication(Buffer& inBuffer, uint32_t seq, const string& routingKey) { - // TODO + vector tokens = qpid::split(routingKey, "."); + uint32_t agentBank; + uint64_t timestamp; + + if (routingKey.empty() || tokens.size() != 4) + agentBank = 0; + else + agentBank = ::atoi(tokens[3].c_str()); + + timestamp = inBuffer.getLongLong(); + map::const_iterator iter = agentList.find(agentBank); + if (iter != agentList.end()) { + console.impl->eventAgentHeartbeat(iter->second, timestamp); + } + QPID_LOG(trace, "RCVD HeartbeatIndication seq=" << seq << " agentBank=" << agentBank); } void BrokerProxyImpl::handleEventIndication(Buffer& /*inBuffer*/, uint32_t /*seq*/) @@ -481,11 +502,24 @@ ObjectPtr BrokerProxyImpl::handleObjectIndication(Buffer& inBuffer, uint32_t seq void BrokerProxyImpl::updateAgentList(ObjectPtr obj) { Value* value = obj->getValue("agentBank"); + Mutex::ScopedLock _lock(lock); if (value != 0 && value->isUint()) { uint32_t agentBank = value->asUint(); if (obj->isDeleted()) { - agentList.erase(agentBank); - QPID_LOG(trace, "Agent at bank " << agentBank << " removed from agent list"); + map::iterator iter = agentList.find(agentBank); + if (iter != agentList.end()) { + AgentProxyPtr agent(iter->second); + console.impl->eventAgentDeleted(agent); + agentList.erase(agentBank); + QPID_LOG(trace, "Agent at bank " << agentBank << " removed from agent list"); + + // + // Release all sequence numbers for requests in-flight to this agent. + // Since the agent is no longer connected, these requests would not + // otherwise complete. + // + agent->impl->releaseInFlight(seqMgr); + } } else { Value* str = obj->getValue("label"); string label; @@ -493,7 +527,9 @@ void BrokerProxyImpl::updateAgentList(ObjectPtr obj) label = str->asString(); map::const_iterator iter = agentList.find(agentBank); if (iter == agentList.end()) { - agentList[agentBank] = AgentProxyPtr(AgentProxyImpl::factory(console, publicObject, agentBank, label)); + AgentProxyPtr agent(AgentProxyImpl::factory(console, publicObject, agentBank, label)); + agentList[agentBank] = agent; + console.impl->eventAgentAdded(agent); QPID_LOG(trace, "Agent '" << label << "' found at bank " << agentBank); } } @@ -572,9 +608,11 @@ MethodResponse* MethodResponseImpl::factory(uint32_t status, const std::string& return new MethodResponse(impl); } -bool StaticContext::handleMessage(uint8_t opcode, uint32_t sequence, Buffer& buffer) +bool StaticContext::handleMessage(uint8_t opcode, uint32_t sequence, const string& routingKey, Buffer& buffer) { + ObjectPtr object; bool completeContext = false; + if (opcode == Protocol::OP_BROKER_RESPONSE) { broker.handleBrokerResponse(buffer, sequence); completeContext = true; @@ -592,15 +630,21 @@ bool StaticContext::handleMessage(uint8_t opcode, uint32_t sequence, Buffer& buf else if (opcode == Protocol::OP_CLASS_INDICATION) broker.handleClassIndication(buffer, sequence); else if (opcode == Protocol::OP_HEARTBEAT_INDICATION) - broker.handleHeartbeatIndication(buffer, sequence); + broker.handleHeartbeatIndication(buffer, sequence, routingKey); else if (opcode == Protocol::OP_EVENT_INDICATION) broker.handleEventIndication(buffer, sequence); - else if (opcode == Protocol::OP_PROPERTY_INDICATION) - broker.handleObjectIndication(buffer, sequence, true, false); - else if (opcode == Protocol::OP_STATISTIC_INDICATION) - broker.handleObjectIndication(buffer, sequence, false, true); - else if (opcode == Protocol::OP_OBJECT_INDICATION) - broker.handleObjectIndication(buffer, sequence, true, true); + else if (opcode == Protocol::OP_PROPERTY_INDICATION) { + object = broker.handleObjectIndication(buffer, sequence, true, false); + broker.console.impl->eventObjectUpdate(object, true, false); + } + else if (opcode == Protocol::OP_STATISTIC_INDICATION) { + object = broker.handleObjectIndication(buffer, sequence, false, true); + broker.console.impl->eventObjectUpdate(object, false, true); + } + else if (opcode == Protocol::OP_OBJECT_INDICATION) { + object = broker.handleObjectIndication(buffer, sequence, true, true); + broker.console.impl->eventObjectUpdate(object, true, true); + } else { QPID_LOG(trace, "StaticContext::handleMessage invalid opcode: " << opcode); completeContext = true; @@ -627,7 +671,7 @@ void QueryContext::release() broker.eventQueue.push_back(broker.eventQueryComplete(userContext, queryResponse)); } -bool QueryContext::handleMessage(uint8_t opcode, uint32_t sequence, Buffer& buffer) +bool QueryContext::handleMessage(uint8_t opcode, uint32_t sequence, const string& /*routingKey*/, Buffer& buffer) { bool completeContext = false; ObjectPtr object; @@ -635,6 +679,19 @@ bool QueryContext::handleMessage(uint8_t opcode, uint32_t sequence, Buffer& buff if (opcode == Protocol::OP_COMMAND_COMPLETE) { broker.handleCommandComplete(buffer, sequence); completeContext = true; + + // + // Visit each agent and remove the sequence from that agent's in-flight list. + // This could be made more efficient because only one agent will have this sequence + // in its list. + // + map copy; + { + Mutex::ScopedLock _block(broker.lock); + copy = broker.agentList; + } + for (map::iterator iter = copy.begin(); iter != copy.end(); iter++) + iter->second->impl->delSequence(sequence); } else if (opcode == Protocol::OP_OBJECT_INDICATION) { object = broker.handleObjectIndication(buffer, sequence, true, true); @@ -655,7 +712,7 @@ void MethodContext::release() broker.eventQueue.push_back(broker.eventMethodResponse(userContext, methodResponse)); } -bool MethodContext::handleMessage(uint8_t opcode, uint32_t sequence, Buffer& buffer) +bool MethodContext::handleMessage(uint8_t opcode, uint32_t sequence, const string& /*routingKey*/, Buffer& buffer) { if (opcode == Protocol::OP_METHOD_RESPONSE) methodResponse = broker.handleMethodResponse(buffer, sequence, schema); diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h index 660cb86c61..738424bce1 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h @@ -36,6 +36,7 @@ #include #include #include +#include #include namespace qmf { @@ -98,6 +99,7 @@ namespace engine { BrokerProxy& broker; uint32_t agentBank; std::string label; + std::set inFlightSequences; AgentProxyImpl(Console& c, BrokerProxy& b, uint32_t ab, const std::string& l) : console(c), broker(b), agentBank(ab), label(l) {} static AgentProxy* factory(Console& c, BrokerProxy& b, uint32_t ab, const std::string& l) { @@ -106,6 +108,13 @@ namespace engine { } ~AgentProxyImpl() {} const std::string& getLabel() const { return label; } + void addSequence(uint32_t seq) { inFlightSequences.insert(seq); } + void delSequence(uint32_t seq) { inFlightSequences.erase(seq); } + void releaseInFlight(SequenceManager& seqMgr) { + for (std::set::iterator iter = inFlightSequences.begin(); iter != inFlightSequences.end(); iter++) + seqMgr.release(*iter); + inFlightSequences.clear(); + } }; class BrokerProxyImpl : public boost::noncopyable { @@ -166,7 +175,7 @@ namespace engine { void handleCommandComplete(qpid::framing::Buffer& inBuffer, uint32_t seq); void handleClassIndication(qpid::framing::Buffer& inBuffer, uint32_t seq); MethodResponsePtr handleMethodResponse(qpid::framing::Buffer& inBuffer, uint32_t seq, const SchemaMethod* schema); - void handleHeartbeatIndication(qpid::framing::Buffer& inBuffer, uint32_t seq); + void handleHeartbeatIndication(qpid::framing::Buffer& inBuffer, uint32_t seq, const std::string& routingKey); void handleEventIndication(qpid::framing::Buffer& inBuffer, uint32_t seq); void handleSchemaResponse(qpid::framing::Buffer& inBuffer, uint32_t seq); ObjectPtr handleObjectIndication(qpid::framing::Buffer& inBuffer, uint32_t seq, bool prop, bool stat); @@ -186,7 +195,7 @@ namespace engine { virtual ~StaticContext() {} void reserve() {} void release() { broker.staticRelease(); } - bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer); + bool handleMessage(uint8_t opcode, uint32_t sequence, const std::string& routingKey, qpid::framing::Buffer& buffer); BrokerProxyImpl& broker; }; @@ -199,7 +208,7 @@ namespace engine { virtual ~QueryContext() {} void reserve(); void release(); - bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer); + bool handleMessage(uint8_t opcode, uint32_t sequence, const std::string& routingKey, qpid::framing::Buffer& buffer); mutable qpid::sys::Mutex lock; BrokerProxyImpl& broker; @@ -213,7 +222,7 @@ namespace engine { virtual ~MethodContext() {} void reserve() {} void release(); - bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer); + bool handleMessage(uint8_t opcode, uint32_t sequence, const std::string& routingKey, qpid::framing::Buffer& buffer); BrokerProxyImpl& broker; void* userContext; diff --git a/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp b/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp index c856f04c51..c2d1f51f2b 100644 --- a/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp @@ -57,11 +57,13 @@ ConsoleEvent ConsoleEventImpl::copy() ::memset(&item, 0, sizeof(ConsoleEvent)); item.kind = kind; item.agent = agent.get(); - item.classKey = classKey.get(); - item.object = object; + item.classKey = classKey; + item.object = object.get(); item.context = context; item.event = event; item.timestamp = timestamp; + item.hasProps = hasProps; + item.hasStats = hasStats; STRING_REF(name); @@ -274,9 +276,11 @@ void ConsoleImpl::endSync(SyncQuery& sync) void ConsoleImpl::learnPackage(const string& packageName) { Mutex::ScopedLock _lock(lock); - if (packages.find(packageName) == packages.end()) + if (packages.find(packageName) == packages.end()) { packages.insert(pair > (packageName, pair(ObjectClassList(), EventClassList()))); + eventNewPackage(packageName); + } } void ConsoleImpl::learnClass(SchemaObjectClass* cls) @@ -288,8 +292,10 @@ void ConsoleImpl::learnClass(SchemaObjectClass* cls) return; ObjectClassList& list = pIter->second.first; - if (list.find(key) == list.end()) + if (list.find(key) == list.end()) { list[key] = cls; + eventNewClass(key); + } } void ConsoleImpl::learnClass(SchemaEventClass* cls) @@ -301,8 +307,10 @@ void ConsoleImpl::learnClass(SchemaEventClass* cls) return; EventClassList& list = pIter->second.second; - if (list.find(key) == list.end()) + if (list.find(key) == list.end()) { list[key] = cls; + eventNewClass(key); + } } bool ConsoleImpl::haveClass(const SchemaClassKey* key) const @@ -333,6 +341,57 @@ SchemaObjectClass* ConsoleImpl::getSchema(const SchemaClassKey* key) const return iter->second; } +void ConsoleImpl::eventAgentAdded(boost::shared_ptr agent) +{ + ConsoleEventImpl::Ptr event(new ConsoleEventImpl(ConsoleEvent::AGENT_ADDED)); + event->agent = agent; + Mutex::ScopedLock _lock(lock); + eventQueue.push_back(event); +} + +void ConsoleImpl::eventAgentDeleted(boost::shared_ptr agent) +{ + ConsoleEventImpl::Ptr event(new ConsoleEventImpl(ConsoleEvent::AGENT_DELETED)); + event->agent = agent; + Mutex::ScopedLock _lock(lock); + eventQueue.push_back(event); +} + +void ConsoleImpl::eventNewPackage(const string& packageName) +{ + ConsoleEventImpl::Ptr event(new ConsoleEventImpl(ConsoleEvent::NEW_PACKAGE)); + event->name = packageName; + Mutex::ScopedLock _lock(lock); + eventQueue.push_back(event); +} + +void ConsoleImpl::eventNewClass(const SchemaClassKey* key) +{ + ConsoleEventImpl::Ptr event(new ConsoleEventImpl(ConsoleEvent::NEW_CLASS)); + event->classKey = key; + Mutex::ScopedLock _lock(lock); + eventQueue.push_back(event); +} + +void ConsoleImpl::eventObjectUpdate(ObjectPtr object, bool prop, bool stat) +{ + ConsoleEventImpl::Ptr event(new ConsoleEventImpl(ConsoleEvent::OBJECT_UPDATE)); + event->object = object; + event->hasProps = prop; + event->hasStats = stat; + Mutex::ScopedLock _lock(lock); + eventQueue.push_back(event); +} + +void ConsoleImpl::eventAgentHeartbeat(boost::shared_ptr agent, uint64_t timestamp) +{ + ConsoleEventImpl::Ptr event(new ConsoleEventImpl(ConsoleEvent::AGENT_HEARTBEAT)); + event->agent = agent; + event->timestamp = timestamp; + Mutex::ScopedLock _lock(lock); + eventQueue.push_back(event); +} + //================================================================== // Wrappers //================================================================== diff --git a/qpid/cpp/src/qmf/engine/ConsoleImpl.h b/qpid/cpp/src/qmf/engine/ConsoleImpl.h index 2c4ee48a02..8f99c5e6b9 100644 --- a/qpid/cpp/src/qmf/engine/ConsoleImpl.h +++ b/qpid/cpp/src/qmf/engine/ConsoleImpl.h @@ -56,14 +56,16 @@ namespace engine { ConsoleEvent::EventKind kind; boost::shared_ptr agent; std::string name; - boost::shared_ptr classKey; - Object* object; + const SchemaClassKey* classKey; + boost::shared_ptr object; void* context; Event* event; uint64_t timestamp; + bool hasProps; + bool hasStats; ConsoleEventImpl(ConsoleEvent::EventKind k) : - kind(k), object(0), context(0), event(0), timestamp(0) {} + kind(k), classKey(0), context(0), event(0), timestamp(0) {} ~ConsoleEventImpl() {} ConsoleEvent copy(); }; @@ -101,6 +103,7 @@ namespace engine { private: friend class BrokerProxyImpl; + friend struct StaticContext; const ConsoleSettings& settings; mutable qpid::sys::Mutex lock; std::deque eventQueue; @@ -127,6 +130,13 @@ namespace engine { void learnClass(SchemaEventClass* cls); bool haveClass(const SchemaClassKey* key) const; SchemaObjectClass* getSchema(const SchemaClassKey* key) const; + + void eventAgentAdded(boost::shared_ptr agent); + void eventAgentDeleted(boost::shared_ptr agent); + void eventNewPackage(const std::string& packageName); + void eventNewClass(const SchemaClassKey* key); + void eventObjectUpdate(ObjectPtr object, bool prop, bool stat); + void eventAgentHeartbeat(boost::shared_ptr agent, uint64_t timestamp); }; } } diff --git a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp index 032bc557c0..5b925045bf 100644 --- a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp @@ -111,13 +111,14 @@ void ObjectIdImpl::fromString(const std::string& repr) agent = 0; } -std::string ObjectIdImpl::asString() const +const string& ObjectIdImpl::asString() const { stringstream val; val << (int) getFlags() << "-" << getSequence() << "-" << getBrokerBank() << "-" << getAgentBank() << "-" << getObjectNum(); - return val.str(); + repr = val.str(); + return repr; } bool ObjectIdImpl::operator==(const ObjectIdImpl& other) const @@ -154,6 +155,7 @@ uint64_t ObjectId::getObjectNum() const { return impl->getObjectNum(); } uint32_t ObjectId::getObjectNumHi() const { return impl->getObjectNumHi(); } uint32_t ObjectId::getObjectNumLo() const { return impl->getObjectNumLo(); } bool ObjectId::isDurable() const { return impl->isDurable(); } +const char* ObjectId::str() const { return impl->asString().c_str(); } 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/engine/ObjectIdImpl.h b/qpid/cpp/src/qmf/engine/ObjectIdImpl.h index 44fa8adffc..d9871ac217 100644 --- a/qpid/cpp/src/qmf/engine/ObjectIdImpl.h +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.h @@ -38,6 +38,7 @@ namespace engine { AgentAttachment* agent; uint64_t first; uint64_t second; + mutable std::string repr; ObjectIdImpl() : agent(0), first(0), second(0) {} ObjectIdImpl(qpid::framing::Buffer& buffer); @@ -49,7 +50,7 @@ namespace engine { void decode(qpid::framing::Buffer& buffer); void encode(qpid::framing::Buffer& buffer) const; void fromString(const std::string& repr); - std::string asString() const; + const std::string& asString() const; uint8_t getFlags() const { return (first & 0xF000000000000000LL) >> 60; } uint16_t getSequence() const { return (first & 0x0FFF000000000000LL) >> 48; } uint32_t getBrokerBank() const { return (first & 0x0000FFFFF0000000LL) >> 28; } diff --git a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp index 709cfd1236..9502130288 100644 --- a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp +++ b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp @@ -171,15 +171,20 @@ void RCSession::received(client::Message& msg) MessageImpl qmsg; qmsg.body = msg.getData(); - qpid::framing::MessageProperties p = msg.getMessageProperties(); - if (p.hasReplyTo()) { - const qpid::framing::ReplyTo& rt = p.getReplyTo(); + qpid::framing::DeliveryProperties dp = msg.getDeliveryProperties(); + if (dp.hasRoutingKey()) { + qmsg.routingKey = dp.getRoutingKey(); + } + + qpid::framing::MessageProperties mp = msg.getMessageProperties(); + if (mp.hasReplyTo()) { + const qpid::framing::ReplyTo& rt = mp.getReplyTo(); qmsg.replyExchange = rt.getExchange(); qmsg.replyKey = rt.getRoutingKey(); } - if (p.hasUserId()) { - qmsg.userId = p.getUserId(); + if (mp.hasUserId()) { + qmsg.userId = mp.getUserId(); } connImpl.EnqueueEvent(ResilientConnectionEvent::RECV, userContext, qmsg); diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp index fb09980680..e366a66826 100644 --- a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp @@ -326,12 +326,13 @@ bool SchemaClassKeyImpl::operator<(const SchemaClassKeyImpl& other) const return hash < other.hash; } -string SchemaClassKeyImpl::str() const +const string& SchemaClassKeyImpl::str() const { Uuid printableHash(hash.get()); stringstream str; str << package << ":" << name << "(" << printableHash << ")"; - return str.str(); + repr = str.str(); + return repr; } SchemaObjectClassImpl::SchemaObjectClassImpl(Buffer& buffer) : hasHash(true), classKey(SchemaClassKeyImpl::factory(package, name, hash)) @@ -579,6 +580,7 @@ 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(); } +const char* SchemaClassKey::asString() const { return impl->str().c_str(); } bool SchemaClassKey::operator==(const SchemaClassKey& other) const { return *impl == *(other.impl); } bool SchemaClassKey::operator<(const SchemaClassKey& other) const { return *impl < *(other.impl); } diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.h b/qpid/cpp/src/qmf/engine/SchemaImpl.h index 865556f076..af3a1d98e4 100644 --- a/qpid/cpp/src/qmf/engine/SchemaImpl.h +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.h @@ -142,6 +142,7 @@ namespace engine { const std::string& package; const std::string& name; const SchemaHash& hash; + mutable std::string repr; // The *Container elements are only used if there isn't an external place to // store these values. @@ -161,7 +162,7 @@ namespace engine { void encode(qpid::framing::Buffer& buffer) const; bool operator==(const SchemaClassKeyImpl& other) const; bool operator<(const SchemaClassKeyImpl& other) const; - std::string str() const; + const std::string& str() const; }; struct SchemaObjectClassImpl { diff --git a/qpid/cpp/src/qmf/engine/SequenceManager.cpp b/qpid/cpp/src/qmf/engine/SequenceManager.cpp index 3708105b46..4a4644a8b9 100644 --- a/qpid/cpp/src/qmf/engine/SequenceManager.cpp +++ b/qpid/cpp/src/qmf/engine/SequenceManager.cpp @@ -68,14 +68,14 @@ void SequenceManager::releaseAll() contextMap.clear(); } -void SequenceManager::dispatch(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer) +void SequenceManager::dispatch(uint8_t opcode, uint32_t sequence, const string& routingKey, qpid::framing::Buffer& buffer) { Mutex::ScopedLock _lock(lock); bool done; if (sequence == 0) { if (unsolicitedContext.get() != 0) { - done = unsolicitedContext->handleMessage(opcode, sequence, buffer); + done = unsolicitedContext->handleMessage(opcode, sequence, routingKey, buffer); if (done) unsolicitedContext->release(); } @@ -85,7 +85,7 @@ void SequenceManager::dispatch(uint8_t opcode, uint32_t sequence, qpid::framing: map::iterator iter = contextMap.find(sequence); if (iter != contextMap.end()) { if (iter->second != 0) { - done = iter->second->handleMessage(opcode, sequence, buffer); + done = iter->second->handleMessage(opcode, sequence, routingKey, buffer); if (done) { iter->second->release(); contextMap.erase(iter); diff --git a/qpid/cpp/src/qmf/engine/SequenceManager.h b/qpid/cpp/src/qmf/engine/SequenceManager.h index 5f7db8bdb3..9e47e38610 100644 --- a/qpid/cpp/src/qmf/engine/SequenceManager.h +++ b/qpid/cpp/src/qmf/engine/SequenceManager.h @@ -40,7 +40,7 @@ namespace engine { virtual ~SequenceContext() {} virtual void reserve() = 0; - virtual bool handleMessage(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer) = 0; + virtual bool handleMessage(uint8_t opcode, uint32_t sequence, const std::string& routingKey, qpid::framing::Buffer& buffer) = 0; virtual void release() = 0; }; @@ -52,7 +52,7 @@ namespace engine { uint32_t reserve(SequenceContext::Ptr ctx = SequenceContext::Ptr()); void release(uint32_t sequence); void releaseAll(); - void dispatch(uint8_t opcode, uint32_t sequence, qpid::framing::Buffer& buffer); + void dispatch(uint8_t opcode, uint32_t sequence, const std::string& routingKey, qpid::framing::Buffer& buffer); private: mutable qpid::sys::Mutex lock; -- cgit v1.2.1 From 0f7c02dae11afed532bf60302cf2185f422fd23c Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Fri, 2 Oct 2009 17:45:58 +0000 Subject: QMF Fixes: - In ruby: waitForStable => wait_for_stable - Setting an unknown attribute on connection settings now throws an exception - Added connected? accessor to Connection git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@821109 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp | 7 +++++-- qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h | 2 +- 2 files changed, 6 insertions(+), 3 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp index 8c918eec20..2cd6af10f8 100644 --- a/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp @@ -56,7 +56,7 @@ ConnectionSettingsImpl::ConnectionSettingsImpl(const string& /*url*/) : // TODO: Parse the URL } -void ConnectionSettingsImpl::setAttr(const string& key, const Value& value) +bool ConnectionSettingsImpl::setAttr(const string& key, const Value& value) { if (key == attrProtocol) clientSettings.protocol = value.asString(); else if (key == attrHost) clientSettings.host = value.asString(); @@ -79,6 +79,9 @@ void ConnectionSettingsImpl::setAttr(const string& key, const Value& value) else if (key == attrRetryDelayMax) retryDelayMax = value.asUint(); else if (key == attrRetryDelayFactor) retryDelayFactor = value.asUint(); else if (key == attrSendUserId) sendUserId = value.asBool(); + else + return false; + return true; } Value ConnectionSettingsImpl::getAttr(const string& key) const @@ -257,7 +260,7 @@ ConnectionSettings::ConnectionSettings(const ConnectionSettings& from) { impl = ConnectionSettings::ConnectionSettings() { impl = new ConnectionSettingsImpl(); } ConnectionSettings::ConnectionSettings(const char* url) { impl = new ConnectionSettingsImpl(url); } ConnectionSettings::~ConnectionSettings() { delete impl; } -void ConnectionSettings::setAttr(const char* key, const Value& value) { impl->setAttr(key, value); } +bool ConnectionSettings::setAttr(const char* key, const Value& value) { return 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); } diff --git a/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h index c47c8bee2e..98bf87868b 100644 --- a/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h +++ b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.h @@ -41,7 +41,7 @@ namespace engine { ConnectionSettingsImpl(); ConnectionSettingsImpl(const std::string& url); ~ConnectionSettingsImpl() {} - void setAttr(const std::string& key, const Value& value); + bool 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); -- cgit v1.2.1 From 2df0b393db42d098f590f6b00fb96b0c15b1f6f9 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Fri, 2 Oct 2009 21:43:22 +0000 Subject: QMF Fixes in Ruby Console: - AgentProxy object now has broker_bank and agent_bank accessors - Fixed a bug in the returning og reference values (ObjectId) - Added package_name accessor to SchemaObjectClass - Added accessors for agent/broker bank to ObjectId git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@821181 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp | 2 ++ qpid/cpp/src/qmf/engine/BrokerProxyImpl.h | 2 ++ qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp | 4 ++++ 3 files changed, 8 insertions(+) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp index e296254bf8..1a2b3e6555 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp @@ -730,6 +730,8 @@ bool MethodContext::handleMessage(uint8_t opcode, uint32_t sequence, const strin AgentProxy::AgentProxy(AgentProxyImpl* i) : impl(i) {} AgentProxy::~AgentProxy() { delete impl; } const char* AgentProxy::getLabel() const { return impl->getLabel().c_str(); } +uint32_t AgentProxy::getBrokerBank() const { return impl->getBrokerBank(); } +uint32_t AgentProxy::getAgentBank() const { return impl->getAgentBank(); } BrokerProxy::BrokerProxy(Console& console) : impl(new BrokerProxyImpl(*this, console)) {} BrokerProxy::~BrokerProxy() { delete impl; } diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h index 738424bce1..798a5fdc76 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h @@ -108,6 +108,8 @@ namespace engine { } ~AgentProxyImpl() {} const std::string& getLabel() const { return label; } + uint32_t getBrokerBank() const { return 1; } + uint32_t getAgentBank() const { return agentBank; } void addSequence(uint32_t seq) { inFlightSequences.insert(seq); } void delSequence(uint32_t seq) { inFlightSequences.erase(seq); } void releaseInFlight(SequenceManager& seqMgr) { diff --git a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp index 5b925045bf..b08ae2756c 100644 --- a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp @@ -156,6 +156,10 @@ uint32_t ObjectId::getObjectNumHi() const { return impl->getObjectNumHi(); } uint32_t ObjectId::getObjectNumLo() const { return impl->getObjectNumLo(); } bool ObjectId::isDurable() const { return impl->isDurable(); } const char* ObjectId::str() const { return impl->asString().c_str(); } +uint8_t ObjectId::getFlags() const { return impl->getFlags(); } +uint16_t ObjectId::getSequence() const { return impl->getSequence(); } +uint32_t ObjectId::getBrokerBank() const { return impl->getBrokerBank(); } +uint32_t ObjectId::getAgentBank() const { return impl->getAgentBank(); } 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; } -- cgit v1.2.1 From 700636714d25b073bdfea2d982b5f67f11b240e9 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Thu, 15 Oct 2009 21:38:07 +0000 Subject: Bug fix: Console sent get-requests (by objectId) to all agents, not just the agent identified in the objectId. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@825672 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp | 19 ++++++++++++++++--- qpid/cpp/src/qmf/engine/BrokerProxyImpl.h | 2 +- qpid/cpp/src/qmf/engine/QueryImpl.h | 2 ++ 3 files changed, 19 insertions(+), 4 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp index 1a2b3e6555..0a1769f891 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp @@ -206,19 +206,31 @@ void BrokerProxyImpl::sendQuery(const Query& query, void* context, const AgentPr { SequenceContext::Ptr queryContext(new QueryContext(*this, context)); Mutex::ScopedLock _lock(lock); + bool sent = false; if (agent != 0) { - sendGetRequestLH(queryContext, query, agent); + if (sendGetRequestLH(queryContext, query, agent)) + sent = true; } else { // TODO (optimization) only send queries to agents that have the requested class+package for (map::const_iterator iter = agentList.begin(); iter != agentList.end(); iter++) { - sendGetRequestLH(queryContext, query, iter->second.get()); + if (sendGetRequestLH(queryContext, query, iter->second.get())) + sent = true; } } + + if (!sent) { + queryContext->reserve(); + queryContext->release(); + } } -void BrokerProxyImpl::sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent) +bool BrokerProxyImpl::sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent) { + if (query.impl->singleAgent()) { + if (query.impl->agentBank() != agent->getAgentBank()) + return false; + } stringstream key; Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); uint32_t sequence(seqMgr.reserve(queryContext)); @@ -229,6 +241,7 @@ void BrokerProxyImpl::sendGetRequestLH(SequenceContext::Ptr queryContext, const key << "agent.1." << agent->impl->agentBank; sendBufferLH(outBuffer, QMF_EXCHANGE, key.str()); QPID_LOG(trace, "SENT GetQuery seq=" << sequence << " key=" << key.str()); + return true; } string BrokerProxyImpl::encodeMethodArguments(const SchemaMethod* schema, const Value* argmap, Buffer& buffer) diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h index 798a5fdc76..b651b52345 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h @@ -139,7 +139,7 @@ namespace engine { uint32_t agentCount() const; const AgentProxy* getAgent(uint32_t idx) const; void sendQuery(const Query& query, void* context, const AgentProxy* agent); - void sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent); + bool sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent); std::string encodeMethodArguments(const SchemaMethod* schema, const Value* args, qpid::framing::Buffer& buffer); void sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls, const std::string& method, const Value* args, void* context); diff --git a/qpid/cpp/src/qmf/engine/QueryImpl.h b/qpid/cpp/src/qmf/engine/QueryImpl.h index 2c64c6739c..8ebe0d932f 100644 --- a/qpid/cpp/src/qmf/engine/QueryImpl.h +++ b/qpid/cpp/src/qmf/engine/QueryImpl.h @@ -85,6 +85,8 @@ namespace engine { bool getDecreasing() const { return orderDecreasing; } void encode(qpid::framing::Buffer& buffer) const; + bool singleAgent() const { return oid.get() != 0; } + uint32_t agentBank() const { return singleAgent() ? oid->getAgentBank() : 0; } std::string packageName; std::string className; -- cgit v1.2.1 From 2d85ecb5af412f5026ca79aaf8ada830d83dfe87 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Tue, 20 Oct 2009 15:22:19 +0000 Subject: QPID-2126 - Sync the python QMF bindings to the current Ruby QMF bindings implementation Applied patch from Ken Giusti git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@827686 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp | 5 +++++ 1 file changed, 5 insertions(+) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp index 2cd6af10f8..22a65f28ca 100644 --- a/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ConnectionSettingsImpl.cpp @@ -185,6 +185,11 @@ Value ConnectionSettingsImpl::getAttr(const string& key) const return intval; } + if (key == attrSendUserId) { + boolval.setBool(sendUserId); + return boolval; + } + return strval; } -- cgit v1.2.1 From 0207a7cd0c71cec7f488c8c326ce05825073a57a Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Fri, 23 Oct 2009 18:31:10 +0000 Subject: Removed references to broker and agent bank from API, replaced with the more generic (and forward compatible) "key". In the Ruby binding, ensured that ruby objects reference their own copies of the wrapped c++ objects to protect from problems when the c++ objects are deleted out from under the wrappers. Added agent discriminator to the console::objects method. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@829167 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp | 1 + 1 file changed, 1 insertion(+) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp index 0a1769f891..2d955b0c26 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp @@ -741,6 +741,7 @@ bool MethodContext::handleMessage(uint8_t opcode, uint32_t sequence, const strin //================================================================== AgentProxy::AgentProxy(AgentProxyImpl* i) : impl(i) {} +AgentProxy::AgentProxy(const AgentProxy& from) : impl(new AgentProxyImpl(*(from.impl))) {} AgentProxy::~AgentProxy() { delete impl; } const char* AgentProxy::getLabel() const { return impl->getLabel().c_str(); } uint32_t AgentProxy::getBrokerBank() const { return impl->getBrokerBank(); } -- cgit v1.2.1 From b6c399944995c33ac199bec28f81a43c0bb52eaa Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Tue, 3 Nov 2009 21:08:58 +0000 Subject: QMF fixes: - Added immediate notification of agent deletion - Fixed some issues with memory management between the wrapped client and the ruby wrapper git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@832557 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ObjectImpl.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ObjectImpl.h b/qpid/cpp/src/qmf/engine/ObjectImpl.h index ddd20bfea2..6f25867004 100644 --- a/qpid/cpp/src/qmf/engine/ObjectImpl.h +++ b/qpid/cpp/src/qmf/engine/ObjectImpl.h @@ -56,7 +56,7 @@ namespace engine { void destroy(); const ObjectId* getObjectId() const { return objectId.get(); } - void setObjectId(ObjectId* oid) { objectId.reset(oid); } + void setObjectId(ObjectId* oid) { objectId.reset(new ObjectId(*oid)); } const SchemaObjectClass* getClass() const { return objectClass; } Value* getValue(const std::string& key) const; void invokeMethod(const std::string& methodName, const Value* inArgs, void* context) const; -- cgit v1.2.1 From 7eaea4e705f4a0e13c71c4124102b7ae0d93c1c8 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Wed, 2 Dec 2009 04:21:05 +0000 Subject: Fixed intermittent hang seen in Qmf interop tests. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@886047 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ResilientConnection.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp index 9502130288..53524fdbd8 100644 --- a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp +++ b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp @@ -116,9 +116,9 @@ namespace engine { int delayMax; int delayFactor; qpid::sys::Condition cond; - qpid::sys::Thread connThread; deque eventQueue; set sessions; + qpid::sys::Thread connThread; }; } } -- cgit v1.2.1 From 344d932451f98abeb249cd38b49663e438a1ecdc Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Fri, 4 Dec 2009 18:59:29 +0000 Subject: QPID-2246 - QMF - Querying objects using a selector of type ObjectId yields incorrect results git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@887320 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp | 51 +++++++++++++++++++++++++------- qpid/cpp/src/qmf/engine/ObjectIdImpl.h | 16 +++++----- 2 files changed, 49 insertions(+), 18 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp index b08ae2756c..76db6d91f9 100644 --- a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp @@ -24,7 +24,6 @@ using namespace std; using namespace qmf::engine; using qpid::framing::Buffer; - void AgentAttachment::setBanks(uint32_t broker, uint32_t agent) { first = @@ -121,25 +120,57 @@ const string& ObjectIdImpl::asString() const return repr; } -bool ObjectIdImpl::operator==(const ObjectIdImpl& other) const +#define ACTUAL_FIRST (agent == 0 ? first : first | agent->first) +#define ACTUAL_OTHER (other.agent == 0 ? other.first : other.first | other.agent->first) + +uint8_t ObjectIdImpl::getFlags() const { - uint64_t otherFirst = agent == 0 ? other.first : other.first & 0xffff000000000000LL; + return (ACTUAL_FIRST & 0xF000000000000000LL) >> 60; +} - return first == otherFirst && second == other.second; +uint16_t ObjectIdImpl::getSequence() const +{ + return (ACTUAL_FIRST & 0x0FFF000000000000LL) >> 48; } -bool ObjectIdImpl::operator<(const ObjectIdImpl& other) const +uint32_t ObjectIdImpl::getBrokerBank() const { - uint64_t otherFirst = agent == 0 ? other.first : other.first & 0xffff000000000000LL; + return (ACTUAL_FIRST & 0x0000FFFFF0000000LL) >> 28; +} - return (first < otherFirst) || ((first == otherFirst) && (second < other.second)); +uint32_t ObjectIdImpl::getAgentBank() const +{ + return ACTUAL_FIRST & 0x000000000FFFFFFFLL; } -bool ObjectIdImpl::operator>(const ObjectIdImpl& other) const +uint64_t ObjectIdImpl::getObjectNum() const +{ + return second; +} + +uint32_t ObjectIdImpl::getObjectNumHi() const +{ + return (uint32_t) (second >> 32); +} + +uint32_t ObjectIdImpl::getObjectNumLo() const { - uint64_t otherFirst = agent == 0 ? other.first : other.first & 0xffff000000000000LL; + return (uint32_t) (second & 0x00000000FFFFFFFFLL); +} + +bool ObjectIdImpl::operator==(const ObjectIdImpl& other) const +{ + return ACTUAL_FIRST == ACTUAL_OTHER && second == other.second; +} - return (first > otherFirst) || ((first == otherFirst) && (second > other.second)); +bool ObjectIdImpl::operator<(const ObjectIdImpl& other) const +{ + return (ACTUAL_FIRST < ACTUAL_OTHER) || ((ACTUAL_FIRST == ACTUAL_OTHER) && (second < other.second)); +} + +bool ObjectIdImpl::operator>(const ObjectIdImpl& other) const +{ + return (ACTUAL_FIRST > ACTUAL_OTHER) || ((ACTUAL_FIRST == ACTUAL_OTHER) && (second > other.second)); } diff --git a/qpid/cpp/src/qmf/engine/ObjectIdImpl.h b/qpid/cpp/src/qmf/engine/ObjectIdImpl.h index d9871ac217..d70c8efff4 100644 --- a/qpid/cpp/src/qmf/engine/ObjectIdImpl.h +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.h @@ -51,15 +51,15 @@ namespace engine { void encode(qpid::framing::Buffer& buffer) const; void fromString(const std::string& repr); const std::string& asString() const; - uint8_t getFlags() const { return (first & 0xF000000000000000LL) >> 60; } - uint16_t getSequence() const { return (first & 0x0FFF000000000000LL) >> 48; } - uint32_t getBrokerBank() const { return (first & 0x0000FFFFF0000000LL) >> 28; } - uint32_t getAgentBank() const { return first & 0x000000000FFFFFFFLL; } - uint64_t getObjectNum() const { return second; } - uint32_t getObjectNumHi() const { return (uint32_t) (second >> 32); } - uint32_t getObjectNumLo() const { return (uint32_t) (second & 0x00000000FFFFFFFFLL); } + uint8_t getFlags() const; + uint16_t getSequence() const; + uint32_t getBrokerBank() const; + uint32_t getAgentBank() const; + uint64_t getObjectNum() const; + uint32_t getObjectNumHi() const; + uint32_t getObjectNumLo() const; bool isDurable() const { return getSequence() == 0; } - void setValue(uint64_t f, uint64_t s) { first = f; second = s; } + void setValue(uint64_t f, uint64_t s) { first = f; second = s; agent = 0; } bool operator==(const ObjectIdImpl& other) const; bool operator<(const ObjectIdImpl& other) const; -- cgit v1.2.1 From 3bc7e889caef716d8fdd8006f2d53b821ecef6c1 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Fri, 11 Dec 2009 13:52:48 +0000 Subject: QPID-2245 QMF protocol changes will make 0.6 incompatible with 0.5 git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@889619 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/Protocol.cpp | 4 ++-- qpid/cpp/src/qmf/engine/SchemaImpl.cpp | 3 +-- qpid/cpp/src/qmf/engine/ValueImpl.cpp | 2 -- 3 files changed, 3 insertions(+), 6 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/Protocol.cpp b/qpid/cpp/src/qmf/engine/Protocol.cpp index 6061b70a8d..9e5f490604 100644 --- a/qpid/cpp/src/qmf/engine/Protocol.cpp +++ b/qpid/cpp/src/qmf/engine/Protocol.cpp @@ -37,14 +37,14 @@ bool Protocol::checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq) *opcode = buf.getOctet(); *seq = buf.getLong(); - return h1 == 'A' && h2 == 'M' && h3 == '3'; + return h1 == 'A' && h2 == 'M' && h3 == '2'; } void Protocol::encodeHeader(qpid::framing::Buffer& buf, uint8_t opcode, uint32_t seq) { buf.putOctet('A'); buf.putOctet('M'); - buf.putOctet('3'); + buf.putOctet('2'); buf.putOctet(opcode); buf.putLong (seq); } diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp index e366a66826..c37ec34890 100644 --- a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp @@ -341,7 +341,6 @@ SchemaObjectClassImpl::SchemaObjectClassImpl(Buffer& buffer) : hasHash(true), cl buffer.getShortString(name); hash.decode(buffer); - /*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(); @@ -374,7 +373,7 @@ void SchemaObjectClassImpl::encode(Buffer& buffer) const buffer.putShortString(package); buffer.putShortString(name); hash.encode(buffer); - buffer.putOctet(0); // No parent class + //buffer.putOctet(0); // No parent class buffer.putShort((uint16_t) properties.size()); buffer.putShort((uint16_t) statistics.size()); buffer.putShort((uint16_t) methods.size()); diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.cpp b/qpid/cpp/src/qmf/engine/ValueImpl.cpp index f80bdab866..1949d4b946 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ValueImpl.cpp @@ -55,8 +55,6 @@ ValueImpl::ValueImpl(Typecode t, Buffer& buf) : typecode(t) case TYPE_MAP: ft.decode(buf); - // TODO: either update to recursively use QMF types or reduce to int/string/... - // (maybe use another ctor with a FieldValue argument) break; case TYPE_LIST: -- cgit v1.2.1 From 25d63a161d11d504e51ead2f76e761f561503b69 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Tue, 5 Jan 2010 19:45:28 +0000 Subject: Added handling of MAP values for the wrapped QMF interfaces. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@896191 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ValueImpl.cpp | 120 ++++++++++++++++++++++++++++++++-- qpid/cpp/src/qmf/engine/ValueImpl.h | 10 +++ 2 files changed, 125 insertions(+), 5 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.cpp b/qpid/cpp/src/qmf/engine/ValueImpl.cpp index 1949d4b946..b1c027520f 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ValueImpl.cpp @@ -19,16 +19,19 @@ #include "qmf/engine/ValueImpl.h" #include +#include using namespace std; using namespace qmf::engine; using qpid::framing::Buffer; +using qpid::framing::FieldTable; +using qpid::framing::FieldValue; ValueImpl::ValueImpl(Typecode t, Buffer& buf) : typecode(t) { uint64_t first; uint64_t second; - qpid::framing::FieldTable ft; + FieldTable ft; switch (typecode) { case TYPE_UINT8 : value.u32 = (uint32_t) buf.getOctet(); break; @@ -55,6 +58,7 @@ ValueImpl::ValueImpl(Typecode t, Buffer& buf) : typecode(t) case TYPE_MAP: ft.decode(buf); + initMap(ft); break; case TYPE_LIST: @@ -90,8 +94,111 @@ ValueImpl::~ValueImpl() { } +void ValueImpl::initMap(const FieldTable& ft) +{ + for (FieldTable::ValueMap::const_iterator iter = ft.begin(); + iter != ft.end(); iter++) { + const string& name(iter->first); + const FieldValue& fvalue(*iter->second); + uint8_t amqType = fvalue.getType(); + + if (amqType == 0x32) { + Value* subval(new Value(TYPE_UINT64)); + subval->setUint64(fvalue.get()); + insert(name.c_str(), subval); + } else if ((amqType & 0xCF) == 0x02) { + Value* subval(new Value(TYPE_UINT32)); + switch (amqType) { + case 0x02 : subval->setUint(fvalue.get()); break; + case 0x12 : subval->setUint(fvalue.get()); break; + case 0x22 : subval->setUint(fvalue.get()); break; + } + insert(name.c_str(), subval); + } else if ((amqType & 0xCF) == 0x01) { + Value* subval(new Value(TYPE_INT64)); + subval->setInt64(fvalue.get()); + insert(name.c_str(), subval); + } else if (amqType == 0x85 || amqType == 0x95) { + Value* subval(new Value(TYPE_LSTR)); + subval->setString(fvalue.get().c_str()); + insert(name.c_str(), subval); + } else if (amqType == 0x23 || amqType == 0x33) { + Value* subval(new Value(TYPE_DOUBLE)); + subval->setDouble(fvalue.get()); + insert(name.c_str(), subval); + } else { + FieldTable subFt; + bool valid = qpid::framing::getEncodedValue(iter->second, subFt); + if (valid) { + Value* subval(new Value(TYPE_MAP)); + subval->impl->initMap(subFt); + insert(name.c_str(), subval); + } + } + } +} + +void ValueImpl::mapToFieldTable(FieldTable& ft) const +{ + FieldTable subFt; + + for (map::const_iterator iter = mapVal.begin(); + iter != mapVal.end(); iter++) { + const string& name(iter->first); + const Value& subval(iter->second); + + switch (subval.getType()) { + case TYPE_UINT8: + case TYPE_UINT16: + case TYPE_UINT32: + ft.setUInt64(name, (uint64_t) subval.asUint()); + break; + case TYPE_UINT64: + case TYPE_DELTATIME: + ft.setUInt64(name, subval.asUint64()); + break; + case TYPE_SSTR: + case TYPE_LSTR: + ft.setString(name, subval.asString()); + break; + case TYPE_INT64: + case TYPE_ABSTIME: + ft.setInt64(name, subval.asInt64()); + break; + case TYPE_BOOL: + ft.setInt(name, subval.asBool() ? 1 : 0); + break; + case TYPE_FLOAT: + ft.setFloat(name, subval.asFloat()); + break; + case TYPE_DOUBLE: + ft.setDouble(name, subval.asDouble()); + break; + case TYPE_INT8: + case TYPE_INT16: + case TYPE_INT32: + ft.setInt(name, subval.asInt()); + break; + case TYPE_MAP: + subFt.clear(); + subval.impl->mapToFieldTable(subFt); + ft.setTable(name, subFt); + break; + case TYPE_LIST: + case TYPE_ARRAY: + case TYPE_OBJECT: + case TYPE_UUID: + case TYPE_REF: + default: + break; + } + } + } + void ValueImpl::encode(Buffer& buf) const { + FieldTable ft; + switch (typecode) { case TYPE_UINT8 : buf.putOctet((uint8_t) value.u32); break; case TYPE_UINT16 : buf.putShort((uint16_t) value.u32); break; @@ -110,7 +217,10 @@ void ValueImpl::encode(Buffer& buf) const case TYPE_INT64 : buf.putLongLong(value.s64); break; case TYPE_UUID : buf.putBin128(value.uuidVal); break; case TYPE_REF : refVal.impl->encode(buf); break; - case TYPE_MAP: // TODO + case TYPE_MAP: + mapToFieldTable(ft); + ft.encode(buf); + break; case TYPE_LIST: case TYPE_ARRAY: case TYPE_OBJECT: @@ -127,7 +237,7 @@ bool ValueImpl::keyInMap(const char* key) const Value* ValueImpl::byKey(const char* key) { if (keyInMap(key)) { - map::iterator iter = mapVal.find(key); + map::iterator iter = mapVal.find(key); if (iter != mapVal.end()) return &iter->second; } @@ -137,7 +247,7 @@ Value* ValueImpl::byKey(const char* key) const Value* ValueImpl::byKey(const char* key) const { if (keyInMap(key)) { - map::const_iterator iter = mapVal.find(key); + map::const_iterator iter = mapVal.find(key); if (iter != mapVal.end()) return &iter->second; } @@ -157,7 +267,7 @@ void ValueImpl::insert(const char* key, Value* val) const char* ValueImpl::key(uint32_t idx) const { - map::const_iterator iter = mapVal.begin(); + map::const_iterator iter = mapVal.begin(); for (uint32_t i = 0; i < idx; i++) { if (iter == mapVal.end()) break; diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.h b/qpid/cpp/src/qmf/engine/ValueImpl.h index b6adae5d93..84b0e768e6 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.h +++ b/qpid/cpp/src/qmf/engine/ValueImpl.h @@ -30,6 +30,12 @@ #include #include +namespace qpid { +namespace framing { + class FieldTable; +} +} + namespace qmf { namespace engine { @@ -142,6 +148,10 @@ namespace engine { Value* arrayItem(uint32_t idx); void appendToArray(Value* val); void deleteArrayItem(uint32_t idx); + + private: + void mapToFieldTable(qpid::framing::FieldTable& ft) const; + void initMap(const qpid::framing::FieldTable& ft); }; } } -- cgit v1.2.1 From fd64f22be60f12e03df8974b547cd9b3af331601 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Wed, 13 Jan 2010 11:57:19 +0000 Subject: Added raise_event support to the Ruby and Python wrapped agent APIs. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@898727 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/Agent.cpp | 13 ++- qpid/cpp/src/qmf/engine/EventImpl.cpp | 106 ++++++++++++++++++++++++ qpid/cpp/src/qmf/engine/EventImpl.h | 53 ++++++++++++ qpid/cpp/src/qmf/engine/ResilientConnection.cpp | 19 +++-- qpid/cpp/src/qmf/engine/SchemaImpl.cpp | 4 +- qpid/cpp/src/qmf/engine/SchemaImpl.h | 6 +- 6 files changed, 191 insertions(+), 10 deletions(-) create mode 100644 qpid/cpp/src/qmf/engine/EventImpl.cpp create mode 100644 qpid/cpp/src/qmf/engine/EventImpl.h (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/Agent.cpp b/qpid/cpp/src/qmf/engine/Agent.cpp index c5d1bff2e0..fe9b84c565 100644 --- a/qpid/cpp/src/qmf/engine/Agent.cpp +++ b/qpid/cpp/src/qmf/engine/Agent.cpp @@ -21,6 +21,7 @@ #include "qmf/engine/MessageImpl.h" #include "qmf/engine/SchemaImpl.h" #include "qmf/engine/Typecode.h" +#include "qmf/engine/EventImpl.h" #include "qmf/engine/ObjectImpl.h" #include "qmf/engine/ObjectIdImpl.h" #include "qmf/engine/QueryImpl.h" @@ -476,9 +477,19 @@ const ObjectId* AgentImpl::allocObjectId(uint32_t persistIdLo, uint32_t persistI return allocObjectId(((uint64_t) persistIdHi) << 32 | (uint64_t) persistIdLo); } -void AgentImpl::raiseEvent(Event&) +void AgentImpl::raiseEvent(Event& event) { Mutex::ScopedLock _lock(lock); + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + Protocol::encodeHeader(buffer, Protocol::OP_EVENT_INDICATION); + + event.impl->encodeSchemaKey(buffer); + buffer.putLongLong(uint64_t(Duration(now()))); + event.impl->encode(buffer); + string key(event.impl->getRoutingKey(assignedBrokerBank, assignedAgentBank)); + + sendBufferLH(buffer, QMF_EXCHANGE, key); + QPID_LOG(trace, "SENT EventIndication"); } AgentEventImpl::Ptr AgentImpl::eventDeclareQueue(const string& name) diff --git a/qpid/cpp/src/qmf/engine/EventImpl.cpp b/qpid/cpp/src/qmf/engine/EventImpl.cpp new file mode 100644 index 0000000000..6bdda9321e --- /dev/null +++ b/qpid/cpp/src/qmf/engine/EventImpl.cpp @@ -0,0 +1,106 @@ +/* + * 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 +#include + +using namespace std; +using namespace qmf::engine; +using qpid::framing::Buffer; + +EventImpl::EventImpl(const SchemaEventClass* type) : eventClass(type) +{ + int argCount = eventClass->getArgumentCount(); + int idx; + + for (idx = 0; idx < argCount; idx++) { + const SchemaArgument* arg = eventClass->getArgument(idx); + arguments[arg->getName()] = ValuePtr(new Value(arg->getType())); + } +} + + +EventImpl::EventImpl(const SchemaEventClass* type, Buffer&) : + eventClass(type) +{ +} + + +Event* EventImpl::factory(const SchemaEventClass* type, Buffer& buffer) +{ + EventImpl* impl(new EventImpl(type, buffer)); + return new Event(impl); +} + + +Value* EventImpl::getValue(const char* key) const +{ + map::const_iterator iter; + + iter = arguments.find(key); + if (iter != arguments.end()) + return iter->second.get(); + + return 0; +} + + +void EventImpl::encodeSchemaKey(Buffer& buffer) const +{ + buffer.putShortString(eventClass->getClassKey()->getPackageName()); + buffer.putShortString(eventClass->getClassKey()->getClassName()); + buffer.putBin128(const_cast(eventClass->getClassKey()->getHash())); +} + + +void EventImpl::encode(Buffer& buffer) const +{ + buffer.putOctet((uint8_t) eventClass->getSeverity()); + + int argCount = eventClass->getArgumentCount(); + for (int idx = 0; idx < argCount; idx++) { + const SchemaArgument* arg = eventClass->getArgument(idx); + ValuePtr value = arguments[arg->getName()]; + value->impl->encode(buffer); + } +} + + +string EventImpl::getRoutingKey(uint32_t brokerBank, uint32_t agentBank) const +{ + stringstream key; + + key << "console.event." << brokerBank << "." << agentBank << "." << + eventClass->getClassKey()->getPackageName() << "." << + eventClass->getClassKey()->getClassName(); + return key.str(); +} + + +//================================================================== +// Wrappers +//================================================================== + +Event::Event(const SchemaEventClass* type) : impl(new EventImpl(type)) {} +Event::Event(EventImpl* i) : impl(i) {} +Event::Event(const Event& from) : impl(new EventImpl(*(from.impl))) {} +Event::~Event() { delete impl; } +const SchemaEventClass* Event::getClass() const { return impl->getClass(); } +Value* Event::getValue(const char* key) const { return impl->getValue(key); } + diff --git a/qpid/cpp/src/qmf/engine/EventImpl.h b/qpid/cpp/src/qmf/engine/EventImpl.h new file mode 100644 index 0000000000..dfdf64e848 --- /dev/null +++ b/qpid/cpp/src/qmf/engine/EventImpl.h @@ -0,0 +1,53 @@ +#ifndef _QmfEngineEventImpl_ +#define _QmfEngineEventImpl_ + +/* + * 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 +#include +#include +#include +#include + +namespace qmf { +namespace engine { + + struct EventImpl { + typedef boost::shared_ptr ValuePtr; + const SchemaEventClass* eventClass; + mutable std::map arguments; + + EventImpl(const SchemaEventClass* type); + EventImpl(const SchemaEventClass* type, qpid::framing::Buffer& buffer); + static Event* factory(const SchemaEventClass* type, qpid::framing::Buffer& buffer); + + const SchemaEventClass* getClass() const { return eventClass; } + Value* getValue(const char* key) const; + + void encodeSchemaKey(qpid::framing::Buffer& buffer) const; + void encode(qpid::framing::Buffer& buffer) const; + std::string getRoutingKey(uint32_t brokerBank, uint32_t agentBank) const; + }; + +} +} + +#endif + diff --git a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp index 53524fdbd8..9c19e4d460 100644 --- a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp +++ b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp @@ -39,6 +39,8 @@ #include #include #include +#include +#include using namespace std; using namespace qmf::engine; @@ -330,6 +332,10 @@ void ResilientConnectionImpl::unbind(SessionHandle handle, void ResilientConnectionImpl::setNotifyFd(int fd) { notifyFd = fd; + if (notifyFd > 0) { + int original = fcntl(notifyFd, F_GETFL); + fcntl(notifyFd, F_SETFL, O_NONBLOCK | original); + } } void ResilientConnectionImpl::run() @@ -403,13 +409,16 @@ void ResilientConnectionImpl::EnqueueEvent(ResilientConnectionEvent::EventKind k const MessageImpl& message, const string& errorText) { - Mutex::ScopedLock _lock(lock); - ResilientConnectionEventImpl event(kind, message); + { + Mutex::ScopedLock _lock(lock); + ResilientConnectionEventImpl event(kind, message); - event.sessionContext = sessionContext; - event.errorText = errorText; + event.sessionContext = sessionContext; + event.errorText = errorText; + + eventQueue.push_back(event); + } - eventQueue.push_back(event); if (notifyFd != -1) { int unused_ret; //Suppress warnings about ignoring return value. diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp index c37ec34890..249a08ba7f 100644 --- a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp @@ -459,7 +459,6 @@ SchemaEventClassImpl::SchemaEventClassImpl(Buffer& buffer) : hasHash(true), clas buffer.getShortString(package); buffer.getShortString(name); hash.decode(buffer); - buffer.putOctet(0); // No parent class uint16_t argCount = buffer.getShort(); @@ -598,13 +597,14 @@ const SchemaProperty* SchemaObjectClass::getProperty(int idx) const { return imp 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(package, name)) {} +SchemaEventClass::SchemaEventClass(const char* package, const char* name, Severity s) : impl(new SchemaEventClassImpl(package, name, s)) {} SchemaEventClass::SchemaEventClass(SchemaEventClassImpl* i) : impl(i) {} SchemaEventClass::SchemaEventClass(const SchemaEventClass& from) : impl(new SchemaEventClassImpl(*(from.impl))) {} 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(); } +Severity SchemaEventClass::getSeverity() const { return impl->getSeverity(); } 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/engine/SchemaImpl.h b/qpid/cpp/src/qmf/engine/SchemaImpl.h index af3a1d98e4..7be757ee8d 100644 --- a/qpid/cpp/src/qmf/engine/SchemaImpl.h +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.h @@ -201,10 +201,11 @@ namespace engine { mutable bool hasHash; std::auto_ptr classKey; std::string description; + Severity severity; std::vector arguments; - SchemaEventClassImpl(const char* p, const char* n) : - package(p), name(n), hasHash(false), classKey(SchemaClassKeyImpl::factory(package, name, hash)) {} + SchemaEventClassImpl(const char* p, const char* n, Severity sev) : + package(p), name(n), hasHash(false), classKey(SchemaClassKeyImpl::factory(package, name, hash)), severity(sev) {} SchemaEventClassImpl(qpid::framing::Buffer& buffer); static SchemaEventClass* factory(qpid::framing::Buffer& buffer); @@ -213,6 +214,7 @@ namespace engine { void setDesc(const char* desc) { description = desc; } const SchemaClassKey* getClassKey() const; + Severity getSeverity() const { return severity; } int getArgumentCount() const { return arguments.size(); } const SchemaArgument* getArgument(int idx) const; }; -- cgit v1.2.1 From 9523fa332a8fd0f713c3c12c941b061316597d84 Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Mon, 8 Feb 2010 21:33:54 +0000 Subject: QPID-2396: add assignment operator to ObjectId class. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@907808 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp | 10 ++++++++++ 1 file changed, 10 insertions(+) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp index 76db6d91f9..670ee385a3 100644 --- a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp @@ -196,4 +196,14 @@ bool ObjectId::operator<(const ObjectId& other) const { return *impl < *other.im 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); } +ObjectId& ObjectId::operator=(const ObjectId& other) { + ObjectIdImpl *old; + if (this != &other) { + old = impl; + impl = new ObjectIdImpl(*(other.impl)); + if (old) + delete old; + } + return *this; +} -- cgit v1.2.1 From 6febb06e1476898cd18d59cc0edf212af3d1efb8 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Fri, 12 Feb 2010 18:23:47 +0000 Subject: QPID-2328 - Applied patch from Ian Main git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@909548 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ResilientConnection.cpp | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp index 9c19e4d460..ab65b8d768 100644 --- a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp +++ b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp @@ -96,6 +96,7 @@ namespace engine { void bind(SessionHandle handle, char* exchange, char* queue, char* key); void unbind(SessionHandle handle, char* exchange, char* queue, char* key); void setNotifyFd(int fd); + void notify(); void run(); void failure(); @@ -329,6 +330,16 @@ void ResilientConnectionImpl::unbind(SessionHandle handle, sess->session.exchangeUnbind(client::arg::exchange=exchange, client::arg::queue=queue, client::arg::bindingKey=key); } +void ResilientConnectionImpl::notify() +{ + if (notifyFd != -1) + { + int unused_ret; //Suppress warnings about ignoring return value. + unused_ret = ::write(notifyFd, ".", 1); + } +} + + void ResilientConnectionImpl::setNotifyFd(int fd) { notifyFd = fd; @@ -496,3 +507,8 @@ void ResilientConnection::setNotifyFd(int fd) impl->setNotifyFd(fd); } +void ResilientConnection::notify() +{ + impl->notify(); +} + -- cgit v1.2.1 From 588219ac990912cecf77c90eaa77fba3dcdd5668 Mon Sep 17 00:00:00 2001 From: Andrew Stitcher Date: Thu, 25 Feb 2010 20:53:08 +0000 Subject: Removed unecessary include of Msg.h in Exception.h git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@916453 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/EventImpl.cpp | 2 ++ qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp | 1 + qpid/cpp/src/qmf/engine/SchemaImpl.cpp | 1 + qpid/cpp/src/qmf/engine/SchemaImpl.h | 4 +++- 4 files changed, 7 insertions(+), 1 deletion(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/EventImpl.cpp b/qpid/cpp/src/qmf/engine/EventImpl.cpp index 6bdda9321e..3509220b0c 100644 --- a/qpid/cpp/src/qmf/engine/EventImpl.cpp +++ b/qpid/cpp/src/qmf/engine/EventImpl.cpp @@ -20,6 +20,8 @@ #include #include +#include + using namespace std; using namespace qmf::engine; using qpid::framing::Buffer; diff --git a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp index 670ee385a3..9216f7bac0 100644 --- a/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ObjectIdImpl.cpp @@ -19,6 +19,7 @@ #include "qmf/engine/ObjectIdImpl.h" #include +#include using namespace std; using namespace qmf::engine; diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp index 249a08ba7f..e0948a9911 100644 --- a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp @@ -24,6 +24,7 @@ #include #include #include +#include using namespace std; using namespace qmf::engine; diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.h b/qpid/cpp/src/qmf/engine/SchemaImpl.h index 7be757ee8d..8b079a5ec6 100644 --- a/qpid/cpp/src/qmf/engine/SchemaImpl.h +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.h @@ -21,9 +21,11 @@ */ #include "qmf/engine/Schema.h" +#include "qpid/framing/Buffer.h" + #include #include -#include +#include namespace qmf { namespace engine { -- cgit v1.2.1 From feed6024a260189ed210f7a04049a44d2a2e9937 Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Tue, 30 Mar 2010 20:09:59 +0000 Subject: add support for QMF TYPE_LIST in engine, ruby and python wrappers git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@929244 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ValueImpl.cpp | 167 ++++++++++++++++++++++++++++++---- qpid/cpp/src/qmf/engine/ValueImpl.h | 13 ++- 2 files changed, 158 insertions(+), 22 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.cpp b/qpid/cpp/src/qmf/engine/ValueImpl.cpp index b1c027520f..72c68c420b 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ValueImpl.cpp @@ -18,20 +18,23 @@ */ #include "qmf/engine/ValueImpl.h" -#include #include +#include +#include using namespace std; using namespace qmf::engine; -using qpid::framing::Buffer; -using qpid::framing::FieldTable; -using qpid::framing::FieldValue; +//using qpid::framing::Buffer; +//using qpid::framing::FieldTable; +//using qpid::framing::FieldValue; +using namespace qpid::framing; ValueImpl::ValueImpl(Typecode t, Buffer& buf) : typecode(t) { uint64_t first; uint64_t second; FieldTable ft; + List fl; switch (typecode) { case TYPE_UINT8 : value.u32 = (uint32_t) buf.getOctet(); break; @@ -62,6 +65,10 @@ ValueImpl::ValueImpl(Typecode t, Buffer& buf) : typecode(t) break; case TYPE_LIST: + fl.decode(buf); + initList(fl); + break; + case TYPE_ARRAY: case TYPE_OBJECT: default: @@ -126,7 +133,7 @@ void ValueImpl::initMap(const FieldTable& ft) Value* subval(new Value(TYPE_DOUBLE)); subval->setDouble(fvalue.get()); insert(name.c_str(), subval); - } else { + } else if (amqType == 0xa8) { FieldTable subFt; bool valid = qpid::framing::getEncodedValue(iter->second, subFt); if (valid) { @@ -134,6 +141,14 @@ void ValueImpl::initMap(const FieldTable& ft) subval->impl->initMap(subFt); insert(name.c_str(), subval); } + } else if (amqType == 0xa9) { + List subList; + bool valid = qpid::framing::getEncodedValue(iter->second, subList); + if (valid) { + Value* subval(new Value(TYPE_LIST)); + subval->impl->initList(subList); + insert(name.c_str(), subval); + } } } } @@ -185,6 +200,14 @@ void ValueImpl::mapToFieldTable(FieldTable& ft) const ft.setTable(name, subFt); break; case TYPE_LIST: + { + List subList; + subval.impl->listToFramingList(subList); + ft.set(name, + ::qpid::framing::FieldTable::ValuePtr( + new ListValue( + subList))); + } break; case TYPE_ARRAY: case TYPE_OBJECT: case TYPE_UUID: @@ -195,9 +218,126 @@ void ValueImpl::mapToFieldTable(FieldTable& ft) const } } + +void ValueImpl::initList(const List& fl) +{ + for (List::const_iterator iter = fl.begin(); + iter != fl.end(); iter++) { + const FieldValue& fvalue(*iter->get()); + uint8_t amqType = fvalue.getType(); + + if (amqType == 0x32) { + Value* subval(new Value(TYPE_UINT64)); + subval->setUint64(fvalue.get()); + appendToList(subval); + } else if ((amqType & 0xCF) == 0x02) { + Value* subval(new Value(TYPE_UINT32)); + switch (amqType) { + case 0x02 : subval->setUint(fvalue.get()); break; + case 0x12 : subval->setUint(fvalue.get()); break; + case 0x22 : subval->setUint(fvalue.get()); break; + } + appendToList(subval); + } else if ((amqType & 0xCF) == 0x01) { + Value* subval(new Value(TYPE_INT64)); + subval->setInt64(fvalue.get()); + appendToList(subval); + } else if (amqType == 0x85 || amqType == 0x95) { + Value* subval(new Value(TYPE_LSTR)); + subval->setString(fvalue.get().c_str()); + appendToList(subval); + } else if (amqType == 0x23 || amqType == 0x33) { + Value* subval(new Value(TYPE_DOUBLE)); + subval->setDouble(fvalue.get()); + appendToList(subval); + } else if (amqType == 0xa8) { + FieldTable subFt; + bool valid = qpid::framing::getEncodedValue(*iter, subFt); + if (valid) { + Value* subval(new Value(TYPE_MAP)); + subval->impl->initMap(subFt); + appendToList(subval); + } + } else if (amqType == 0xa9) { + List subList; + bool valid = qpid::framing::getEncodedValue(*iter, subList); + if (valid) { + Value *subVal(new Value(TYPE_LIST)); + subVal->impl->initList(subList); + appendToList(subVal); + } + } + } +} + +void ValueImpl::listToFramingList(List& fl) const +{ + for (vector::const_iterator iter = vectorVal.begin(); + iter != vectorVal.end(); iter++) { + const Value& subval(*iter); + + switch (subval.getType()) { + case TYPE_UINT8: + case TYPE_UINT16: + case TYPE_UINT32: + fl.push_back(List::ValuePtr(new Unsigned64Value((uint64_t) subval.asUint()))); + break; + case TYPE_UINT64: + case TYPE_DELTATIME: + fl.push_back(List::ValuePtr(new Unsigned64Value(subval.asUint64()))); + break; + case TYPE_SSTR: + case TYPE_LSTR: + fl.push_back(List::ValuePtr(new Str16Value(subval.asString()))); + break; + case TYPE_INT64: + case TYPE_ABSTIME: + fl.push_back(List::ValuePtr(new Integer64Value(subval.asInt64()))); + break; + case TYPE_BOOL: + fl.push_back(List::ValuePtr(new IntegerValue(subval.asBool() ? 1 : 0))); + break; + case TYPE_FLOAT: + fl.push_back(List::ValuePtr(new FloatValue(subval.asFloat()))); + break; + case TYPE_DOUBLE: + fl.push_back(List::ValuePtr(new DoubleValue(subval.asDouble()))); + break; + case TYPE_INT8: + case TYPE_INT16: + case TYPE_INT32: + fl.push_back(List::ValuePtr(new IntegerValue(subval.asInt()))); + break; + case TYPE_MAP: + { + FieldTable subFt; + subval.impl->mapToFieldTable(subFt); + fl.push_back(List::ValuePtr(new FieldTableValue(subFt))); + + } break; + case TYPE_LIST: + { + List subList; + subval.impl->listToFramingList(subList); + fl.push_back(List::ValuePtr(new ListValue(subList))); + } break; + + case TYPE_ARRAY: + case TYPE_OBJECT: + case TYPE_UUID: + case TYPE_REF: + default: + break; + } + } + } + + + void ValueImpl::encode(Buffer& buf) const { FieldTable ft; + List fl; switch (typecode) { case TYPE_UINT8 : buf.putOctet((uint8_t) value.u32); break; @@ -222,6 +362,10 @@ void ValueImpl::encode(Buffer& buf) const ft.encode(buf); break; case TYPE_LIST: + listToFramingList(fl); + fl.encode(buf); + break; + case TYPE_ARRAY: case TYPE_OBJECT: default: @@ -280,19 +424,6 @@ const char* ValueImpl::key(uint32_t idx) const return iter->first.c_str(); } -Value* ValueImpl::listItem(uint32_t) -{ - return 0; -} - -void ValueImpl::appendToList(Value*) -{ -} - -void ValueImpl::deleteListItem(uint32_t) -{ -} - Value* ValueImpl::arrayItem(uint32_t) { return 0; diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.h b/qpid/cpp/src/qmf/engine/ValueImpl.h index 84b0e768e6..3b535834fd 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.h +++ b/qpid/cpp/src/qmf/engine/ValueImpl.h @@ -33,6 +33,7 @@ namespace qpid { namespace framing { class FieldTable; + class List; } } @@ -138,13 +139,14 @@ namespace engine { bool isList() const { return typecode == TYPE_LIST; } uint32_t listItemCount() const { return vectorVal.size(); } - Value* listItem(uint32_t idx); - void appendToList(Value* val); - void deleteListItem(uint32_t idx); + Value* listItem(uint32_t idx) { return idx < listItemCount() ? &vectorVal[idx] : 0; } + const Value* listItem(uint32_t idx) const { return idx < listItemCount() ? &vectorVal[idx] : 0; } + void appendToList(Value* val) { vectorVal.push_back(*val); } + void deleteListItem(uint32_t idx) { if (idx < listItemCount()) vectorVal.erase(vectorVal.begin()+idx); } bool isArray() const { return typecode == TYPE_ARRAY; } Typecode arrayType() const { return arrayTypecode; } - uint32_t arrayItemCount() const { return vectorVal.size(); } + uint32_t arrayItemCount() const { return 0; } Value* arrayItem(uint32_t idx); void appendToArray(Value* val); void deleteArrayItem(uint32_t idx); @@ -152,6 +154,9 @@ namespace engine { private: void mapToFieldTable(qpid::framing::FieldTable& ft) const; void initMap(const qpid::framing::FieldTable& ft); + + void listToFramingList(qpid::framing::List& fl) const; + void initList(const qpid::framing::List& fl); }; } } -- cgit v1.2.1 From b4738c5e99ee9e751f45deb177270bc0bb0a775e Mon Sep 17 00:00:00 2001 From: Andrew Stitcher Date: Fri, 23 Apr 2010 03:59:52 +0000 Subject: QPID-1904: Ensure that all timestamp uses are correctly relative to 1/1/1970 epoch. - Removed the hacky way to access the internal time value in AbsTime now that there is a defined AbsTime value EPOCH. - Changed all the code to use Duration(EPOCH, abtime) git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@937147 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/Agent.cpp | 4 ++-- qpid/cpp/src/qmf/engine/ObjectImpl.cpp | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/Agent.cpp b/qpid/cpp/src/qmf/engine/Agent.cpp index fe9b84c565..7667213e5e 100644 --- a/qpid/cpp/src/qmf/engine/Agent.cpp +++ b/qpid/cpp/src/qmf/engine/Agent.cpp @@ -348,7 +348,7 @@ void AgentImpl::heartbeat() Buffer buffer(outputBuffer, MA_BUFFER_SIZE); Protocol::encodeHeader(buffer, Protocol::OP_HEARTBEAT_INDICATION); - buffer.putLongLong(uint64_t(Duration(now()))); + buffer.putLongLong(uint64_t(Duration(EPOCH, now()))); stringstream key; key << "console.heartbeat." << assignedBrokerBank << "." << assignedAgentBank; sendBufferLH(buffer, QMF_EXCHANGE, key.str()); @@ -484,7 +484,7 @@ void AgentImpl::raiseEvent(Event& event) Protocol::encodeHeader(buffer, Protocol::OP_EVENT_INDICATION); event.impl->encodeSchemaKey(buffer); - buffer.putLongLong(uint64_t(Duration(now()))); + buffer.putLongLong(uint64_t(Duration(EPOCH, now()))); event.impl->encode(buffer); string key(event.impl->getRoutingKey(assignedBrokerBank, assignedAgentBank)); diff --git a/qpid/cpp/src/qmf/engine/ObjectImpl.cpp b/qpid/cpp/src/qmf/engine/ObjectImpl.cpp index cae0e0da68..45925cb804 100644 --- a/qpid/cpp/src/qmf/engine/ObjectImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ObjectImpl.cpp @@ -27,7 +27,7 @@ using namespace qmf::engine; using namespace qpid::sys; using qpid::framing::Buffer; -ObjectImpl::ObjectImpl(const SchemaObjectClass* type) : objectClass(type), broker(0), createTime(uint64_t(Duration(now()))), destroyTime(0), lastUpdatedTime(createTime) +ObjectImpl::ObjectImpl(const SchemaObjectClass* type) : objectClass(type), broker(0), createTime(uint64_t(Duration(EPOCH, now()))), destroyTime(0), lastUpdatedTime(createTime) { int propCount = objectClass->getPropertyCount(); int statCount = objectClass->getStatisticCount(); @@ -93,7 +93,7 @@ ObjectImpl::~ObjectImpl() void ObjectImpl::destroy() { - destroyTime = uint64_t(Duration(now())); + destroyTime = uint64_t(Duration(EPOCH, now())); // TODO - flag deletion } -- cgit v1.2.1 From 944b2eb07f51118a37bacf3124dc19c0b7d85ca8 Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Fri, 23 Apr 2010 20:27:27 +0000 Subject: QPID-2546: implement event support for ruby wrapper. Python pending... git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@937506 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp | 14 ++++++++++++-- qpid/cpp/src/qmf/engine/BrokerProxyImpl.h | 1 + qpid/cpp/src/qmf/engine/ConsoleImpl.cpp | 11 ++++++++++- qpid/cpp/src/qmf/engine/ConsoleImpl.h | 5 +++-- qpid/cpp/src/qmf/engine/EventImpl.cpp | 18 +++++++++++++++--- qpid/cpp/src/qmf/engine/EventImpl.h | 4 ++++ 6 files changed, 45 insertions(+), 8 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp index 2d955b0c26..b99cb414dc 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp @@ -445,9 +445,19 @@ void BrokerProxyImpl::handleHeartbeatIndication(Buffer& inBuffer, uint32_t seq, QPID_LOG(trace, "RCVD HeartbeatIndication seq=" << seq << " agentBank=" << agentBank); } -void BrokerProxyImpl::handleEventIndication(Buffer& /*inBuffer*/, uint32_t /*seq*/) +void BrokerProxyImpl::handleEventIndication(Buffer& inBuffer, uint32_t seq) { - // TODO + auto_ptr classKey(SchemaClassKeyImpl::factory(inBuffer)); + const SchemaEventClass *schema = console.impl->getEventClass(classKey.get()); + if (schema == 0) { + QPID_LOG(trace, "No Schema Found for EventIndication. seq=" << seq << " key=" << classKey->impl->str()); + return; + } + + EventPtr eptr(EventImpl::factory(schema, inBuffer)); + + console.impl->eventEventReceived(eptr); + QPID_LOG(trace, "RCVD EventIndication seq=" << seq << " key=" << classKey->impl->str()); } void BrokerProxyImpl::handleSchemaResponse(Buffer& inBuffer, uint32_t seq) diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h index b651b52345..494da5e239 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h @@ -22,6 +22,7 @@ #include "qmf/engine/Console.h" #include "qmf/engine/ObjectImpl.h" +#include "qmf/engine/EventImpl.h" #include "qmf/engine/SchemaImpl.h" #include "qmf/engine/ValueImpl.h" #include "qmf/engine/QueryImpl.h" diff --git a/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp b/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp index c2d1f51f2b..1b66d9e81f 100644 --- a/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp @@ -60,7 +60,7 @@ ConsoleEvent ConsoleEventImpl::copy() item.classKey = classKey; item.object = object.get(); item.context = context; - item.event = event; + item.event = event.get(); item.timestamp = timestamp; item.hasProps = hasProps; item.hasStats = hasStats; @@ -392,6 +392,15 @@ void ConsoleImpl::eventAgentHeartbeat(boost::shared_ptr agent, uint6 eventQueue.push_back(event); } + +void ConsoleImpl::eventEventReceived(EventPtr event) +{ + ConsoleEventImpl::Ptr console_event(new ConsoleEventImpl(ConsoleEvent::EVENT_RECEIVED)); + console_event->event = event; + Mutex::ScopedLock _lock(lock); + eventQueue.push_back(console_event); +} + //================================================================== // Wrappers //================================================================== diff --git a/qpid/cpp/src/qmf/engine/ConsoleImpl.h b/qpid/cpp/src/qmf/engine/ConsoleImpl.h index 8f99c5e6b9..ace47ec87c 100644 --- a/qpid/cpp/src/qmf/engine/ConsoleImpl.h +++ b/qpid/cpp/src/qmf/engine/ConsoleImpl.h @@ -59,13 +59,13 @@ namespace engine { const SchemaClassKey* classKey; boost::shared_ptr object; void* context; - Event* event; + boost::shared_ptr event; uint64_t timestamp; bool hasProps; bool hasStats; ConsoleEventImpl(ConsoleEvent::EventKind k) : - kind(k), classKey(0), context(0), event(0), timestamp(0) {} + kind(k), classKey(0), context(0), timestamp(0) {} ~ConsoleEventImpl() {} ConsoleEvent copy(); }; @@ -137,6 +137,7 @@ namespace engine { void eventNewClass(const SchemaClassKey* key); void eventObjectUpdate(ObjectPtr object, bool prop, bool stat); void eventAgentHeartbeat(boost::shared_ptr agent, uint64_t timestamp); + void eventEventReceived(boost::shared_ptr event); }; } } diff --git a/qpid/cpp/src/qmf/engine/EventImpl.cpp b/qpid/cpp/src/qmf/engine/EventImpl.cpp index 3509220b0c..4b034e8e83 100644 --- a/qpid/cpp/src/qmf/engine/EventImpl.cpp +++ b/qpid/cpp/src/qmf/engine/EventImpl.cpp @@ -26,7 +26,7 @@ using namespace std; using namespace qmf::engine; using qpid::framing::Buffer; -EventImpl::EventImpl(const SchemaEventClass* type) : eventClass(type) +EventImpl::EventImpl(const SchemaEventClass* type) : eventClass(type), timestamp(0), severity(0) { int argCount = eventClass->getArgumentCount(); int idx; @@ -38,9 +38,21 @@ EventImpl::EventImpl(const SchemaEventClass* type) : eventClass(type) } -EventImpl::EventImpl(const SchemaEventClass* type, Buffer&) : - eventClass(type) +EventImpl::EventImpl(const SchemaEventClass* type, Buffer& buffer) : + eventClass(type), timestamp(0), severity(0) { + int argCount = eventClass->getArgumentCount(); + int idx; + + timestamp = buffer.getLongLong(); + severity = buffer.getOctet(); + + for (idx = 0; idx < argCount; idx++) + { + const SchemaArgument *arg = eventClass->getArgument(idx); + Value* pval = ValueImpl::factory(arg->getType(), buffer); + arguments[arg->getName()] = ValuePtr(pval); + } } diff --git a/qpid/cpp/src/qmf/engine/EventImpl.h b/qpid/cpp/src/qmf/engine/EventImpl.h index dfdf64e848..4046e71ef9 100644 --- a/qpid/cpp/src/qmf/engine/EventImpl.h +++ b/qpid/cpp/src/qmf/engine/EventImpl.h @@ -29,9 +29,13 @@ namespace qmf { namespace engine { + typedef boost::shared_ptr EventPtr; + struct EventImpl { typedef boost::shared_ptr ValuePtr; const SchemaEventClass* eventClass; + uint64_t timestamp; + uint8_t severity; mutable std::map arguments; EventImpl(const SchemaEventClass* type); -- cgit v1.2.1 From 07c5b6e839e23e72d6d107cc7ed979db412aa53d Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Tue, 27 Apr 2010 15:19:14 +0000 Subject: QPID-2556: fix conversion of signed integers in maps and lists. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@938506 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ValueImpl.cpp | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.cpp b/qpid/cpp/src/qmf/engine/ValueImpl.cpp index 72c68c420b..c58c28e166 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ValueImpl.cpp @@ -121,10 +121,14 @@ void ValueImpl::initMap(const FieldTable& ft) case 0x22 : subval->setUint(fvalue.get()); break; } insert(name.c_str(), subval); - } else if ((amqType & 0xCF) == 0x01) { + } else if (amqType == 0x31) { // int64 Value* subval(new Value(TYPE_INT64)); subval->setInt64(fvalue.get()); insert(name.c_str(), subval); + } else if ((amqType & 0xCF) == 0x01) { // 0x01:int8, 0x11:int16, 0x21:int21 + Value* subval(new Value(TYPE_INT32)); + subval->setInt((int32_t)fvalue.get()); + insert(name.c_str(), subval); } else if (amqType == 0x85 || amqType == 0x95) { Value* subval(new Value(TYPE_LSTR)); subval->setString(fvalue.get().c_str()); @@ -233,15 +237,19 @@ void ValueImpl::initList(const List& fl) } else if ((amqType & 0xCF) == 0x02) { Value* subval(new Value(TYPE_UINT32)); switch (amqType) { - case 0x02 : subval->setUint(fvalue.get()); break; - case 0x12 : subval->setUint(fvalue.get()); break; - case 0x22 : subval->setUint(fvalue.get()); break; + case 0x02 : subval->setUint(fvalue.get()); break; // uint8 + case 0x12 : subval->setUint(fvalue.get()); break; // uint16 + case 0x22 : subval->setUint(fvalue.get()); break; // uint32 } appendToList(subval); - } else if ((amqType & 0xCF) == 0x01) { + } else if (amqType == 0x31) { // int64 Value* subval(new Value(TYPE_INT64)); subval->setInt64(fvalue.get()); appendToList(subval); + } else if ((amqType & 0xCF) == 0x01) { // 0x01:int8, 0x11:int16, 0x21:int32 + Value* subval(new Value(TYPE_INT32)); + subval->setInt((int32_t)fvalue.get()); + appendToList(subval); } else if (amqType == 0x85 || amqType == 0x95) { Value* subval(new Value(TYPE_LSTR)); subval->setString(fvalue.get().c_str()); -- cgit v1.2.1 From a988bad81f9b5e9fd505a21c509ba5138791f74c Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Fri, 30 Apr 2010 16:58:36 +0000 Subject: QMF: Agent now sends class indications for schema added after attachement completes. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@939736 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/Agent.cpp | 31 ++++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/Agent.cpp b/qpid/cpp/src/qmf/engine/Agent.cpp index 7667213e5e..64c5961385 100644 --- a/qpid/cpp/src/qmf/engine/Agent.cpp +++ b/qpid/cpp/src/qmf/engine/Agent.cpp @@ -127,6 +127,7 @@ namespace engine { deque eventQueue; deque xmtQueue; map contextMap; + bool attachComplete; static const char* QMF_EXCHANGE; static const char* DIR_EXCHANGE; @@ -235,7 +236,7 @@ AgentImpl::AgentImpl(char* _label, bool i) : label(_label), queueName("qmfa-"), internalStore(i), nextTransientId(1), requestedBrokerBank(0), requestedAgentBank(0), assignedBrokerBank(0), assignedAgentBank(0), - bootSequence(1), nextObjectId(1), nextContextNum(1) + bootSequence(1), nextObjectId(1), nextContextNum(1), attachComplete(false) { queueName += label; } @@ -425,35 +426,53 @@ void AgentImpl::queryComplete(uint32_t sequence) void AgentImpl::registerClass(SchemaObjectClass* cls) { Mutex::ScopedLock _lock(lock); + bool newPackage = false; map::iterator iter = packages.find(cls->getClassKey()->getPackageName()); if (iter == packages.end()) { packages[cls->getClassKey()->getPackageName()] = ClassMaps(); iter = packages.find(cls->getClassKey()->getPackageName()); - // TODO: Indicate this package if connected + newPackage = true; } AgentClassKey key(cls->getClassKey()->getClassName(), cls->getClassKey()->getHash()); iter->second.objectClasses[key] = cls; - // TODO: Indicate this schema if connected. + // Indicate this new schema if connected. + + if (attachComplete) { + + if (newPackage) { + sendPackageIndicationLH(iter->first); + } + sendClassIndicationLH(CLASS_OBJECT, iter->first, key); + } } void AgentImpl::registerClass(SchemaEventClass* cls) { Mutex::ScopedLock _lock(lock); + bool newPackage = false; map::iterator iter = packages.find(cls->getClassKey()->getPackageName()); if (iter == packages.end()) { packages[cls->getClassKey()->getPackageName()] = ClassMaps(); iter = packages.find(cls->getClassKey()->getPackageName()); - // TODO: Indicate this package if connected + newPackage = true; } AgentClassKey key(cls->getClassKey()->getClassName(), cls->getClassKey()->getHash()); iter->second.eventClasses[key] = cls; - // TODO: Indicate this schema if connected. + // Indicate this new schema if connected. + + if (attachComplete) { + + if (newPackage) { + sendPackageIndicationLH(iter->first); + } + sendClassIndicationLH(CLASS_EVENT, iter->first, key); + } } const ObjectId* AgentImpl::addObject(Object&, uint64_t) @@ -662,6 +681,8 @@ void AgentImpl::handleAttachResponse(Buffer& inBuffer) cIter != cMap.eventClasses.end(); cIter++) sendClassIndicationLH(CLASS_EVENT, pIter->first, cIter->first); } + + attachComplete = true; } void AgentImpl::handlePackageRequest(Buffer&) -- cgit v1.2.1 From f3e8a605653ebb9ff310d0982e58721168a72fc5 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Tue, 11 May 2010 14:39:58 +0000 Subject: Support for multiple protocols in qpid::Url. - simplified qpid::Address to hold (protocol,host,port) triples. - protocol plugins call Url:addProtocol to add tags to Url parser. - use Address::protocol when establishing connections. - ssl_test: tests using URL to connect. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@943130 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp index b99cb414dc..69acc2dd37 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp @@ -78,7 +78,7 @@ BrokerEvent BrokerEventImpl::copy() BrokerProxyImpl::BrokerProxyImpl(BrokerProxy& pub, Console& _console) : publicObject(pub), console(_console) { stringstream qn; - qpid::TcpAddress addr; + qpid::Address addr; SystemInfo::getLocalHostname(addr); qn << "qmfc-" << SystemInfo::getProcessName() << "-" << addr << "-" << SystemInfo::getProcessId(); -- cgit v1.2.1 From 5afdc67935d07852c7c166741401ec4a77604d9b Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Fri, 21 May 2010 17:39:51 +0000 Subject: QMF: add bindEvent api to allow filtering of unsolicted events. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@947084 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ConsoleImpl.cpp | 30 ++++++++++++++++++++++++++++++ qpid/cpp/src/qmf/engine/ConsoleImpl.h | 2 ++ 2 files changed, 32 insertions(+) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp b/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp index 1b66d9e81f..4a5da31bdc 100644 --- a/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ConsoleImpl.cpp @@ -259,6 +259,32 @@ void ConsoleImpl::bindClass(const char* packageName, const char* className) (*iter)->addBinding(QMF_EXCHANGE, key.str()); } + +void ConsoleImpl::bindEvent(const SchemaClassKey* classKey) +{ + bindEvent(classKey->getPackageName(), classKey->getClassName()); +} + +void ConsoleImpl::bindEvent(const char* packageName, const char* eventName) +{ + if (!settings.userBindings) throw qpid::Exception("Console not configured for userBindings."); + if (settings.rcvEvents) throw qpid::Exception("Console already configured to receive all events."); + + stringstream key; + key << "console.event.*.*." << packageName; + if (eventName && *eventName) { + key << "." << eventName << ".#"; + } else { + key << ".#"; + } + + Mutex::ScopedLock _lock(lock); + bindingList.push_back(pair(string(), key.str())); + for (vector::iterator iter = brokerList.begin(); + iter != brokerList.end(); iter++) + (*iter)->addBinding(QMF_EXCHANGE, key.str()); +} + /* void ConsoleImpl::startSync(const Query& query, void* context, SyncQuery& sync) { @@ -421,6 +447,10 @@ const SchemaEventClass* Console::getEventClass(const SchemaClassKey* key) const void Console::bindPackage(const char* packageName) { impl->bindPackage(packageName); } void Console::bindClass(const SchemaClassKey* key) { impl->bindClass(key); } void Console::bindClass(const char* packageName, const char* className) { impl->bindClass(packageName, className); } + +void Console::bindEvent(const SchemaClassKey *key) { impl->bindEvent(key); } +void Console::bindEvent(const char* packageName, const char* eventName) { impl->bindEvent(packageName, eventName); } + //void Console::startSync(const Query& query, void* context, SyncQuery& sync) { impl->startSync(query, context, sync); } //void Console::touchSync(SyncQuery& sync) { impl->touchSync(sync); } //void Console::endSync(SyncQuery& sync) { impl->endSync(sync); } diff --git a/qpid/cpp/src/qmf/engine/ConsoleImpl.h b/qpid/cpp/src/qmf/engine/ConsoleImpl.h index ace47ec87c..0c27fdabcd 100644 --- a/qpid/cpp/src/qmf/engine/ConsoleImpl.h +++ b/qpid/cpp/src/qmf/engine/ConsoleImpl.h @@ -94,6 +94,8 @@ namespace engine { void bindPackage(const char* packageName); void bindClass(const SchemaClassKey* key); void bindClass(const char* packageName, const char* className); + void bindEvent(const SchemaClassKey* key); + void bindEvent(const char* packageName, const char* eventName); /* void startSync(const Query& query, void* context, SyncQuery& sync); -- cgit v1.2.1 From d06a4f9aa864b15d63ee967c5337699dc76a4157 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Thu, 22 Jul 2010 21:57:37 +0000 Subject: Bugfix: agent label in wrapped (python, ruby) qmf agent was not transmitted to the consoles. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@966871 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/Agent.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/Agent.cpp b/qpid/cpp/src/qmf/engine/Agent.cpp index 64c5961385..067f53471f 100644 --- a/qpid/cpp/src/qmf/engine/Agent.cpp +++ b/qpid/cpp/src/qmf/engine/Agent.cpp @@ -238,7 +238,7 @@ AgentImpl::AgentImpl(char* _label, bool i) : assignedBrokerBank(0), assignedAgentBank(0), bootSequence(1), nextObjectId(1), nextContextNum(1), attachComplete(false) { - queueName += label; + queueName += Uuid(true).str(); } AgentImpl::~AgentImpl() @@ -334,7 +334,7 @@ void AgentImpl::startProtocol() Buffer buffer(rawbuffer, 512); Protocol::encodeHeader(buffer, Protocol::OP_ATTACH_REQUEST); - buffer.putShortString("qmfa"); + buffer.putShortString(label); systemId.encode(buffer); buffer.putLong(requestedBrokerBank); buffer.putLong(requestedAgentBank); -- cgit v1.2.1 From 475d0274faa05752358c451c4b08bfba4724498c Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Wed, 22 Sep 2010 12:51:10 +0000 Subject: QPID-2880: allow boolean values in method call map/list arguments. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@999919 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/ValueImpl.cpp | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.cpp b/qpid/cpp/src/qmf/engine/ValueImpl.cpp index c58c28e166..409bf64c35 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ValueImpl.cpp @@ -21,6 +21,7 @@ #include #include #include +#include using namespace std; using namespace qmf::engine; @@ -153,6 +154,12 @@ void ValueImpl::initMap(const FieldTable& ft) subval->impl->initList(subList); insert(name.c_str(), subval); } + } else if (amqType == 0x08) { + Value* subval(new Value(TYPE_BOOL)); + subval->setBool(fvalue.get() ? true : false); + insert(name.c_str(), subval); + } else { + QPID_LOG(error, "Unable to decode unsupported AMQP typecode=" << amqType << " map index=" << name); } } } @@ -185,7 +192,7 @@ void ValueImpl::mapToFieldTable(FieldTable& ft) const ft.setInt64(name, subval.asInt64()); break; case TYPE_BOOL: - ft.setInt(name, subval.asBool() ? 1 : 0); + ft.set(name, FieldTable::ValuePtr(new BoolValue(subval.asBool()))); break; case TYPE_FLOAT: ft.setFloat(name, subval.asFloat()); @@ -274,6 +281,12 @@ void ValueImpl::initList(const List& fl) subVal->impl->initList(subList); appendToList(subVal); } + } else if (amqType == 0x08) { + Value* subval(new Value(TYPE_BOOL)); + subval->setBool(fvalue.get() ? true : false); + appendToList(subval); + } else { + QPID_LOG(error, "Unable to decode unsupported AMQP typecode =" << amqType); } } } @@ -303,7 +316,7 @@ void ValueImpl::listToFramingList(List& fl) const fl.push_back(List::ValuePtr(new Integer64Value(subval.asInt64()))); break; case TYPE_BOOL: - fl.push_back(List::ValuePtr(new IntegerValue(subval.asBool() ? 1 : 0))); + fl.push_back(List::ValuePtr(new BoolValue(subval.asBool() ? 1 : 0))); break; case TYPE_FLOAT: fl.push_back(List::ValuePtr(new FloatValue(subval.asFloat()))); -- cgit v1.2.1 From 51b79cafaa57b231f3fd87e29fe190a442bd8a81 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Thu, 13 Jan 2011 19:10:07 +0000 Subject: In qmfengine, if a method call or method response requires a buffer larger than the static buffer used for communication, allocate a large-enough buffer temporarily from the heap. A test is included to verify large-buffer behavior. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1058709 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/Agent.cpp | 32 +++++++++++++-- qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp | 60 ++++++++++++++++++++++++----- qpid/cpp/src/qmf/engine/BrokerProxyImpl.h | 1 + qpid/cpp/src/qmf/engine/ValueImpl.cpp | 45 ++++++++++++++++++++++ qpid/cpp/src/qmf/engine/ValueImpl.h | 1 + 5 files changed, 126 insertions(+), 13 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/Agent.cpp b/qpid/cpp/src/qmf/engine/Agent.cpp index 067f53471f..1f08dded94 100644 --- a/qpid/cpp/src/qmf/engine/Agent.cpp +++ b/qpid/cpp/src/qmf/engine/Agent.cpp @@ -356,8 +356,7 @@ void AgentImpl::heartbeat() QPID_LOG(trace, "SENT HeartbeatIndication"); } -void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, - const Value& argMap) +void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& argMap) { Mutex::ScopedLock _lock(lock); map::iterator iter = contextMap.find(sequence); @@ -366,7 +365,32 @@ void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, AgentQueryContext::Ptr context = iter->second; contextMap.erase(iter); - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + char* buf(outputBuffer); + uint32_t bufLen(114 + strlen(text)); // header(8) + status(4) + mstring(2 + size) + margin(100) + bool allocated(false); + + if (status == 0) { + for (vector::const_iterator aIter = context->schemaMethod->impl->arguments.begin(); + aIter != context->schemaMethod->impl->arguments.end(); aIter++) { + const SchemaArgument* schemaArg = *aIter; + if (schemaArg->getDirection() == DIR_OUT || schemaArg->getDirection() == DIR_IN_OUT) { + if (argMap.keyInMap(schemaArg->getName())) { + const Value* val = argMap.byKey(schemaArg->getName()); + bufLen += val->impl->encodedSize(); + } else { + Value val(schemaArg->getType()); + bufLen += val.impl->encodedSize(); + } + } + } + } + + if (bufLen > MA_BUFFER_SIZE) { + buf = (char*) malloc(bufLen); + allocated = true; + } + + Buffer buffer(buf, bufLen); Protocol::encodeHeader(buffer, Protocol::OP_METHOD_RESPONSE, context->sequence); buffer.putLong(status); buffer.putMediumString(text); @@ -386,6 +410,8 @@ void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, } } sendBufferLH(buffer, context->exchange, context->key); + if (allocated) + free(buf); QPID_LOG(trace, "SENT MethodResponse seq=" << context->sequence << " status=" << status << " text=" << text); } diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp index 69acc2dd37..5fc71979fd 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp @@ -269,6 +269,31 @@ string BrokerProxyImpl::encodeMethodArguments(const SchemaMethod* schema, const return string(); } +string BrokerProxyImpl::encodedSizeMethodArguments(const SchemaMethod* schema, const Value* argmap, uint32_t& size) +{ + int argCount = schema->getArgumentCount(); + + if (argmap == 0 || !argmap->isMap()) + return string("Arguments must be in a map value"); + + for (int aIdx = 0; aIdx < argCount; aIdx++) { + const SchemaArgument* arg(schema->getArgument(aIdx)); + if (arg->getDirection() == DIR_IN || arg->getDirection() == DIR_IN_OUT) { + if (argmap->keyInMap(arg->getName())) { + const Value* argVal(argmap->byKey(arg->getName())); + if (argVal->getType() != arg->getType()) + return string("Argument is the wrong type: ") + arg->getName(); + size += argVal->impl->encodedSize(); + } else { + Value defaultValue(arg->getType()); + size += defaultValue.impl->encodedSize(); + } + } + } + + return string(); +} + void BrokerProxyImpl::sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls, const string& methodName, const Value* args, void* userContext) { @@ -280,7 +305,23 @@ void BrokerProxyImpl::sendMethodRequest(ObjectId* oid, const SchemaObjectClass* Mutex::ScopedLock _lock(lock); SequenceContext::Ptr methodContext(new MethodContext(*this, userContext, method)); stringstream key; - Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); + char* buf(outputBuffer); + uint32_t bufLen(1024); + bool allocated(false); + + string argErrorString = encodedSizeMethodArguments(method, args, bufLen); + if (!argErrorString.empty()) { + MethodResponsePtr argError(MethodResponseImpl::factory(1, argErrorString)); + eventQueue.push_back(eventMethodResponse(userContext, argError)); + return; + } + + if (bufLen > MA_BUFFER_SIZE) { + buf = (char*) malloc(bufLen); + allocated = true; + } + + Buffer outBuffer(buf, bufLen); uint32_t sequence(seqMgr.reserve(methodContext)); Protocol::encodeHeader(outBuffer, Protocol::OP_METHOD_REQUEST, sequence); @@ -288,15 +329,14 @@ void BrokerProxyImpl::sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls->getClassKey()->impl->encode(outBuffer); outBuffer.putShortString(methodName); - string argErrorString = encodeMethodArguments(method, args, outBuffer); - if (argErrorString.empty()) { - key << "agent.1." << oid->impl->getAgentBank(); - sendBufferLH(outBuffer, QMF_EXCHANGE, key.str()); - QPID_LOG(trace, "SENT MethodRequest seq=" << sequence << " method=" << methodName << " key=" << key.str()); - } else { - MethodResponsePtr argError(MethodResponseImpl::factory(1, argErrorString)); - eventQueue.push_back(eventMethodResponse(userContext, argError)); - } + encodeMethodArguments(method, args, outBuffer); + key << "agent.1." << oid->impl->getAgentBank(); + sendBufferLH(outBuffer, QMF_EXCHANGE, key.str()); + QPID_LOG(trace, "SENT MethodRequest seq=" << sequence << " method=" << methodName << " key=" << key.str()); + + if (allocated) + free(buf); + return; } } diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h index 494da5e239..0542b67dbb 100644 --- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h +++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h @@ -142,6 +142,7 @@ namespace engine { void sendQuery(const Query& query, void* context, const AgentProxy* agent); bool sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent); std::string encodeMethodArguments(const SchemaMethod* schema, const Value* args, qpid::framing::Buffer& buffer); + std::string encodedSizeMethodArguments(const SchemaMethod* schema, const Value* args, uint32_t& size); void sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls, const std::string& method, const Value* args, void* context); void addBinding(const std::string& exchange, const std::string& key); diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.cpp b/qpid/cpp/src/qmf/engine/ValueImpl.cpp index 409bf64c35..f9ebbf5028 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.cpp +++ b/qpid/cpp/src/qmf/engine/ValueImpl.cpp @@ -394,6 +394,51 @@ void ValueImpl::encode(Buffer& buf) const } } +uint32_t ValueImpl::encodedSize() const +{ + FieldTable ft; + List fl; + + switch (typecode) { + case TYPE_UINT8 : + case TYPE_BOOL : + case TYPE_INT8 : return 1; + + case TYPE_UINT16 : + case TYPE_INT16 : return 2; + + case TYPE_UINT32 : + case TYPE_INT32 : + case TYPE_FLOAT : return 4; + + case TYPE_UINT64 : + case TYPE_INT64 : + case TYPE_DOUBLE : + case TYPE_ABSTIME : + case TYPE_DELTATIME : return 8; + + case TYPE_UUID : + case TYPE_REF : return 16; + + case TYPE_SSTR : return 1 + stringVal.size(); + case TYPE_LSTR : return 2 + stringVal.size(); + case TYPE_MAP: + mapToFieldTable(ft); + return ft.encodedSize(); + + case TYPE_LIST: + listToFramingList(fl); + return fl.encodedSize(); + + case TYPE_ARRAY: + case TYPE_OBJECT: + default: + break; + } + + return 0; +} + bool ValueImpl::keyInMap(const char* key) const { return typecode == TYPE_MAP && mapVal.count(key) > 0; diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.h b/qpid/cpp/src/qmf/engine/ValueImpl.h index 3b535834fd..8de8c5329f 100644 --- a/qpid/cpp/src/qmf/engine/ValueImpl.h +++ b/qpid/cpp/src/qmf/engine/ValueImpl.h @@ -79,6 +79,7 @@ namespace engine { ~ValueImpl(); void encode(qpid::framing::Buffer& b) const; + uint32_t encodedSize() const; Typecode getType() const { return typecode; } bool isNull() const { return !valid; } -- cgit v1.2.1 From 7bfe51aa2850d8f3623fb439caa8ce123aba0a4e Mon Sep 17 00:00:00 2001 From: Andrew Stitcher Date: Mon, 9 May 2011 19:56:20 +0000 Subject: QPID-3004: Get Clang to compile qpid c++ - Avoid pointer alignment conversion error by using union instead git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1101184 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qmf/engine/SchemaImpl.cpp | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) (limited to 'qpid/cpp/src/qmf/engine') diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp index e0948a9911..f75663e131 100644 --- a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp +++ b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp @@ -55,9 +55,12 @@ void SchemaHash::update(uint8_t data) void SchemaHash::update(const char* data, uint32_t len) { - uint64_t* first = (uint64_t*) hash; - uint64_t* second = (uint64_t*) hash + 1; - + union h { + uint8_t b[16]; + uint64_t q[2]; + }* h = reinterpret_cast(&hash[0]); + uint64_t* first = &h->q[0]; + uint64_t* second = &h->q[1]; for (uint32_t idx = 0; idx < len; idx++) { *first = *first ^ (uint64_t) data[idx]; *second = *second << 1; -- cgit v1.2.1