#include <functional>
#include <boost/pointer_cast.hpp>

#include "protocol/AbstractAggreGateDeviceController.h"
#include "context/ContextException.h"
#include "context/ContextManager.h"
#include "context/ContextSecurityException.h"
#include "protocol/AggreGateCommandUtils.h"
#include "protocol/AggreGateDevice.h"
#include "protocol/IncomingAggreGateCommand.h"
#include "protocol/RemoteContextManager.h"
#include "protocol/ProxyContext.h"
#include "datatable/DataTable.h"
#include "datatable/DataTableException.h"
#include "datatable/encoding/ClassicEncodingSettings.h"
#include "device/RemoteDeviceErrorException.h"
#include "util/simpleobject/AgString.h"
#include "util/simpleobject/AgDate.h"
#include "util/Log.h"
#include "AggreGateException.h"
#include "util/Util.h"

#define EVENT_PROCESSOR_THREADS 4 /* 4, because complexity of target/source thread is ~ 1/4. Do not depends on hardware core counts */

AbstractAggreGateDeviceController::AbstractAggreGateDeviceController(AggreGateDevice* dev)
    : AbstractDeviceController(dev->getCommandTimeout()),
      device(dev), contextManager(NULL)
{
    this->userSettings = UserSettingsPtr(new UserSettings());
    this->formatCache = FormatCachePtr(new FormatCache(this));
    this->avoidSendingFormats = false;
    this->eventPreprocessor = AgThreadPoolPtr(new AgThreadPool("Event Processor", EVENT_PROCESSOR_THREADS, 500000));
    // -------- ThreadPool Debug --------
    this->eventPreprocessor->setDebugInfoTimeout(0);

    LOG_CORE_DEBUG("Starting AbstractAggreGateDeviceController, thread id: " << std::this_thread::get_id());
}

AbstractAggreGateDeviceController::~AbstractAggreGateDeviceController()
{
    if (eventPreprocessor != NULL)
        eventPreprocessor->terminate();

    SAFE_DELETE(this->contextManager);
}

ContextManager* AbstractAggreGateDeviceController::getContextManager()
{
    return contextManager;
}

void AbstractAggreGateDeviceController::setContextManager(ContextManager* contextManager)
{
    SAFE_DELETE(this->contextManager);
    this->contextManager = contextManager;
}

void AbstractAggreGateDeviceController::setDevice(AggreGateDevice* device)
{
    this->device = device;
}

AggreGateDevice* AbstractAggreGateDeviceController::getDevice()
{
    return device;
}

CallerControllerPtr AbstractAggreGateDeviceController::getCallerController()
{
    return callerController;
}

void AbstractAggreGateDeviceController::setCallerController(CallerControllerPtr callerController)
{
    this->callerController = callerController;
}

FormatCachePtr AbstractAggreGateDeviceController::getFormatCache()
{
    return formatCache;
}

UserSettingsPtr AbstractAggreGateDeviceController::getSettings()
{
    return userSettings;
}

ClassicEncodingSettingsPtr AbstractAggreGateDeviceController::createClassicEncodingSettings(bool forSending)
{
    ClassicEncodingSettingsPtr es = ClassicEncodingSettingsPtr(new ClassicEncodingSettings(false));
    if (!forSending) {
        es->setFormatCache(formatCache);
    }
    es->setEncodeFormat(!avoidSendingFormats);

    return es;
}


bool AbstractAggreGateDeviceController::isAvoidSendingFormats()
{
    return avoidSendingFormats;
}

void AbstractAggreGateDeviceController::setAvoidSendingFormats(bool avoidSendingFormats)
{
    this->avoidSendingFormats = avoidSendingFormats;
}

bool AbstractAggreGateDeviceController::connectImpl()
{
    OutgoingAggreGateCommandPtr outCmd = AggreGateCommandUtils::startMessage();
    IncomingAggreGateCommandPtr ans = boost::static_pointer_cast<IncomingAggreGateCommand>(
                sendCommand(outCmd) );

    if (ans->getReplyCode() != AggreGateCommand::REPLY_CODE_OK()) {
        throw RemoteDeviceErrorException(Cres::get()->getString("devUncompatibleVersion"));
    }

    formatCache->clear();

    return true;
}

