/*
 *
 * 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/console/SessionManager.h"
#include "qpid/console/Schema.h"
#include "qpid/console/Agent.h"
#include "qpid/console/ConsoleListener.h"
#include "qpid/log/Statement.h"
#include "qpid/sys/Time.h"
#include "qpid/framing/Buffer.h"
#include "qpid/framing/Uuid.h"
#include "qpid/framing/FieldTable.h"

using namespace qpid::console;
using namespace qpid::sys;
using namespace qpid;
using namespace std;
using qpid::framing::Buffer;
using qpid::framing::FieldTable;

SessionManager::SessionManager(ConsoleListener* _listener, Settings _settings) :
    listener(_listener), settings(_settings)
{
    bindingKeys();
}

SessionManager::~SessionManager()
{
    for (vector<Broker*>::iterator iter = brokers.begin();
         iter != brokers.end(); iter++)
        delete *iter;

    for (map<string, Package*>::iterator iter = packages.begin();
         iter != packages.end(); iter++) {
        for (Package::ClassMap::iterator citer = iter->second->classes.begin();
             citer != iter->second->classes.end();
             citer++)
            delete citer->second;
        delete iter->second;
    }
}

Broker* SessionManager::addBroker(client::ConnectionSettings& settings)
{
    Broker* broker(new Broker(*this, settings));
    {
        Mutex::ScopedLock l(brokerListLock);
        brokers.push_back(broker);
    }
    return broker;
}

void SessionManager::delBroker(Broker* broker)
{
    Mutex::ScopedLock l(brokerListLock);
    for (vector<Broker*>::iterator iter = brokers.begin();
         iter != brokers.end(); iter++)
        if (*iter == broker) {
            brokers.erase(iter);
            delete broker;
            return;
        }
}

void SessionManager::getPackages(NameVector& packageNames)
{
    allBrokersStable();
    packageNames.clear();
    {
        Mutex::ScopedLock l(lock);
        for (map<string, Package*>::iterator iter = packages.begin();
             iter != packages.end(); iter++)
            packageNames.push_back(iter->first);
    }
}

void SessionManager::getClasses(KeyVector& classKeys, const std::string& packageName)
{
    allBrokersStable();
    classKeys.clear();
    map<string, Package*>::iterator iter = packages.find(packageName);
    if (iter == packages.end())
        return;

    Package& package = *(iter->second);
    for (Package::ClassMap::const_iterator piter = package.classes.begin();
         piter != package.classes.end(); piter++) {
        ClassKey key(piter->second->getClassKey());
        classKeys.push_back(key);
    }
}

SchemaClass& SessionManager::getSchema(const ClassKey& classKey)
{
    allBrokersStable();
    map<string, Package*>::iterator iter = packages.find(classKey.getPackageName());
    if (iter == packages.end())
        throw Exception("Unknown package");

    Package& package = *(iter->second);
    Package::NameHash key(classKey.getClassName(), classKey.getHash());
    Package::ClassMap::iterator cIter = package.classes.find(key);
    if (cIter == package.classes.end())
        throw Exception("Unknown class");

    return *(cIter->second);
}

void SessionManager::bindPackage(const std::string& packageName)
{
    stringstream key;
    key << "console.obj.*.*." << packageName << ".#";
    bindingKeyList.push_back(key.str());
    for (vector<Broker*>::iterator iter = brokers.begin(); iter != brokers.end(); iter++)
        (*iter)->addBinding(key.str());
}

void SessionManager::bindClass(const ClassKey& classKey)
{
    bindClass(classKey.getPackageName(), classKey.getClassName());
}

void SessionManager::bindClass(const std::string& packageName, const std::string& className)
{
    stringstream key;
    key << "console.obj.*.*." << packageName << "." << className << ".#";
    bindingKeyList.push_back(key.str());
    for (vector<Broker*>::iterator iter = brokers.begin();
         iter != brokers.end(); iter++)
        (*iter)->addBinding(key.str());
}


void SessionManager::bindEvent(const ClassKey& classKey)
{
    bindEvent(classKey.getPackageName(), classKey.getClassName());
}


void SessionManager::bindEvent(const std::string& packageName, const std::string& eventName)
{
    if (!settings.userBindings) throw Exception("Session not configured for userBindings.");
    if (settings.rcvEvents) throw Exception("Session already configured to receive all events.");

    stringstream key;
    key << "console.event.*.*." << packageName;
    if (eventName.length()) {
        key << "." << eventName << ".#";
    } else {
        key << ".#";
    }

    bindingKeyList.push_back(key.str());
    for (vector<Broker*>::iterator iter = brokers.begin();
         iter != brokers.end(); iter++)
        (*iter)->addBinding(key.str());
}


void SessionManager::getAgents(Agent::Vector& agents, Broker* broker)
{
    agents.clear();
    if (broker != 0) {
        broker->appendAgents(agents);
    } else {
        for (vector<Broker*>::iterator iter = brokers.begin(); iter != brokers.end(); iter++) {
            (*iter)->appendAgents(agents);
        }
    }
}

void SessionManager::getObjects(Object::Vector& objects, const std::string& className,
                                Broker* _broker, Agent* _agent)
{
    Agent::Vector agentList;

    if (_agent != 0) {
        agentList.push_back(_agent);
        _agent->getBroker()->waitForStable();
    } else {
        if (_broker != 0) {
            _broker->appendAgents(agentList);
            _broker->waitForStable();
        } else {
            allBrokersStable();
            Mutex::ScopedLock _lock(brokerListLock);
            for (vector<Broker*>::iterator iter = brokers.begin(); iter != brokers.end(); iter++) {
                (*iter)->appendAgents(agentList);
            }
        }
    }

    FieldTable ft;
    uint32_t sequence;
    ft.setString("_class", className);

    getResult.clear();
    syncSequenceList.clear();
    error = string();

    if (agentList.empty()) {
        objects = getResult;
        return;
    }

    for (Agent::Vector::iterator iter = agentList.begin(); iter != agentList.end(); iter++) {
        Agent* agent = *iter;
        char rawbuffer[512];
        Buffer buffer(rawbuffer, 512);
        stringstream routingKey;
        routingKey << "agent." << agent->getBrokerBank() << "." << agent->getAgentBank();
        {
            Mutex::ScopedLock _lock(lock);
            sequence = sequenceManager.reserve("multiget");
            syncSequenceList.insert(sequence);
        }
        agent->getBroker()->encodeHeader(buffer, 'G', sequence);
        ft.encode(buffer);
        uint32_t length = buffer.getPosition();
        buffer.reset();
        agent->getBroker()->connThreadBody.sendBuffer(buffer, length, "qpid.management", routingKey.str());
    }

    {
        Mutex::ScopedLock _lock(lock);
        sys::AbsTime startTime = sys::now();
        while (!syncSequenceList.empty() && error.empty()) {
            cv.wait(lock, AbsTime(now(), settings.getTimeout * TIME_SEC));
            sys::AbsTime currTime = sys::now();
            if (sys::Duration(startTime, currTime) > settings.getTimeout * TIME_SEC)
                break;
        }
    }

    objects = getResult;
}

void SessionManager::bindingKeys()
{
    bindingKeyList.push_back("schema.#");
    if (settings.rcvObjects && settings.rcvEvents && settings.rcvHeartbeats && !settings.userBindings) {
        bindingKeyList.push_back("console.#");
    } else {
        if (settings.rcvObjects && !settings.userBindings)
            bindingKeyList.push_back("console.obj.#");
        else
            bindingKeyList.push_back("console.obj.*.*.org.apache.qpid.broker.agent");
        if (settings.rcvEvents)
            bindingKeyList.push_back("console.event.#");
        if (settings.rcvHeartbeats)
            bindingKeyList.push_back("console.heartbeat");
    }
}

void SessionManager::allBrokersStable()
{
    Mutex::ScopedLock l(brokerListLock);
    for (vector<Broker*>::iterator iter = brokers.begin();
         iter != brokers.end(); iter++)
        if ((*iter)->isConnected())
            (*iter)->waitForStable();
}

void SessionManager::startProtocol(Broker* broker)
{
    char    rawbuffer[512];
    Buffer  buffer(rawbuffer, 512);

    broker->encodeHeader(buffer, 'B');
    uint32_t length = 512 - buffer.available();
    buffer.reset();
    broker->connThreadBody.sendBuffer(buffer, length);
}


void SessionManager::handleBrokerResp(Broker* broker, Buffer& inBuffer, uint32_t)
{
    framing::Uuid brokerId;

    brokerId.decode(inBuffer);
    broker->setBrokerId(brokerId);

    char    rawbuffer[512];
    Buffer  buffer(rawbuffer, 512);

    uint32_t sequence = sequenceManager.reserve("startup");
    broker->encodeHeader(buffer, 'P', sequence);
    uint32_t length = 512 - buffer.available();
    buffer.reset();
    broker->connThreadBody.sendBuffer(buffer, length);

    if (listener != 0) {
        listener->brokerInfo(*broker);
    }
}

void SessionManager::handlePackageInd(Broker* broker, Buffer& inBuffer, uint32_t)
{
    string packageName;
    inBuffer.getShortString(packageName);

    {
        Mutex::ScopedLock l(lock);
        map<string, Package*>::iterator iter = packages.find(packageName);
        if (iter == packages.end()) {
            packages[packageName] = new Package(packageName);
            if (listener != 0)
                listener->newPackage(packageName);
        }
    }

    broker->incOutstanding();
    char    rawbuffer[512];
    Buffer  buffer(rawbuffer, 512);

    uint32_t sequence = sequenceManager.reserve("startup");
    broker->encodeHeader(buffer, 'Q', sequence);
    buffer.putShortString(packageName);
    uint32_t length = 512 - buffer.available();
    buffer.reset();
    broker->connThreadBody.sendBuffer(buffer, length);
}

void SessionManager::handleCommandComplete(Broker* broker, Buffer& inBuffer, uint32_t sequence)
{
    Mutex::ScopedLock l(lock);
    uint32_t resultCode = inBuffer.getLong();
    string resultText;
    inBuffer.getShortString(resultText);
    string context = sequenceManager.release(sequence);
    if (resultCode != 0)
        QPID_LOG(debug, "Received error in completion: " << resultCode << " " << resultText);
    if (context == "startup") {
        broker->decOutstanding();
    } else if (context == "multiget") {
        if (syncSequenceList.count(sequence) == 1) {
            syncSequenceList.erase(sequence);
            if (syncSequenceList.empty()) {
                cv.notify();
            }
        }
    }
    // TODO: Other context cases
}

void SessionManager::handleClassInd(Broker* broker, Buffer& inBuffer, uint32_t)
{
    string packageName;
    string className;
    uint8_t hash[16];

    /*kind*/ (void) inBuffer.getOctet();
    inBuffer.getShortString(packageName);
    inBuffer.getShortString(className);
    inBuffer.getBin128(hash);

    {
        Mutex::ScopedLock l(lock);
        map<string, Package*>::iterator pIter = packages.find(packageName);
        if (pIter == packages.end() || pIter->second->getClass(className, hash))
            return;
    }

    broker->incOutstanding();
    char    rawbuffer[512];
    Buffer  buffer(rawbuffer, 512);

    uint32_t sequence = sequenceManager.reserve("startup");
    broker->encodeHeader(buffer, 'S', sequence);
    buffer.putShortString(packageName);
    buffer.putShortString(className);
    buffer.putBin128(hash);
    uint32_t length = 512 - buffer.available();
    buffer.reset();
    broker->connThreadBody.sendBuffer(buffer, length);
}

