QPID-2792 - QMF Clients in C++ following the ABI-stable pattern of the new messaging API

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

1 files changed, 578 insertions, 0 deletions

diff --git a/cpp/src/qmf/Agent.cpp b/cpp/src/qmf/Agent.cpp
new file mode 100644
index 0000000000..c7ccea35d5
--- /dev/null
+++ b/cpp/src/qmf/Agent.cpp
@@ -0,0 +1,578 @@
+/*
+ *
+ * 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/AgentImpl.h"
+#include "qmf/PrivateImplRef.h"
+#include "qmf/ConsoleEventImpl.h"
+#include "qmf/ConsoleSession.h"
+#include "qmf/DataImpl.h"
+#include "qmf/Query.h"
+#include "qpid/messaging/Sender.h"
+#include "qpid/messaging/AddressParser.h"
+#include "qpid/management/Buffer.h"
+#include "qpid/log/Statement.h"
+#include <boost/lexical_cast.hpp>
+
+using qpid::types::Variant;
+using qpid::messaging::Duration;
+using qpid::messaging::Message;
+using qpid::messaging::Sender;
+using namespace std;
+using namespace qmf;
+
+typedef PrivateImplRef<Agent> PI;
+
+Agent::Agent(AgentImpl* impl) { PI::ctor(*this, impl); }
+Agent::Agent(const Agent& s) : qmf::Handle<AgentImpl>() { PI::copy(*this, s); }
+Agent::~Agent() { PI::dtor(*this); }
+Agent& Agent::operator=(const Agent& s) { return PI::assign(*this, s); }
+string Agent::getName() const { return isValid() ? impl->getName() : ""; }
+uint32_t Agent::getEpoch() const { return isValid() ? impl->getEpoch() : 0; }
+string Agent::getVendor() const { return isValid() ? impl->getVendor() : ""; }
+string Agent::getProduct() const { return isValid() ? impl->getProduct() : ""; }
+string Agent::getInstance() const { return isValid() ? impl->getInstance() : ""; }
+const Variant& Agent::getAttribute(const string& k) const { return impl->getAttribute(k); }
+const Variant::Map& Agent::getAttributes() const { return impl->getAttributes(); }
+ConsoleEvent Agent::query(const Query& q, Duration t) { return impl->query(q, t); }
+ConsoleEvent Agent::query(const string& q, Duration t) { return impl->query(q, t); }
+uint32_t Agent::queryAsync(const Query& q) { return impl->queryAsync(q); }
+uint32_t Agent::queryAsync(const string& q) { return impl->queryAsync(q); }
+ConsoleEvent Agent::callMethod(const string& m, const Variant::Map& a, const DataAddr& d, Duration t) { return impl->callMethod(m, a, d, t); }
+uint32_t Agent::callMethodAsync(const string& m, const Variant::Map& a, const DataAddr& d) { return impl->callMethodAsync(m, a, d); }
+uint32_t Agent::getPackageCount() const { return impl->getPackageCount(); }
+const string& Agent::getPackage(uint32_t i) const { return impl->getPackage(i); }
+uint32_t Agent::getSchemaIdCount(const string& p) const { return impl->getSchemaIdCount(p); }
+SchemaId Agent::getSchemaId(const string& p, uint32_t i) const { return impl->getSchemaId(p, i); }
+Schema Agent::getSchema(const SchemaId& s, Duration t) { return impl->getSchema(s, t); }
+
+
+
+AgentImpl::AgentImpl(const std::string& n, uint32_t e, ConsoleSessionImpl& s) :
+    name(n), epoch(e), session(s), touched(true), untouchedCount(0),
+    nextCorrelator(1), schemaCache(s.schemaCache)
+{
+}
+
+const Variant& AgentImpl::getAttribute(const string& k) const
+{
+    Variant::Map::const_iterator iter = attributes.find(k);
+    if (iter == attributes.end())
+        throw KeyNotFound(k);
+    return iter->second;
+}
+
+ConsoleEvent AgentImpl::query(const Query& query, Duration timeout)
+{
+    boost::shared_ptr<SyncContext> context(new SyncContext());
+    uint32_t correlator;
+    ConsoleEvent result;
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        correlator = nextCorrelator++;
+        contextMap[correlator] = context;
+    }
+    try {
+        sendQuery(query, correlator);
+        {
+            uint64_t milliseconds = timeout.getMilliseconds();
+            qpid::sys::Mutex::ScopedLock cl(context->lock);
+            if (!context->response.isValid() || !context->response.isFinal())
+                context->cond.wait(context->lock,
+                                   qpid::sys::AbsTime(qpid::sys::now(),
+                                                      qpid::sys::Duration(milliseconds * qpid::sys::TIME_MSEC)));
+            if (context->response.isValid() && context->response.isFinal())
+                result = context->response;
+            else {
+                auto_ptr<ConsoleEventImpl> impl(new ConsoleEventImpl(CONSOLE_EXCEPTION));
+                Data exception(new DataImpl());
+                exception.setProperty("error_text", "Timed out waiting for the agent to respond");
+                impl->addData(exception);
+                result = ConsoleEvent(impl.release());
+            }
+        }
+    } catch (qpid::types::Exception&) {
+    }
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        contextMap.erase(correlator);
+    }
+
+    return result;
+}
+
+
+ConsoleEvent AgentImpl::query(const string& text, Duration timeout)
+{
+    return query(stringToQuery(text), timeout);
+}
+
+
+uint32_t AgentImpl::queryAsync(const Query& query)
+{
+    uint32_t correlator;
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        correlator = nextCorrelator++;
+    }
+
+    sendQuery(query, correlator);
+    return correlator;
+}
+
+
+uint32_t AgentImpl::queryAsync(const string& text)
+{
+    return queryAsync(stringToQuery(text));
+}
+
+
+ConsoleEvent AgentImpl::callMethod(const string& method, const Variant::Map& args, const DataAddr& addr, Duration timeout)
+{
+    boost::shared_ptr<SyncContext> context(new SyncContext());
+    uint32_t correlator;
+    ConsoleEvent result;
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        correlator = nextCorrelator++;
+        contextMap[correlator] = context;
+    }
+    try {
+        sendMethod(method, args, addr, correlator);
+        {
+            uint64_t milliseconds = timeout.getMilliseconds();
+            qpid::sys::Mutex::ScopedLock cl(context->lock);
+            if (!context->response.isValid())
+                context->cond.wait(context->lock,
+                                   qpid::sys::AbsTime(qpid::sys::now(),
+                                                      qpid::sys::Duration(milliseconds * qpid::sys::TIME_MSEC)));
+            if (context->response.isValid())
+                result = context->response;
+            else {
+                auto_ptr<ConsoleEventImpl> impl(new ConsoleEventImpl(CONSOLE_EXCEPTION));
+                Data exception(new DataImpl());
+                exception.setProperty("error_text", "Timed out waiting for the agent to respond");
+                impl->addData(exception);
+                result = ConsoleEvent(impl.release());
+            }
+        }
+    } catch (qpid::types::Exception&) {
+    }
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        contextMap.erase(correlator);
+    }
+
+    return result;
+}
+
+
+uint32_t AgentImpl::callMethodAsync(const string& method, const Variant::Map& args, const DataAddr& addr)
+{
+    uint32_t correlator;
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        correlator = nextCorrelator++;
+    }
+
+    sendMethod(method, args, addr, correlator);
+    return correlator;
+}
+
+
+uint32_t AgentImpl::getPackageCount() const
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+
+    //
+    // Populate the package set.
+    //
+    for (set<SchemaId>::const_iterator iter = schemaIdSet.begin(); iter != schemaIdSet.end(); iter++)
+        packageSet.insert(iter->getPackageName());
+
+    return packageSet.size();
+}
+
+
+const string& AgentImpl::getPackage(uint32_t idx) const
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+    uint32_t count(0);
+    for (set<string>::const_iterator iter = packageSet.begin(); iter != packageSet.end(); iter++) {
+        if (idx == count)
+            return *iter;
+        count++;
+    }
+    throw IndexOutOfRange();
+}
+
+
+uint32_t AgentImpl::getSchemaIdCount(const string& pname) const
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+    uint32_t count(0);
+    for (set<SchemaId>::const_iterator iter = schemaIdSet.begin(); iter != schemaIdSet.end(); iter++)
+        if (iter->getPackageName() == pname)
+            count++;
+    return count;
+}
+
+
+SchemaId AgentImpl::getSchemaId(const string& pname, uint32_t idx) const
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+    uint32_t count(0);
+    for (set<SchemaId>::const_iterator iter = schemaIdSet.begin(); iter != schemaIdSet.end(); iter++) {
+        if (iter->getPackageName() == pname) {
+            if (idx == count)
+                return *iter;
+            count++;
+        }
+    }
+    throw IndexOutOfRange();
+}
+
+
+Schema AgentImpl::getSchema(const SchemaId& id, Duration timeout)
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+    if (!schemaCache->haveSchema(id))
+        //
+        // The desired schema is not in the cache.  We need to asynchronously query the remote
+        // agent for the information.  The call to schemaCache->getSchema will block waiting for
+        // the response to be received.
+        //
+        sendSchemaRequest(id);
+
+    return schemaCache->getSchema(id, timeout);
+}
+
+
+void AgentImpl::handleException(const Variant::Map& content, const Message& msg)
+{
+    const string& cid(msg.getCorrelationId());
+    Variant::Map::const_iterator aIter;
+    uint32_t correlator;
+    boost::shared_ptr<SyncContext> context;
+
+    try { correlator = boost::lexical_cast<uint32_t>(cid); }
+    catch(const boost::bad_lexical_cast&) { correlator = 0; }
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        map<uint32_t, boost::shared_ptr<SyncContext> >::iterator iter = contextMap.find(correlator);
+        if (iter != contextMap.end())
+            context = iter->second;
+    }
+
+    if (context.get() != 0) {
+        //
+        // This exception is associated with a synchronous request.
+        //
+        qpid::sys::Mutex::ScopedLock cl(context->lock);
+        context->response = ConsoleEvent(new ConsoleEventImpl(CONSOLE_EXCEPTION));
+        ConsoleEventImplAccess::get(context->response).addData(new DataImpl(content, this));
+        ConsoleEventImplAccess::get(context->response).setAgent(this);
+        context->cond.notify();
+    } else {
+        //
+        // This exception is associated with an asynchronous request.
+        //
+        auto_ptr<ConsoleEventImpl> eventImpl(new ConsoleEventImpl(CONSOLE_EXCEPTION));
+        eventImpl->setCorrelator(correlator);
+        eventImpl->setAgent(this);
+        eventImpl->addData(new DataImpl(content, this));
+        session.enqueueEvent(eventImpl.release());
+    }
+}
+
+
+void AgentImpl::handleMethodResponse(const Variant::Map& response, const Message& msg)
+{
+    const string& cid(msg.getCorrelationId());
+    Variant::Map::const_iterator aIter;
+    Variant::Map argMap;
+    uint32_t correlator;
+    boost::shared_ptr<SyncContext> context;
+
+    QPID_LOG(trace, "RCVD MethodResponse map=" << response);
+
+    aIter = response.find("_arguments");
+    if (aIter != response.end())
+        argMap = aIter->second.asMap();
+
+    try { correlator = boost::lexical_cast<uint32_t>(cid); }
+    catch(const boost::bad_lexical_cast&) { correlator = 0; }
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        map<uint32_t, boost::shared_ptr<SyncContext> >::iterator iter = contextMap.find(correlator);
+        if (iter != contextMap.end())
+            context = iter->second;
+    }
+
+    if (context.get() != 0) {
+        //
+        // This response is associated with a synchronous request.
+        //
+        qpid::sys::Mutex::ScopedLock cl(context->lock);
+        context->response = ConsoleEvent(new ConsoleEventImpl(CONSOLE_METHOD_RESPONSE));
+        ConsoleEventImplAccess::get(context->response).setArguments(argMap);
+        ConsoleEventImplAccess::get(context->response).setAgent(this);
+        context->cond.notify();
+    } else {
+        //
+        // This response is associated with an asynchronous request.
+        //
+        auto_ptr<ConsoleEventImpl> eventImpl(new ConsoleEventImpl(CONSOLE_METHOD_RESPONSE));
+        eventImpl->setCorrelator(correlator);
+        eventImpl->setAgent(this);
+        eventImpl->setArguments(argMap);
+        session.enqueueEvent(eventImpl.release());
+    }
+}
+
+
+void AgentImpl::handleDataIndication(const Variant::List&, const Message&)
+{
+    // TODO
+}
+
+
+void AgentImpl::handleQueryResponse(const Variant::List& list, const Message& msg)
+{
+    const string& cid(msg.getCorrelationId());
+    Variant::Map::const_iterator aIter;
+    const Variant::Map& props(msg.getProperties());
+    uint32_t correlator;
+    bool final(false);
+    boost::shared_ptr<SyncContext> context;
+
+    aIter = props.find("partial");
+    if (aIter == props.end())
+        final = true;
+
+    try { correlator = boost::lexical_cast<uint32_t>(cid); }
+    catch(const boost::bad_lexical_cast&) { correlator = 0; }
+
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        map<uint32_t, boost::shared_ptr<SyncContext> >::iterator iter = contextMap.find(correlator);
+        if (iter != contextMap.end())
+            context = iter->second;
+    }
+
+    if (context.get() != 0) {
+        //
+        // This response is associated with a synchronous request.
+        //
+        qpid::sys::Mutex::ScopedLock cl(context->lock);
+        if (!context->response.isValid())
+            context->response = ConsoleEvent(new ConsoleEventImpl(CONSOLE_QUERY_RESPONSE));
+        for (Variant::List::const_iterator lIter = list.begin(); lIter != list.end(); lIter++) {
+            Data data(new DataImpl(lIter->asMap(), this));
+            ConsoleEventImplAccess::get(context->response).addData(data);
+            if (data.hasSchema())
+                learnSchemaId(data.getSchemaId());
+        }
+        if (final) {
+            ConsoleEventImplAccess::get(context->response).setFinal();
+            ConsoleEventImplAccess::get(context->response).setAgent(this);
+            context->cond.notify();
+        }
+    } else {
+        //
+        // This response is associated with an asynchronous request.
+        //
+        auto_ptr<ConsoleEventImpl> eventImpl(new ConsoleEventImpl(CONSOLE_QUERY_RESPONSE));
+        eventImpl->setCorrelator(correlator);
+        eventImpl->setAgent(this);
+        for (Variant::List::const_iterator lIter = list.begin(); lIter != list.end(); lIter++) {
+            Data data(new DataImpl(lIter->asMap(), this));
+            eventImpl->addData(data);
+            if (data.hasSchema())
+                learnSchemaId(data.getSchemaId());
+        }
+        if (final)
+            eventImpl->setFinal();
+        session.enqueueEvent(eventImpl.release());
+    }
+}
+
+
+Query AgentImpl::stringToQuery(const std::string& text)
+{
+    qpid::messaging::AddressParser parser(text);
+    Variant::Map map;
+    Variant::Map::const_iterator iter;
+    string className;
+    string packageName;
+
+    parser.parseMap(map);
+
+    iter = map.find("class");
+    if (iter != map.end())
+        className = iter->second.asString();
+
+    iter = map.find("package");
+    if (iter != map.end())
+        packageName = iter->second.asString();
+
+    Query query(className, packageName);
+
+    iter = map.find("where");
+    if (iter != map.end()) {
+        const Variant::Map& pred(iter->second.asMap());
+        for (iter = pred.begin(); iter != pred.end(); iter++)
+            query.addPredicate(iter->first, iter->second);
+    }
+
+    return query;
+}
+
+
+void AgentImpl::sendQuery(const Query& query, uint32_t correlator)
+{
+    Message msg;
+    Variant::Map map;
+    Variant::Map& headers(msg.getProperties());
+
+    headers["method"] = "request";
+    headers["qmf.opcode"] = "_query_request";
+    headers["x-amqp-0-10.app-id"] = "qmf2";
+
+    map["_what"] = "OBJECT";
+
+    const DataAddr& dataAddr(query.getDataAddr());
+    const SchemaId& schemaId(query.getSchemaId());
+
+    if (dataAddr.isValid())
+        map["_object_id"] = dataAddr.asMap();
+    else {
+        string className;
+        string packageName;
+        if (schemaId.isValid()) {
+            className = schemaId.getName();
+            packageName = schemaId.getPackageName();
+        } else {
+            className = query.getClassName();
+            if (className.empty())
+                throw QmfException("Invalid Query");
+            packageName = query.getPackageName();
+        }
+        Variant::Map idMap;
+        idMap["_class_name"] = className;
+        if (!packageName.empty())
+            idMap["_package_name"] = packageName;
+        map["_schema_id"] = idMap;
+    }
+
+    //
+    // TODO: Encode a simple-predicate if present.
+    //
+
+    msg.setReplyTo(session.replyAddress);
+    msg.setCorrelationId(boost::lexical_cast<string>(correlator));
+    encode(map, msg);
+    Sender sender(session.session.createSender(session.directBase + "/" + name));
+    sender.send(msg);
+    sender.close();
+
+    QPID_LOG(trace, "SENT QueryRequest to=" << name);
+}
+
+
+void AgentImpl::sendMethod(const string& method, const Variant::Map& args, const DataAddr& addr, uint32_t correlator)
+{
+    Message msg;
+    Variant::Map map;
+    Variant::Map& headers(msg.getProperties());
+
+    headers["method"] = "request";
+    headers["qmf.opcode"] = "_method_request";
+    headers["x-amqp-0-10.app-id"] = "qmf2";
+
+    map["_method_name"] = method;
+    map["_object_id"] = addr.asMap();
+    map["_arguments"] = args;
+
+    msg.setReplyTo(session.replyAddress);
+    msg.setCorrelationId(boost::lexical_cast<string>(correlator));
+    encode(map, msg);
+    Sender sender(session.session.createSender(session.directBase + "/" + name));
+    sender.send(msg);
+    sender.close();
+
+    QPID_LOG(trace, "SENT MethodRequest method=" << method << " to=" << name);
+}
+
+void AgentImpl::sendSchemaRequest(const SchemaId& id)
+{
+    // TODO: Check agent's capability value to determine which kind of schema request to make
+
+#define RAW_BUFFER_SIZE 1024
+    char rawBuffer[RAW_BUFFER_SIZE];
+    qpid::management::Buffer buffer(rawBuffer, RAW_BUFFER_SIZE);
+
+    buffer.putOctet('A');
+    buffer.putOctet('M');
+    buffer.putOctet('2');
+    buffer.putOctet('S');
+    buffer.putLong(nextCorrelator++);
+    buffer.putShortString(id.getPackageName());
+    buffer.putShortString(id.getName());
+    buffer.putBin128(id.getHash().data());
+
+    string content(rawBuffer, buffer.getPosition());
+
+    Message msg;
+    msg.setReplyTo(session.replyAddress);
+    msg.setContent(content);
+    Sender sender(session.session.createSender(session.directBase + "/" + name));
+    sender.send(msg);
+    sender.close();
+
+    QPID_LOG(trace, "SENT V1SchemaRequest to=" << name);
+}
+
+
+void AgentImpl::learnSchemaId(const SchemaId& id)
+{
+    schemaCache->declareSchemaId(id);
+    schemaIdSet.insert(id);
+}
+
+
+AgentImpl& AgentImplAccess::get(Agent& item)
+{
+    return *item.impl;
+}
+
+
+const AgentImpl& AgentImplAccess::get(const Agent& item)
+{
+    return *item.impl;
+}