void AbstractAggreGateDeviceController::destroy()
{
}

void AbstractAggreGateDeviceController::disconnectImpl()
{
    if (contextManager != NULL) {
        contextManager->stop();
    }

    if (eventPreprocessor != NULL) {
        eventPreprocessor->terminate();
    }

    formatCache->clear();
}

std::list<ProxyContext*> AbstractAggreGateDeviceController::getProxyContexts(AgString path)
{
    std::list<ProxyContext*> list;

    ProxyContext* con = dynamic_cast<ProxyContext*>(getContextManager()->get(path));
    if (con != NULL)
    {
        list.push_back(con);
    }


    return list;
}


IncomingAggreGateCommandPtr AbstractAggreGateDeviceController::sendCommandAndCheckReplyCode(OutgoingAggreGateCommandPtr cmd)
{
    IncomingAggreGateCommandPtr ans = boost::static_pointer_cast<IncomingAggreGateCommand>( sendCommand(cmd));

    if (ans->getReplyCode() == AggreGateCommand::REPLY_CODE_DENIED()) {
        AgString message = (ans->getNumberOfParameters() > AggreGateCommand::INDEX_REPLY_MESSAGE)
                ? ": " + DataTableUtils::transferDecode(ans->getParameter(AggreGateCommand::INDEX_REPLY_MESSAGE)) : "";
        throw ContextSecurityException(Cres::get()->getString("devAccessDeniedReply") + message);
    }

    if (ans->getReplyCode() == AggreGateCommand::REPLY_CODE_ERROR())
    {
        AgString message = (ans->getNumberOfParameters() > AggreGateCommand::INDEX_REPLY_MESSAGE)
                ? ": " + DataTableUtils::transferDecode(ans->getParameter(AggreGateCommand::INDEX_REPLY_MESSAGE)) : "";
        AgString details = ans->getNumberOfParameters() > AggreGateCommand::INDEX_REPLY_DETAILS
                ? DataTableUtils::transferDecode(ans->getParameter(AggreGateCommand::INDEX_REPLY_DETAILS)) : "";
        throw RemoteDeviceErrorException(Cres::get()->getString("devServerReturnedError") + message, details);
    }

    if (ans->getReplyCode() != AggreGateCommand::REPLY_CODE_OK())
    {
        throw RemoteDeviceErrorException(Cres::get()->getString("devServerReturnedError") + ": "
                                         + ans->toString() + " (error code: '" + ans->getReplyCode() + "')");
    }

    return ans;
}

void AbstractAggreGateDeviceController::processAsyncCommand(CommandPtr cmd)
{
    // AGDEBUG(DBG_DEBUG, "COMMANDS", "Async command received from server: " + cmd->toString());

    IncomingAggreGateCommandPtr incCmd = boost::static_pointer_cast<IncomingAggreGateCommand>(cmd);
    if (incCmd->getMessageCode().at(0) == AggreGateCommand::MESSAGE_CODE_EVENT){
        processEvent(incCmd);
    }
}

void AbstractAggreGateDeviceController::processEvent(IncomingAggreGateCommandPtr cmd)
{
    if (eventPreprocessor == NULL || eventPreprocessor->isTerminated()) return;

    /* Using async preprocessor is necessary, since event's data table may be based on a table format that may be received in one of the later commands */
    std::function<void()> task = std::bind(&AbstractAggreGateDeviceController::taskProcessEvent, this, cmd);
    eventPreprocessor->execute(task);
}