void SessionManager::handleMethodResp(Broker* broker, Buffer& buffer, uint32_t sequence)
{
    if (broker->methodObject) {
        broker->methodObject->handleMethodResp(buffer, sequence);
    }
}

void SessionManager::handleHeartbeatInd(Broker* /*broker*/, Buffer& /*inBuffer*/, uint32_t /*sequence*/)
{
}

void SessionManager::handleEventInd(Broker* broker, Buffer& buffer, uint32_t /*sequence*/)
{
    string packageName;
    string className;
    uint8_t hash[16];
    SchemaClass* schemaClass;

    buffer.getShortString(packageName);
    buffer.getShortString(className);
    buffer.getBin128(hash);

    {
        Mutex::ScopedLock l(lock);
        map<string, Package*>::iterator pIter = packages.find(packageName);
        if (pIter == packages.end())
            return;
        schemaClass = pIter->second->getClass(className, hash);
        if (schemaClass == 0)
            return;
    }

    Event event(broker, schemaClass, buffer);    

    if (listener)
        listener->event(event);
}

void SessionManager::handleSchemaResp(Broker* broker, Buffer& inBuffer, uint32_t sequence)
{
    uint8_t kind;
    string packageName;
    string className;
    uint8_t hash[16];

    kind = inBuffer.getOctet();
    inBuffer.getShortString(packageName);
    inBuffer.getShortString(className);
    inBuffer.getBin128(hash);

    {
        Mutex::ScopedLock l(lock);
        map<string, Package*>::iterator pIter = packages.find(packageName);
        if (pIter != packages.end() && !pIter->second->getClass(className, hash)) {
            ClassKey key(packageName, className, hash);
            SchemaClass* schemaClass(new SchemaClass(kind, key, inBuffer));
            pIter->second->addClass(className, hash, schemaClass);
            if (listener != 0) {
                listener->newClass(schemaClass->getClassKey());
            }
        }
    }

    sequenceManager.release(sequence);
    broker->decOutstanding();
}

