path: root/cpp/src/qmf/engine/Agent.cpp

/*
 * 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/EventImpl.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 <qpid/framing/Buffer.h>
#include <qpid/framing/Uuid.h>
#include <qpid/framing/FieldTable.h>
#include <qpid/framing/FieldValue.h>
#include <qpid/sys/Mutex.h>
#include <qpid/log/Statement.h>
#include <qpid/sys/Time.h>
#include <string.h>
#include <string>
#include <deque>
#include <map>
#include <iostream>
#include <fstream>
#include <boost/shared_ptr.hpp>
#include <boost/noncopyable.hpp>

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<AgentEventImpl> Ptr;
        AgentEvent::EventKind kind;
        uint32_t    sequence;
        string      authUserId;
        string      authToken;
        string      name;
        Object*     object;
        boost::shared_ptr<ObjectId> objectId;
        boost::shared_ptr<Query> query;
        boost::shared_ptr<Value> 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<AgentQueryContext> 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<AgentEventImpl::Ptr> eventQueue;
        deque<MessageImpl::Ptr> xmtQueue;
        map<uint32_t, AgentQueryContext::Ptr> contextMap;
        bool      attachComplete;

        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<AgentClassKey, SchemaObjectClass*, AgentClassKeyComp> ObjectClassMap;
        typedef map<AgentClassKey, SchemaEventClass*, AgentClassKeyComp>  EventClassMap;

        struct ClassMaps {
            ObjectClassMap objectClasses;
            EventClassMap  eventClasses;
        };

        map<string, ClassMaps> 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<ObjectId> oid);
        AgentEventImpl::Ptr eventMethod(uint32_t num, const string& userId, const string& method,
                                        boost::shared_ptr<ObjectId> oid, boost::shared_ptr<Value> 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<char*>(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), attachComplete(false)
{
    queueName += Uuid(true).str();
}

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(label);
    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(EPOCH, 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<uint32_t, AgentQueryContext::Ptr>::iterator iter = contextMap.find(sequence);
    if (iter == contextMap.end())
        return;
    AgentQueryContext::Ptr context = iter->second;
    contextMap.erase(iter);

    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 SchemaArgument*>::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);
    if (status == 0) {
        for (vector<const SchemaArgument*>::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);
    if (allocated)
        free(buf);
    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<uint32_t, AgentQueryContext::Ptr>::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<uint32_t, AgentQueryContext::Ptr>::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);
    bool newPackage = false;

    map<string, ClassMaps>::iterator iter = packages.find(cls->getClassKey()->getPackageName());
    if (iter == packages.end()) {
        packages[cls->getClassKey()->getPackageName()] = ClassMaps();
        iter = packages.find(cls->getClassKey()->getPackageName());
        newPackage = true;
    }

    AgentClassKey key(cls->getClassKey()->getClassName(), cls->getClassKey()->getHash());
    iter->second.objectClasses[key] = cls;

    // 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<string, ClassMaps>::iterator iter = packages.find(cls->getClassKey()->getPackageName());
    if (iter == packages.end()) {
        packages[cls->getClassKey()->getPackageName()] = ClassMaps();
        iter = packages.find(cls->getClassKey()->getPackageName());
        newPackage = true;
    }

    AgentClassKey key(cls->getClassKey()->getClassName(), cls->getClassKey()->getHash());
    iter->second.eventClasses[key] = cls;

    // 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)
{
    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& 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(EPOCH, 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)
{
    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<ObjectId> 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<ObjectId> oid, boost::shared_ptr<Value> 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<uint8_t*>(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<string, ClassMaps>::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);
    }

    attachComplete = true;
}

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<string, ClassMaps>::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<string, ClassMaps>::const_iterator pIter = packages.end();
    string pname;
    string cname;
    string oidRepr;
    boost::shared_ptr<ObjectId> oid;

    ft.decode(inBuffer);
    
    QPID_LOG(trace, "RCVD GetQuery: seq=" << sequence << " map=" << ft);

    value = ft.get("_package");
    if (value.get() && value->convertsTo<string>()) {
        pname = value->get<string>();
        pIter = packages.find(pname);
        if (pIter == packages.end()) {
            sendCommandCompleteLH(DIR_EXCHANGE, replyTo, sequence);
            return;
        }
    }

    value = ft.get("_class");
    if (value.get() && value->convertsTo<string>()) {
        cname = value->get<string>();
        // 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<string>()) {
        oidRepr = value->get<string>();
        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<ObjectId> oid(ObjectIdImpl::factory(buffer));
    buffer.getShortString(pname);
    AgentClassKey classKey(buffer);
    buffer.getShortString(method);

    QPID_LOG(trace, "RCVD MethodRequest seq=" << sequence << " method=" << method);

    map<string, ClassMaps>::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 SchemaMethod*>::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<Value> argMap(new Value(TYPE_MAP));
    Value* value;
    for (vector<const SchemaArgument*>::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); }