void AbstractAggreGateDeviceController::taskProcessEvent(IncomingAggreGateCommandPtr cmd)
{
    if (!isConnected()) {
        return;
    }

    try
    {
        AgString contextPath = cmd->getParameter(AggreGateCommand::INDEX_EVENT_CONTEXT);
        AgString eventName = cmd->getParameter(AggreGateCommand::INDEX_EVENT_NAME);

        int level = AgString(cmd->getParameter(AggreGateCommand::INDEX_EVENT_LEVEL)).toInt();

        AgString idstr = cmd->getParameter(AggreGateCommand::INDEX_EVENT_ID);
        int64_t id = idstr.length() > 0 ? AgString(idstr).toInt64() : 0;

        AgString listenerstr = cmd->getParameter(AggreGateCommand::INDEX_EVENT_LISTENER);
        int listener = listenerstr.length() > 0 ? AgString(listenerstr).toInt() : 0;

        std::list<ProxyContext*> contexts = getProxyContexts(contextPath);

        if (contexts.size() == 0) {
           LOG_CONTEXT_EVENTS_INFO("Error firing event '" + eventName.toUtf8() + "': context '" + contextPath.toUtf8()  + "' not found");

           return;
        }

        for (std::list<ProxyContext*>::iterator con = contexts.begin(); con != contexts.end(); ++con) {
            EventDefinitionPtr ed = (*con)->getEventDefinition(eventName);

            if (ed == NULL) {
                LOG_CONTEXT_EVENTS_WARN("Error firing event: event '" + eventName.toUtf8()  + "' not available in context '" + contextPath.toUtf8()  + "'");
                continue;
            }

            DataTablePtr data = decodeRemoteDataTable(ed->getFormat(), cmd->getEncodedDataTableFromEventMessage());

            AgDatePtr timestamp = AgDatePtr();
            if (cmd->hasParameter(AggreGateCommand::INDEX_EVENT_TIMESTAMP)) {
                AgString timestampstr = cmd->getParameter(AggreGateCommand::INDEX_EVENT_TIMESTAMP);
                timestamp = timestampstr.length() > 0 ? AgDatePtr(new AgDate(timestampstr.toInt64())) : AgDatePtr();
            }

            EventPtr event = (*con)->fireEvent(ed->getName(), data, level, id, timestamp, listener, CallerControllerPtr(),
                                            FireEventRequestControllerPtr(new FireEventRequestController(false)));

            confirmEvent(*con, ed, event);
        }
    } catch (AggreGateException ex)	{
        LOG_AG_ERROR("Error processing async command '" + cmd->toString().toUtf8() + "'\n\tex: " + ex.getMessage().toUtf8());
    } catch(...) {
        LOG_AG_ERROR("Unknown error processing async command '" + cmd->toString().toUtf8() + "'");
    }
}

void AbstractAggreGateDeviceController::confirmEvent(Context* con, EventDefinitionPtr def, EventPtr event)
{
    UNUSED(con);
    UNUSED(def);
    UNUSED(event);
}


void AbstractAggreGateDeviceController::setCommandParser(CommandParserPtr commandBuffer)
{
    AbstractDeviceController::setCommandParser(commandBuffer);
}


AgString AbstractAggreGateDeviceController::toString()
{
    return getDevice()->toString();
}

DataTablePtr AbstractAggreGateDeviceController::callRemoteFunction(const AgString& context,
    const AgString& name, TableFormatPtr outputFormat, DataTablePtr parameters)

{
    try
    {
        AgString encodedParameters = parameters->encode(createClassicEncodingSettings(true));
        IncomingAggreGateCommandPtr ans = sendCommandAndCheckReplyCode(AggreGateCommandUtils::callFunctionOperation(context, name, encodedParameters));
        return decodeRemoteDataTable(outputFormat, ans->getEncodedDataTableFromReply());
    }catch (ContextException ex) {
        throw ex;
    }catch (AggreGateException ex) {
        throw ContextException(ex.getMessage(), ex.getDetails());
    }
}

DataTablePtr AbstractAggreGateDeviceController::decodeRemoteDataTable(TableFormatPtr format, AgString encodedReply)
{
    if (isAvoidSendingFormats())
    {
        ClassicEncodingSettingsPtr settings = ClassicEncodingSettingsPtr(new ClassicEncodingSettings(false, format));
        settings->setFormatCache(formatCache);
        return DataTablePtr(new DataTable(encodedReply, settings, true));
    }

    try {
        return DataTablePtr(new DataTable(encodedReply, createClassicEncodingSettings(false), false));
    }catch (AggreGateException ex){
        throw DataTableException("Error parsing encoded data table '" + encodedReply + "': " + ex.getMessage(), ex.getDetails());
    }
}