void SessionManager::handleContentInd(Broker* broker, Buffer& buffer, uint32_t sequence, bool prop, bool stat)
{
    string packageName;
    string className;
    uint8_t hash[16];
    SchemaClass* schemaClass;

    buffer.getShortString(packageName);
    buffer.getShortString(className);
    buffer.getBin128(hash);

    {
        Mutex::ScopedLock l(lock);
        map<string, Package*>::iterator pIter = packages.find(packageName);
        if (pIter == packages.end())
            return;
        schemaClass = pIter->second->getClass(className, hash);
        if (schemaClass == 0)
            return;
    }

    Object object(broker, schemaClass, buffer, prop, stat);

    if (prop && className == "agent" && packageName == "org.apache.qpid.broker")
        broker->updateAgent(object);

    {
        Mutex::ScopedLock l(lock);
        if (syncSequenceList.count(sequence) == 1) {
            if (!object.isDeleted())
                getResult.push_back(object);
            return;
        }
    }

    if (listener) {
        if (prop)
            listener->objectProps(*broker, object);
        if (stat)
            listener->objectStats(*broker, object);
    }
}

void SessionManager::handleBrokerConnect(Broker* broker)
{
    if (listener != 0)
        listener->brokerConnected(*broker);
}

void SessionManager::handleBrokerDisconnect(Broker* broker)
{
    if (listener != 0)
        listener->brokerDisconnected(*broker);
}