Creating a branch for experimenting with some ideas for JMS client.rajith_jms_client

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

1 files changed, 0 insertions, 1072 deletions

diff --git a/cpp/src/qmf/AgentSession.cpp b/cpp/src/qmf/AgentSession.cpp
deleted file mode 100644
index 71d369325f..0000000000
--- a/cpp/src/qmf/AgentSession.cpp
+++ /dev/null
@@ -1,1072 +0,0 @@
-/*
- *
- * 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/RefCounted.h"
-#include "qmf/PrivateImplRef.h"
-#include "qmf/exceptions.h"
-#include "qmf/AgentSession.h"
-#include "qmf/AgentEventImpl.h"
-#include "qmf/SchemaIdImpl.h"
-#include "qmf/SchemaImpl.h"
-#include "qmf/DataAddrImpl.h"
-#include "qmf/DataImpl.h"
-#include "qmf/QueryImpl.h"
-#include "qmf/agentCapability.h"
-#include "qmf/constants.h"
-#include "qpid/sys/Mutex.h"
-#include "qpid/sys/Condition.h"
-#include "qpid/sys/Thread.h"
-#include "qpid/sys/Runnable.h"
-#include "qpid/log/Statement.h"
-#include "qpid/messaging/Connection.h"
-#include "qpid/messaging/Session.h"
-#include "qpid/messaging/Receiver.h"
-#include "qpid/messaging/Sender.h"
-#include "qpid/messaging/Message.h"
-#include "qpid/messaging/AddressParser.h"
-#include "qpid/management/Buffer.h"
-#include <queue>
-#include <map>
-#include <set>
-#include <iostream>
-#include <memory>
-
-using namespace std;
-using namespace qpid::messaging;
-using namespace qmf;
-using qpid::types::Variant;
-
-namespace qmf {
-    class AgentSessionImpl : public virtual qpid::RefCounted, public qpid::sys::Runnable {
-    public:
-        ~AgentSessionImpl();
-
-        //
-        // Methods from API handle
-        //
-        AgentSessionImpl(Connection& c, const string& o);
-        void setDomain(const string& d) { checkOpen(); domain = d; }
-        void setVendor(const string& v) { checkOpen(); attributes["_vendor"] = v; }
-        void setProduct(const string& p) { checkOpen(); attributes["_product"] = p; }
-        void setInstance(const string& i) { checkOpen(); attributes["_instance"] = i; }
-        void setAttribute(const string& k, const qpid::types::Variant& v) { checkOpen(); attributes[k] = v; }
-        const string& getName() const { return agentName; }
-        void open();
-        void close();
-        bool nextEvent(AgentEvent& e, Duration t);
-        int pendingEvents() const;
-
-        void registerSchema(Schema& s);
-        DataAddr addData(Data& d, const string& n, bool persist);
-        void delData(const DataAddr&);
-
-        void authAccept(AgentEvent& e);
-        void authReject(AgentEvent& e, const string& m);
-        void raiseException(AgentEvent& e, const string& s);
-        void raiseException(AgentEvent& e, const Data& d);
-        void response(AgentEvent& e, const Data& d);
-        void complete(AgentEvent& e);
-        void methodSuccess(AgentEvent& e);
-        void raiseEvent(const Data& d);
-        void raiseEvent(const Data& d, int s);
-
-    private:
-        typedef map<DataAddr, Data, DataAddrCompare> DataIndex;
-        typedef map<SchemaId, Schema, SchemaIdCompare> SchemaMap;
-
-        mutable qpid::sys::Mutex lock;
-        qpid::sys::Condition cond;
-        Connection connection;
-        Session session;
-        Sender directSender;
-        Sender topicSender;
-        string domain;
-        Variant::Map attributes;
-        Variant::Map options;
-        string agentName;
-        bool opened;
-        queue<AgentEvent> eventQueue;
-        qpid::sys::Thread* thread;
-        bool threadCanceled;
-        uint32_t bootSequence;
-        uint32_t interval;
-        uint64_t lastHeartbeat;
-        uint64_t lastVisit;
-        bool forceHeartbeat;
-        bool externalStorage;
-        bool autoAllowQueries;
-        bool autoAllowMethods;
-        uint32_t maxSubscriptions;
-        uint32_t minSubInterval;
-        uint32_t subLifetime;
-        bool publicEvents;
-        bool listenOnDirect;
-        bool strictSecurity;
-        uint64_t schemaUpdateTime;
-        string directBase;
-        string topicBase;
-
-        SchemaMap schemata;
-        DataIndex globalIndex;
-        map<SchemaId, DataIndex, SchemaIdCompareNoHash> schemaIndex;
-
-        void checkOpen();
-        void setAgentName();
-        void enqueueEvent(const AgentEvent&);
-        void handleLocateRequest(const Variant::List& content, const Message& msg);
-        void handleMethodRequest(const Variant::Map& content, const Message& msg);
-        void handleQueryRequest(const Variant::Map& content, const Message& msg);
-        void handleSchemaRequest(AgentEvent&);
-        void handleV1SchemaRequest(qpid::management::Buffer&, uint32_t, const Message&);
-        void dispatch(Message);
-        void sendHeartbeat();
-        void send(Message, const Address&);
-        void flushResponses(AgentEvent&, bool);
-        void periodicProcessing(uint64_t);
-        void run();
-    };
-}
-
-typedef qmf::PrivateImplRef<AgentSession> PI;
-
-AgentSession::AgentSession(AgentSessionImpl* impl) { PI::ctor(*this, impl); }
-AgentSession::AgentSession(const AgentSession& s) : qmf::Handle<AgentSessionImpl>() { PI::copy(*this, s); }
-AgentSession::~AgentSession() { PI::dtor(*this); }
-AgentSession& AgentSession::operator=(const AgentSession& s) { return PI::assign(*this, s); }
-
-AgentSession::AgentSession(Connection& c, const string& o) { PI::ctor(*this, new AgentSessionImpl(c, o)); }
-void AgentSession::setDomain(const string& d) { impl->setDomain(d); }
-void AgentSession::setVendor(const string& v) { impl->setVendor(v); }
-void AgentSession::setProduct(const string& p) { impl->setProduct(p); }
-void AgentSession::setInstance(const string& i) { impl->setInstance(i); }
-void AgentSession::setAttribute(const string& k, const qpid::types::Variant& v) { impl->setAttribute(k, v); }
-const string& AgentSession::getName() const { return impl->getName(); }
-void AgentSession::open() { impl->open(); }
-void AgentSession::close() { impl->close(); }
-bool AgentSession::nextEvent(AgentEvent& e, Duration t) { return impl->nextEvent(e, t); }
-int AgentSession::pendingEvents() const { return impl->pendingEvents(); }
-void AgentSession::registerSchema(Schema& s) { impl->registerSchema(s); }
-DataAddr AgentSession::addData(Data& d, const string& n, bool p) { return impl->addData(d, n, p); }
-void AgentSession::delData(const DataAddr& a) { impl->delData(a); }
-void AgentSession::authAccept(AgentEvent& e) { impl->authAccept(e); }
-void AgentSession::authReject(AgentEvent& e, const string& m) { impl->authReject(e, m); }
-void AgentSession::raiseException(AgentEvent& e, const string& s) { impl->raiseException(e, s); }
-void AgentSession::raiseException(AgentEvent& e, const Data& d) { impl->raiseException(e, d); }
-void AgentSession::response(AgentEvent& e, const Data& d) { impl->response(e, d); }
-void AgentSession::complete(AgentEvent& e) { impl->complete(e); }
-void AgentSession::methodSuccess(AgentEvent& e) { impl->methodSuccess(e); }
-void AgentSession::raiseEvent(const Data& d) { impl->raiseEvent(d); }
-void AgentSession::raiseEvent(const Data& d, int s) { impl->raiseEvent(d, s); }
-
-//========================================================================================
-// Impl Method Bodies
-//========================================================================================
-
-AgentSessionImpl::AgentSessionImpl(Connection& c, const string& options) :
-    connection(c), domain("default"), opened(false), thread(0), threadCanceled(false),
-    bootSequence(1), interval(60), lastHeartbeat(0), lastVisit(0), forceHeartbeat(false),
-    externalStorage(false), autoAllowQueries(true), autoAllowMethods(true),
-    maxSubscriptions(64), minSubInterval(3000), subLifetime(300), publicEvents(true),
-    listenOnDirect(true), strictSecurity(false),
-    schemaUpdateTime(uint64_t(qpid::sys::Duration(qpid::sys::EPOCH, qpid::sys::now())))
-{
-    //
-    // Set Agent Capability Level
-    //
-    attributes["qmf.agent_capability"] = AGENT_CAPABILITY_0_8;
-
-    if (!options.empty()) {
-        qpid::messaging::AddressParser parser(options);
-        Variant::Map optMap;
-        Variant::Map::const_iterator iter;
-
-        parser.parseMap(optMap);
-
-        iter = optMap.find("domain");
-        if (iter != optMap.end())
-            domain = iter->second.asString();
-
-        iter = optMap.find("interval");
-        if (iter != optMap.end()) {
-            interval = iter->second.asUint32();
-            if (interval < 1)
-                interval = 1;
-        }
-
-        iter = optMap.find("external");
-        if (iter != optMap.end())
-            externalStorage = iter->second.asBool();
-
-        iter = optMap.find("allow-queries");
-        if (iter != optMap.end())
-            autoAllowQueries = iter->second.asBool();
-
-        iter = optMap.find("allow-methods");
-        if (iter != optMap.end())
-            autoAllowMethods = iter->second.asBool();
-
-        iter = optMap.find("max-subscriptions");
-        if (iter != optMap.end())
-            maxSubscriptions = iter->second.asUint32();
-
-        iter = optMap.find("min-sub-interval");
-        if (iter != optMap.end())
-            minSubInterval = iter->second.asUint32();
-
-        iter = optMap.find("sub-lifetime");
-        if (iter != optMap.end())
-            subLifetime = iter->second.asUint32();
-
-        iter = optMap.find("public-events");
-        if (iter != optMap.end())
-            publicEvents = iter->second.asBool();
-
-        iter = optMap.find("listen-on-direct");
-        if (iter != optMap.end())
-            listenOnDirect = iter->second.asBool();
-
-        iter = optMap.find("strict-security");
-        if (iter != optMap.end())
-            strictSecurity = iter->second.asBool();
-    }
-}
-
-
-AgentSessionImpl::~AgentSessionImpl()
-{
-    if (opened)
-        close();
-}
-
-
-void AgentSessionImpl::open()
-{
-    if (opened)
-        throw QmfException("The session is already open");
-
-    const string addrArgs(";{create:never,node:{type:topic}}");
-    const string routableAddr("direct-agent.route." + qpid::types::Uuid(true).str());
-    attributes["_direct_subject"] = routableAddr;
-
-    // Establish messaging addresses
-    setAgentName();
-    directBase = "qmf." + domain + ".direct";
-    topicBase = "qmf." + domain + ".topic";
-
-    // Create AMQP session, receivers, and senders
-    session = connection.createSession();
-    Receiver directRx;
-    Receiver routableDirectRx = session.createReceiver(topicBase + "/" + routableAddr + addrArgs);
-    Receiver topicRx = session.createReceiver(topicBase + "/console.#" + addrArgs);
-
-    if (listenOnDirect && !strictSecurity) {
-        directRx = session.createReceiver(directBase + "/" + agentName + addrArgs);
-        directRx.setCapacity(64);
-    }
-
-    routableDirectRx.setCapacity(64);
-    topicRx.setCapacity(64);
-
-    if (!strictSecurity)
-        directSender = session.createSender(directBase + addrArgs);
-    topicSender = session.createSender(topicBase + addrArgs);
-
-    // Start the receiver thread
-    threadCanceled = false;
-    opened = true;
-    thread = new qpid::sys::Thread(*this);
-
-    // Send an initial agent heartbeat message
-    sendHeartbeat();
-}
-
-
-void AgentSessionImpl::close()
-{
-    if (!opened)
-        return;
-
-    // Stop and join the receiver thread
-    threadCanceled = true;
-    thread->join();
-    delete thread;
-
-    // Close the AMQP session
-    session.close();
-    opened = false;
-}
-
-
-bool AgentSessionImpl::nextEvent(AgentEvent& event, Duration timeout)
-{
-    uint64_t milliseconds = timeout.getMilliseconds();
-    qpid::sys::Mutex::ScopedLock l(lock);
-
-    if (eventQueue.empty() && milliseconds > 0)
-        cond.wait(lock, qpid::sys::AbsTime(qpid::sys::now(),
-                                           qpid::sys::Duration(milliseconds * qpid::sys::TIME_MSEC)));
-
-    if (!eventQueue.empty()) {
-        event = eventQueue.front();
-        eventQueue.pop();
-        return true;
-    }
-
-    return false;
-}
-
-
-int AgentSessionImpl::pendingEvents() const
-{
-    qpid::sys::Mutex::ScopedLock l(lock);
-    return eventQueue.size();
-}
-
-
-void AgentSessionImpl::registerSchema(Schema& schema)
-{
-    if (!schema.isFinalized())
-        schema.finalize();
-    const SchemaId& schemaId(schema.getSchemaId());
-
-    qpid::sys::Mutex::ScopedLock l(lock);
-    schemata[schemaId] = schema;
-    schemaIndex[schemaId] = DataIndex();
-
-    //
-    // Get the news out at the next periodic interval that there is new schema information.
-    //
-    schemaUpdateTime = uint64_t(qpid::sys::Duration(qpid::sys::EPOCH, qpid::sys::now()));
-    forceHeartbeat = true;
-}
-
-
-DataAddr AgentSessionImpl::addData(Data& data, const string& name, bool persistent)
-{
-    if (externalStorage)
-        throw QmfException("addData() must not be called when the 'external' option is enabled.");
-
-    string dataName;
-    if (name.empty())
-        dataName = qpid::types::Uuid(true).str();
-    else
-        dataName = name;
-
-    DataAddr addr(dataName, agentName, persistent ? 0 : bootSequence);
-    data.setAddr(addr);
-
-    {
-        qpid::sys::Mutex::ScopedLock l(lock);
-        DataIndex::const_iterator iter = globalIndex.find(addr);
-        if (iter != globalIndex.end())
-            throw QmfException("Duplicate Data Address");
-
-        globalIndex[addr] = data;
-        if (data.hasSchema())
-            schemaIndex[data.getSchemaId()][addr] = data;
-    }
-
-    //
-    // TODO: Survey active subscriptions to see if they need to hear about this new data.
-    //
-
-    return addr;
-}
-
-
-void AgentSessionImpl::delData(const DataAddr& addr)
-{
-    {
-        qpid::sys::Mutex::ScopedLock l(lock);
-        DataIndex::iterator iter = globalIndex.find(addr);
-        if (iter == globalIndex.end())
-            return;
-        if (iter->second.hasSchema()) {
-            const SchemaId& schemaId(iter->second.getSchemaId());
-            schemaIndex[schemaId].erase(addr);
-        }
-        globalIndex.erase(iter);
-    }
-
-    //
-    // TODO: Survey active subscriptions to see if they need to hear about this deleted data.
-    //
-}
-
-
-void AgentSessionImpl::authAccept(AgentEvent& authEvent)
-{
-    auto_ptr<AgentEventImpl> eventImpl(new AgentEventImpl(AGENT_QUERY));
-    eventImpl->setQuery(authEvent.getQuery());
-    eventImpl->setUserId(authEvent.getUserId());
-    eventImpl->setReplyTo(AgentEventImplAccess::get(authEvent).getReplyTo());
-    eventImpl->setCorrelationId(AgentEventImplAccess::get(authEvent).getCorrelationId());
-    AgentEvent event(eventImpl.release());
-
-    if (externalStorage) {
-        enqueueEvent(event);
-        return;
-    }
-
-    const Query& query(authEvent.getQuery());
-    if (query.getDataAddr().isValid()) {
-        {
-            qpid::sys::Mutex::ScopedLock l(lock);
-            DataIndex::const_iterator iter = globalIndex.find(query.getDataAddr());
-            if (iter != globalIndex.end())
-                response(event, iter->second);
-        }
-        complete(event);
-        return;
-    }
-
-    if (query.getSchemaId().isValid()) {
-        {
-            qpid::sys::Mutex::ScopedLock l(lock);
-            map<SchemaId, DataIndex>::const_iterator iter = schemaIndex.find(query.getSchemaId());
-            if (iter != schemaIndex.end())
-                for (DataIndex::const_iterator dIter = iter->second.begin(); dIter != iter->second.end(); dIter++)
-                    if (query.matchesPredicate(dIter->second.getProperties()))
-                        response(event, dIter->second);
-        }
-        complete(event);
-        return;
-    }
-
-    raiseException(event, "Query is Invalid");
-}
-
-
-void AgentSessionImpl::authReject(AgentEvent& event, const string& error)
-{
-    raiseException(event, "Action Forbidden - " + error);
-}
-
-
-void AgentSessionImpl::raiseException(AgentEvent& event, const string& error)
-{
-    Data exception(new DataImpl());
-    exception.setProperty("error_text", error);
-    raiseException(event, exception);
-}
-
-
-void AgentSessionImpl::raiseException(AgentEvent& event, const Data& data)
-{
-    Message msg;
-    Variant::Map map;
-    Variant::Map& headers(msg.getProperties());
-
-    headers[protocol::HEADER_KEY_METHOD] = protocol::HEADER_METHOD_RESPONSE;
-    headers[protocol::HEADER_KEY_OPCODE] = protocol::HEADER_OPCODE_EXCEPTION;
-    headers[protocol::HEADER_KEY_CONTENT] = protocol::HEADER_CONTENT_DATA;
-    headers[protocol::HEADER_KEY_AGENT] = agentName;
-    headers[protocol::HEADER_KEY_APP_ID] = protocol::HEADER_APP_ID_QMF;
-
-    AgentEventImpl& eventImpl(AgentEventImplAccess::get(event));
-    const DataImpl& dataImpl(DataImplAccess::get(data));
-
-    msg.setCorrelationId(eventImpl.getCorrelationId());
-    encode(dataImpl.asMap(), msg);
-    send(msg, eventImpl.getReplyTo());
-
-    QPID_LOG(trace, "SENT Exception to=" << eventImpl.getReplyTo());
-}
-
-
-void AgentSessionImpl::response(AgentEvent& event, const Data& data)
-{
-    AgentEventImpl& impl(AgentEventImplAccess::get(event));
-    uint32_t count = impl.enqueueData(data);
-    if (count >= 8)
-        flushResponses(event, false);
-}
-
-
-void AgentSessionImpl::complete(AgentEvent& event)
-{
-    flushResponses(event, true);
-}
-
-
-void AgentSessionImpl::methodSuccess(AgentEvent& event)
-{
-    Message msg;
-    Variant::Map map;
-    Variant::Map& headers(msg.getProperties());
-
-    headers[protocol::HEADER_KEY_METHOD] = protocol::HEADER_METHOD_RESPONSE;
-    headers[protocol::HEADER_KEY_OPCODE] = protocol::HEADER_OPCODE_METHOD_RESPONSE;
-    headers[protocol::HEADER_KEY_AGENT] = agentName;
-    headers[protocol::HEADER_KEY_APP_ID] = protocol::HEADER_APP_ID_QMF;
-
-    AgentEventImpl& eventImpl(AgentEventImplAccess::get(event));
-
-    const Variant::Map& outArgs(eventImpl.getReturnArguments());
-    const Variant::Map& outSubtypes(eventImpl.getReturnArgumentSubtypes());
-
-    map["_arguments"] = outArgs;
-    if (!outSubtypes.empty())
-        map["_subtypes"] = outSubtypes;
-
-    msg.setCorrelationId(eventImpl.getCorrelationId());
-    encode(map, msg);
-    send(msg, eventImpl.getReplyTo());
-
-    QPID_LOG(trace, "SENT MethodResponse to=" << eventImpl.getReplyTo());
-}
-
-
-void AgentSessionImpl::raiseEvent(const Data& data)
-{
-    int severity(SEV_NOTICE);
-    if (data.hasSchema()) {
-        const Schema& schema(DataImplAccess::get(data).getSchema());
-        if (schema.isValid())
-            severity = schema.getDefaultSeverity();
-    }
-
-    raiseEvent(data, severity);
-}
-
-
-void AgentSessionImpl::raiseEvent(const Data& data, int severity)
-{
-    Message msg;
-    Variant::Map map;
-    Variant::Map& headers(msg.getProperties());
-    string subject("agent.ind.event");
-
-    if (data.hasSchema()) {
-        const SchemaId& schemaId(data.getSchemaId());
-        if (schemaId.getType() != SCHEMA_TYPE_EVENT)
-            throw QmfException("Cannot call raiseEvent on data that is not an Event");
-        subject = subject + "." + schemaId.getPackageName() + "." + schemaId.getName();
-    }
-
-    if (severity < SEV_EMERG || severity > SEV_DEBUG)
-        throw QmfException("Invalid severity value");
-
-    headers[protocol::HEADER_KEY_METHOD] = protocol::HEADER_METHOD_INDICATION;
-    headers[protocol::HEADER_KEY_OPCODE] = protocol::HEADER_OPCODE_DATA_INDICATION;
-    headers[protocol::HEADER_KEY_CONTENT] = protocol::HEADER_CONTENT_EVENT;
-    headers[protocol::HEADER_KEY_AGENT] = agentName;
-    headers[protocol::HEADER_KEY_APP_ID] = protocol::HEADER_APP_ID_QMF;
-    msg.setSubject(subject);
-
-    Variant::List list;
-    Variant::Map dataAsMap(DataImplAccess::get(data).asMap());
-    dataAsMap["_severity"] = severity;
-    dataAsMap["_timestamp"] = uint64_t(qpid::sys::Duration(qpid::sys::EPOCH, qpid::sys::now()));
-    list.push_back(dataAsMap);
-    encode(list, msg);
-    topicSender.send(msg);
-
-    QPID_LOG(trace, "SENT EventIndication to=" << topicSender.getName() << "/" << subject);
-}
-
-
-void AgentSessionImpl::checkOpen()
-{
-    if (opened)
-        throw QmfException("Operation must be performed before calling open()");
-}
-
-
-void AgentSessionImpl::enqueueEvent(const AgentEvent& event)
-{
-    qpid::sys::Mutex::ScopedLock l(lock);
-    bool notify = eventQueue.empty();
-    eventQueue.push(event);
-    if (notify)
-        cond.notify();
-}
-
-
-void AgentSessionImpl::setAgentName()
-{
-    Variant::Map::iterator iter;
-    string vendor;
-    string product;
-    string instance;
-
-    iter = attributes.find("_vendor");
-    if (iter == attributes.end())
-        attributes["_vendor"] = vendor;
-    else
-        vendor = iter->second.asString();
-
-    iter = attributes.find("_product");
-    if (iter == attributes.end())
-        attributes["_product"] = product;
-    else
-        product = iter->second.asString();
-
-    iter = attributes.find("_instance");
-    if (iter == attributes.end()) {
-        instance = qpid::types::Uuid(true).str();
-        attributes["_instance"] = instance;
-    } else
-        instance = iter->second.asString();
-
-    agentName = vendor + ":" + product + ":" + instance;
-    attributes["_name"] = agentName;
-}
-
-
-void AgentSessionImpl::handleLocateRequest(const Variant::List& predicate, const Message& msg)
-{
-    QPID_LOG(trace, "RCVD AgentLocateRequest from=" << msg.getReplyTo());
-
-    if (!predicate.empty()) {
-        Query agentQuery(QUERY_OBJECT);
-        agentQuery.setPredicate(predicate);
-        if (!agentQuery.matchesPredicate(attributes)) {
-            QPID_LOG(trace, "AgentLocate predicate does not match this agent, ignoring");
-            return;
-        }
-    }
-
-    Message reply;
-    Variant::Map map;
-    Variant::Map& headers(reply.getProperties());
-
-    headers[protocol::HEADER_KEY_METHOD] = protocol::HEADER_METHOD_INDICATION;
-    headers[protocol::HEADER_KEY_OPCODE] = protocol::HEADER_OPCODE_AGENT_LOCATE_RESPONSE;
-    headers[protocol::HEADER_KEY_AGENT] = agentName;
-    headers[protocol::HEADER_KEY_APP_ID] = protocol::HEADER_APP_ID_QMF;
-
-    map["_values"] = attributes;
-    map["_values"].asMap()[protocol::AGENT_ATTR_TIMESTAMP] = uint64_t(qpid::sys::Duration(qpid::sys::EPOCH, qpid::sys::now()));
-    map["_values"].asMap()[protocol::AGENT_ATTR_HEARTBEAT_INTERVAL] = interval;
-    map["_values"].asMap()[protocol::AGENT_ATTR_EPOCH] = bootSequence;
-    map["_values"].asMap()[protocol::AGENT_ATTR_SCHEMA_UPDATED_TIMESTAMP] = schemaUpdateTime;
-
-    encode(map, reply);
-    send(reply, msg.getReplyTo());
-    QPID_LOG(trace, "SENT AgentLocateResponse to=" << msg.getReplyTo());
-}
-
-
-void AgentSessionImpl::handleMethodRequest(const Variant::Map& content, const Message& msg)
-{
-    QPID_LOG(trace, "RCVD MethodRequest map=" << content << " from=" << msg.getReplyTo() << " cid=" << msg.getCorrelationId());
-
-    //
-    // Construct an AgentEvent to be sent to the application.
-    //
-    auto_ptr<AgentEventImpl> eventImpl(new AgentEventImpl(AGENT_METHOD));
-    eventImpl->setUserId(msg.getUserId());
-    eventImpl->setReplyTo(msg.getReplyTo());
-    eventImpl->setCorrelationId(msg.getCorrelationId());
-
-    Variant::Map::const_iterator iter;
-
-    iter = content.find("_method_name");
-    if (iter == content.end()) {
-        AgentEvent event(eventImpl.release());
-        raiseException(event, "Malformed MethodRequest: missing _method_name field");
-        return;
-    }
-    eventImpl->setMethodName(iter->second.asString());
-
-    iter = content.find("_arguments");
-    if (iter != content.end())
-        eventImpl->setArguments(iter->second.asMap());
-
-    iter = content.find("_subtypes");
-    if (iter != content.end())
-        eventImpl->setArgumentSubtypes(iter->second.asMap());
-
-    iter = content.find("_object_id");
-    if (iter != content.end()) {
-        DataAddr addr(new DataAddrImpl(iter->second.asMap()));
-        eventImpl->setDataAddr(addr);
-        DataIndex::const_iterator iter(globalIndex.find(addr));
-        if (iter == globalIndex.end()) {
-            AgentEvent event(eventImpl.release());
-            raiseException(event, "No data object found with the specified address");
-            return;
-        }
-
-        const Schema& schema(DataImplAccess::get(iter->second).getSchema());
-        if (schema.isValid()) {
-            eventImpl->setSchema(schema);
-            for (Variant::Map::const_iterator aIter = eventImpl->getArguments().begin();
-                 aIter != eventImpl->getArguments().end(); aIter++) {
-                const Schema& schema(DataImplAccess::get(iter->second).getSchema());
-                if (!SchemaImplAccess::get(schema).isValidMethodInArg(eventImpl->getMethodName(), aIter->first, aIter->second)) {
-                    AgentEvent event(eventImpl.release());
-                    raiseException(event, "Invalid argument: " + aIter->first);
-                    return;
-                }
-            }
-        }
-    }
-
-    enqueueEvent(AgentEvent(eventImpl.release()));
-}
-
-
-void AgentSessionImpl::handleQueryRequest(const Variant::Map& content, const Message& msg)
-{
-    QPID_LOG(trace, "RCVD QueryRequest query=" << content << " from=" << msg.getReplyTo() << " cid=" << msg.getCorrelationId());
-
-    //
-    // Construct an AgentEvent to be sent to the application or directly handled by the agent.
-    //
-    auto_ptr<QueryImpl> queryImpl(new QueryImpl(content));
-    auto_ptr<AgentEventImpl> eventImpl(new AgentEventImpl(AGENT_AUTH_QUERY));
-    eventImpl->setUserId(msg.getUserId());
-    eventImpl->setReplyTo(msg.getReplyTo());
-    eventImpl->setCorrelationId(msg.getCorrelationId());
-    eventImpl->setQuery(queryImpl.release());
-    AgentEvent ae(eventImpl.release());
-
-    if (ae.getQuery().getTarget() == QUERY_SCHEMA_ID || ae.getQuery().getTarget() == QUERY_SCHEMA) {
-        handleSchemaRequest(ae);
-        return;
-    }
-
-    if (autoAllowQueries)
-        authAccept(ae);
-    else
-        enqueueEvent(ae);
-}
-
-
-void AgentSessionImpl::handleSchemaRequest(AgentEvent& event)
-{
-    SchemaMap::const_iterator iter;
-    string error;
-    const Query& query(event.getQuery());
-
-    Message msg;
-    Variant::List content;
-    Variant::Map map;
-    Variant::Map& headers(msg.getProperties());
-
-    headers[protocol::HEADER_KEY_METHOD] = protocol::HEADER_METHOD_RESPONSE;
-    headers[protocol::HEADER_KEY_OPCODE] = protocol::HEADER_OPCODE_QUERY_RESPONSE;
-    headers[protocol::HEADER_KEY_AGENT] = agentName;
-    headers[protocol::HEADER_KEY_APP_ID] = protocol::HEADER_APP_ID_QMF;
-
-    {
-        qpid::sys::Mutex::ScopedLock l(lock);
-        if (query.getTarget() == QUERY_SCHEMA_ID) {
-            headers[protocol::HEADER_KEY_CONTENT] = "_schema_id";
-            for (iter = schemata.begin(); iter != schemata.end(); iter++)
-                content.push_back(SchemaIdImplAccess::get(iter->first).asMap());
-        } else if (query.getSchemaId().isValid()) {
-            headers[protocol::HEADER_KEY_CONTENT] = "_schema";
-            iter = schemata.find(query.getSchemaId());
-            if (iter != schemata.end())
-                content.push_back(SchemaImplAccess::get(iter->second).asMap());
-        } else {
-            error = "Invalid Schema Query: Requests for SCHEMA must supply a valid schema ID.";
-        }
-    }
-
-    if (!error.empty()) {
-        raiseException(event, error);
-        return;
-    }
-
-    AgentEventImpl& eventImpl(AgentEventImplAccess::get(event));
-
-    msg.setCorrelationId(eventImpl.getCorrelationId());
-    encode(content, msg);
-    send(msg, eventImpl.getReplyTo());
-
-    QPID_LOG(trace, "SENT QueryResponse(Schema) to=" << eventImpl.getReplyTo());
-}
-
-
-void AgentSessionImpl::handleV1SchemaRequest(qpid::management::Buffer& buffer, uint32_t seq, const Message& msg)
-{
-    string packageName;
-    string className;
-    uint8_t hashBits[16];
-
-    buffer.getShortString(packageName);
-    buffer.getShortString(className);
-    buffer.getBin128(hashBits);
-
-    QPID_LOG(trace, "RCVD QMFv1 SchemaRequest for " << packageName << ":" << className);
-
-    qpid::types::Uuid hash(hashBits);
-    map<SchemaId, Schema>::const_iterator iter;
-    string replyContent;
-
-    SchemaId dataId(SCHEMA_TYPE_DATA, packageName, className);
-    dataId.setHash(hash);
-
-    {
-        qpid::sys::Mutex::ScopedLock l(lock);
-        iter = schemata.find(dataId);
-        if (iter != schemata.end())
-            replyContent = SchemaImplAccess::get(iter->second).asV1Content(seq);
-        else {
-            SchemaId eventId(SCHEMA_TYPE_EVENT, packageName, className);
-            eventId.setHash(hash);
-            iter = schemata.find(dataId);
-            if (iter != schemata.end())
-                replyContent = SchemaImplAccess::get(iter->second).asV1Content(seq);
-            else
-                return;
-        }
-    }
-
-    Message reply;
-    Variant::Map& headers(reply.getProperties());
-
-    headers[protocol::HEADER_KEY_AGENT] = agentName;
-    reply.setContent(replyContent);
-
-    send(reply, msg.getReplyTo());
-    QPID_LOG(trace, "SENT QMFv1 SchemaResponse to=" << msg.getReplyTo());
-}
-
-
-void AgentSessionImpl::dispatch(Message msg)
-{
-    const Variant::Map& properties(msg.getProperties());
-    Variant::Map::const_iterator iter;
-
-    //
-    // If strict-security is enabled, make sure that reply-to address complies with the
-    // strict-security addressing pattern (i.e. start with 'qmf.<domain>.topic/direct-console.').
-    //
-    if (strictSecurity && msg.getReplyTo()) {
-        if (msg.getReplyTo().getName() != topicBase || msg.getReplyTo().getSubject().find("direct-console.") != 0) {
-            QPID_LOG(warning, "Reply-to violates strict-security policy: " << msg.getReplyTo().str());
-            return;
-        }
-    }
-
-    iter = properties.find(protocol::HEADER_KEY_APP_ID);
-    if (iter != properties.end() && iter->second.asString() == protocol::HEADER_APP_ID_QMF) {
-        //
-        // Dispatch a QMFv2 formatted message
-        //
-        iter = properties.find(protocol::HEADER_KEY_OPCODE);
-        if (iter == properties.end()) {
-            QPID_LOG(trace, "Message received with no 'qmf.opcode' header");
-            return;
-        }
-
-        const string& opcode = iter->second.asString();
-
-        if (msg.getContentType() == "amqp/list") {
-            Variant::List content;
-            decode(msg, content);
-
-            if (opcode == protocol::HEADER_OPCODE_AGENT_LOCATE_REQUEST) handleLocateRequest(content, msg);
-            else {
-                QPID_LOG(trace, "Unexpected QMFv2 opcode with 'amqp/list' content: " << opcode);
-            }
-
-        } else if (msg.getContentType() == "amqp/map") {
-            Variant::Map content;
-            decode(msg, content);
-
-            if      (opcode == protocol::HEADER_OPCODE_METHOD_REQUEST) handleMethodRequest(content, msg);
-            else if (opcode == protocol::HEADER_OPCODE_QUERY_REQUEST)  handleQueryRequest(content, msg);
-            else {
-                QPID_LOG(trace, "Unexpected QMFv2 opcode with 'amqp/map' content: " << opcode);
-            }
-        } else {
-            QPID_LOG(trace, "Unexpected QMFv2 content type.  Expected amqp/list or amqp/map");
-        }
-
-    } else {
-        //
-        // Dispatch a QMFv1 formatted message
-        //
-        const string& body(msg.getContent());
-        if (body.size() < 8)
-            return;
-        qpid::management::Buffer buffer(const_cast<char*>(body.c_str()), body.size());
-
-        if (buffer.getOctet() != 'A') return;
-        if (buffer.getOctet() != 'M') return;
-        if (buffer.getOctet() != '2') return;
-        char v1Opcode(buffer.getOctet());
-        uint32_t seq(buffer.getLong());
-
-        if (v1Opcode == 'S') handleV1SchemaRequest(buffer, seq, msg);
-        else {
-            QPID_LOG(trace, "Unknown or Unsupported QMFv1 opcode: " << v1Opcode);
-        }
-    }
-}
-
-
-void AgentSessionImpl::sendHeartbeat()
-{
-    Message msg;
-    Variant::Map map;
-    Variant::Map& headers(msg.getProperties());
-    std::stringstream address;
-
-    address << "agent.ind.heartbeat";
-
-    // append .<vendor>.<product> to address key if present.
-    Variant::Map::const_iterator v;
-    if ((v = attributes.find("_vendor")) != attributes.end() && !v->second.getString().empty()) {
-        address << "." << v->second.getString();
-        if ((v = attributes.find("_product")) != attributes.end() && !v->second.getString().empty()) {
-            address << "." << v->second.getString();
-        }
-    }
-
-    headers[protocol::HEADER_KEY_METHOD] = protocol::HEADER_METHOD_INDICATION;
-    headers[protocol::HEADER_KEY_OPCODE] = protocol::HEADER_OPCODE_AGENT_HEARTBEAT_INDICATION;
-    headers[protocol::HEADER_KEY_AGENT] = agentName;
-    headers[protocol::HEADER_KEY_APP_ID] = protocol::HEADER_APP_ID_QMF;
-    msg.setSubject(address.str());
-
-    map["_values"] = attributes;
-    map["_values"].asMap()[protocol::AGENT_ATTR_TIMESTAMP] = uint64_t(qpid::sys::Duration(qpid::sys::EPOCH, qpid::sys::now()));
-    map["_values"].asMap()[protocol::AGENT_ATTR_HEARTBEAT_INTERVAL] = interval;
-    map["_values"].asMap()[protocol::AGENT_ATTR_EPOCH] = bootSequence;
-    map["_values"].asMap()[protocol::AGENT_ATTR_SCHEMA_UPDATED_TIMESTAMP] = schemaUpdateTime;
-
-    encode(map, msg);
-    topicSender.send(msg);
-    QPID_LOG(trace, "SENT AgentHeartbeat name=" << agentName);
-}
-
-
-void AgentSessionImpl::send(Message msg, const Address& to)
-{
-    Sender sender;
-
-    if (strictSecurity && to.getName() != topicBase) {
-        QPID_LOG(warning, "Address violates strict-security policy: " << to);
-        return;
-    }
-
-    if (to.getName() == directBase) {
-        msg.setSubject(to.getSubject());
-        sender = directSender;
-    } else if (to.getName() == topicBase) {
-        msg.setSubject(to.getSubject());
-        sender = topicSender;
-    } else
-        sender = session.createSender(to);
-
-    sender.send(msg);
-}
-
-
-void AgentSessionImpl::flushResponses(AgentEvent& event, bool final)
-{
-    Message msg;
-    Variant::Map map;
-    Variant::Map& headers(msg.getProperties());
-
-    headers[protocol::HEADER_KEY_METHOD] = protocol::HEADER_METHOD_RESPONSE;
-    headers[protocol::HEADER_KEY_OPCODE] = protocol::HEADER_OPCODE_QUERY_RESPONSE;
-    headers[protocol::HEADER_KEY_CONTENT] = protocol::HEADER_CONTENT_DATA;
-    headers[protocol::HEADER_KEY_AGENT] = agentName;
-    headers[protocol::HEADER_KEY_APP_ID] = protocol::HEADER_APP_ID_QMF;
-    if (!final)
-        headers[protocol::HEADER_KEY_PARTIAL] = Variant();
-
-    Variant::List body;
-    AgentEventImpl& eventImpl(AgentEventImplAccess::get(event));
-    Data data(eventImpl.dequeueData());
-    while (data.isValid()) {
-        DataImpl& dataImpl(DataImplAccess::get(data));
-        body.push_back(dataImpl.asMap());
-        data = eventImpl.dequeueData();
-    }
-
-    msg.setCorrelationId(eventImpl.getCorrelationId());
-    encode(body, msg);
-    send(msg, eventImpl.getReplyTo());
-
-    QPID_LOG(trace, "SENT QueryResponse to=" << eventImpl.getReplyTo());
-}
-
-
-void AgentSessionImpl::periodicProcessing(uint64_t seconds)
-{
-    //
-    // The granularity of this timer is seconds.  Don't waste time looking for work if
-    // it's been less than a second since we last visited.
-    //
-    if (seconds == lastVisit)
-        return;
-    //uint64_t thisInterval(seconds - lastVisit);
-    lastVisit = seconds;
-
-    //
-    // First time through, set lastHeartbeat to the current time.
-    //
-    if (lastHeartbeat == 0)
-        lastHeartbeat = seconds;
-
-    //
-    // If the hearbeat interval has elapsed, send a heartbeat.
-    //
-    if (forceHeartbeat || (seconds - lastHeartbeat >= interval)) {
-        lastHeartbeat = seconds;
-        forceHeartbeat = false;
-        sendHeartbeat();
-    }
-
-    //
-    // TODO: process any active subscriptions on their intervals.
-    //
-}
-
-
-void AgentSessionImpl::run()
-{
-    QPID_LOG(debug, "AgentSession thread started for agent " << agentName);
-
-    try {
-        while (!threadCanceled) {
-            periodicProcessing((uint64_t) qpid::sys::Duration(qpid::sys::EPOCH, qpid::sys::now()) / qpid::sys::TIME_SEC);
-
-            Receiver rx;
-            bool valid = session.nextReceiver(rx, Duration::SECOND);
-            if (threadCanceled)
-                break;
-            if (valid) {
-                try {
-                    dispatch(rx.fetch());
-                } catch (qpid::types::Exception& e) {
-                    QPID_LOG(error, "Exception caught in message dispatch: " << e.what());
-                }
-                session.acknowledge();
-            }
-        }
-    } catch (qpid::types::Exception& e) {
-        QPID_LOG(error, "Exception caught in message thread - exiting: " << e.what());
-        enqueueEvent(AgentEvent(new AgentEventImpl(AGENT_THREAD_FAILED)));
-    }
-
-    QPID_LOG(debug, "AgentSession thread exiting for agent " << agentName);
-}
-