#include "protocol/DefaultClientController.h"
#include "protocol/AggreGateCommandParser.h"
#include "protocol/AggreGateCommandUtils.h"
#include "communication/Command.h"
#include "communication/SocketException.h"
#include "context/AbstractContext.h"
#include "context/VariableDefinition.h"
#include "context/ContextSecurityException.h"
#include "datatable/encoding/TransferEncodingHelper.h"
#include "event/EventHandlingException.h"
#include "datatable/DataTable.h"
#include "datatable/DataRecord.h"
#include "datatable/encoding/ClassicEncodingSettings.h"
#include "server/ServerContextConstants.h"
#include "server/RootContextConstants.h"
#include "util/BlockingChannel.h"
#include "util/AggreReadableByteChannel.h"
#include "util/simpleobject/AgString.h"
#include "util/Log.h"
#include "util/SyntaxErrorException.h"
#include "util/MessageFormat.h"
#include "IOException.h"
#include "RejectedExecutionException.h"
#include "Cres.h"
#include "context/FunctionDefinition.h"
#include <boost/pointer_cast.hpp>
#include <boost/make_shared.hpp>
#include <boost/thread/locks.hpp>

/*******************************************************************************
 *      ForwardingEventListener
 * ****************************************************************************/

ForwardingEventListener::ForwardingEventListener(int listenerCode, ExpressionPtr filter,
                                                 DefaultClientController* defaultClientController)
    : DefaultContextEventListener(CallerControllerPtr(), listenerCode, filter)
{
    this->owningClientController = defaultClientController;
    setAcceptEventsWithoutListenerCode(true);
}


bool ForwardingEventListener::shouldHandle(EventPtr event)// throws EventHandlingException
{
    try {
        if (!DefaultContextEventListener::shouldHandle(event)) {
            return false;
        }

        if (!owningClientController->controllerShouldHandle(event, this)) {
            return false;
        }

        if (event->getName()==AbstractContextStrings::E_CHILD_ADDED)
        {
            Context* con = owningClientController->getContext(event->getContext());
            if (con == NULL || con->getChild(event->getData()->rec()->getString(AbstractContextStrings::EF_CHILD_ADDED_CHILD),
                                             getCallerController()) == NULL) {
                return false;
            }
        }

        if (event->getName() == ServerContextConstants::E_VISIBLE_CHILD_ADDED) {
            if (owningClientController->getContextManager()->get(
                        event->getData()->rec()->getString(ServerContextConstants::EF_VISIBLE_CHILD_ADDED_PATH),
                        getCallerController()) == NULL) {
                return false;
            }
        }

        return true;
    }catch (AggreGateException ex){
        throw EventHandlingException(ex.getMessage(), ex.getDetails());
    }
}


void ForwardingEventListener::handle(EventPtr event)// throws EventHandlingException
{
    if(Log::CLIENTS.isDebugEnabled())
    {
        LOG_CLIENTS_DEBUG("Handling event '" + event->getName().toUtf8()+ "' from context '"
                          + event->getContext().toUtf8() + "' for listener '" + std::to_string(getListenerCode()) + "' (" + toString().toUtf8() + ")");
    }

    try
    {
        OutgoingAggreGateCommandPtr cmd = owningClientController->constructEventCommand(event, getListenerCode());

        bool wasEmpty;
        {
            boost::lock_guard<boost::mutex> lock(owningClientController->pendingCommandsQueueMutex);
            wasEmpty = owningClientController->pendingCommandsQueue.empty();
            owningClientController->pendingCommandsQueue.push(cmd);
        }

        if(Log::CLIENTS.isDebugEnabled())
        {
            LOG_CLIENTS_DEBUG("Event queue overflow for client controller: " + this->toString().toUtf8());
        }

        if (wasEmpty)
        {
            try
            {
                PendingEventProcessingTask *pt = new PendingEventProcessingTask(owningClientController);
                owningClientController->commandExecutionService->submit(pt);
                delete pt;
            }
            catch (RejectedExecutionException ex)
            {
                LOG_CLIENTS_WARN("Cannot schedule new event delivery since command processing pool is overloaded");
            }
        }
    }
    catch (AggreGateException ex)
    {
        throw EventHandlingException(MessageFormat::format(Cres::get()->getString("clErrorHandlingEvent"), event->getName() + ex.getMessage(), ex.getDetails()));
    }
}

AgString ForwardingEventListener::toString()
{
    return "ForwardingEventListener: " + AgString::fromInt(getListenerCode());
}

int ForwardingEventListener::hashCode()
{
    //    const int prime = 31;
    int result = 1;
    return result;
}

bool ForwardingEventListener::equals(AgObject* obj)
{
    if (this == obj) {
        return true;
    }
    if (obj == NULL) {
        return false;
    }


    ForwardingEventListener* other = dynamic_cast<ForwardingEventListener*>(obj);
    if (getOwningController() != other->getOwningController()) {
        return false;
    }

    if (getListenerCode() != other->getListenerCode()) {
        return false;
    }

    if (getFilter() == NULL) {
        if (other->getFilter() != NULL) {
            return false;
        }
    }else if (!getFilter()->equals(other->getFilter().get())) {
        return false;
    }

    return true;
}

DefaultClientController* ForwardingEventListener::getOwningController()
{
    return owningClientController;
}

CallerControllerPtr ForwardingEventListener::getCallerController()
{
    return owningClientController->getCallerController();
}


/*******************************************************************************
 *      ProcessCommandTask
 ******************************************************************************/

ProcessCommandTask::ProcessCommandTask(IncomingAggreGateCommandPtr cmd, DefaultClientController* owningClientController)
{
    this->cmd = cmd;
    this->owningClientController = owningClientController;
}

int ProcessCommandTask::call()// throws Exception
{
    try
    {
        DefaultClientController* ownerCtrl = owningClientController;
        if (!ownerCtrl) {
            throw AggreGateException("owningClientController is null");
        }

        if (!ownerCtrl->shutDown) {
            OutgoingAggreGateCommandPtr reply = ownerCtrl->processCommand(cmd);

            ownerCtrl->processPendingEvents(); // If some events were generated by the processed commands, deliver them ASAP

            ownerCtrl->sendCommand(reply);
        }
    }catch (DisconnectionException ex) {
        LOG_CLIENTS_DEBUG("Disconnection exception while processing command: " + ex.getMessage().toUtf8());
    }catch (SocketException ex) {
        LOG_CLIENTS_INFO("Socket exception while processing command: " + ex.getMessage().toUtf8());
    }catch (AggreGateException/*Exception*/ ex) {
        LOG_CLIENTS_WARN("Error processing command: " + ex.getMessage().toUtf8());
    }

    return 0;
}


/*******************************************************************************
 *              PendingEventProcessingTask
 ******************************************************************************/

PendingEventProcessingTask::PendingEventProcessingTask(DefaultClientController* owningClientController)
{
    this->owningClientController = owningClientController;
}

void PendingEventProcessingTask::run()
{
    owningClientController->processPendingEvents();
}


/*******************************************************************************
 *      DefaultClientController
 * ****************************************************************************/

DefaultClientController::DefaultClientController(
        BlockingChannelPtr dataChannel,
        ContextManager* contextManager,
        /*ExecutorService* */AgObjectPtr /*commandExecutionService*/,
        int maxEventQueueLength)
    : AbstractClientController(contextManager)
{ 
    this->dataChannel = dataChannel;
    this->startMessageReceived = false;
    this->shutDown = false;
    defaultEventListener = ContextEventListenerPtr(new ForwardingEventListener(0, ExpressionPtr(), this));
    this->commandExecutionService = boost::shared_ptr<ThreadPoolCachedEx>(new ThreadPoolCachedEx(maxEventQueueLength, 100, 60/*second*/));
    commandParser = AggreGateCommandParserPtr(new AggreGateCommandParser(
                                                  boost::static_pointer_cast<AggreReadableByteChannel>(dataChannel),
                                                  AggreGateCommand::START_CHAR, AggreGateCommand::END_CHAR) );

}

//Not used
ActionManagerPtr DefaultClientController::getActionManager()
{
    assert(0);
    return ActionManagerPtr();
}

void DefaultClientController::processOperationGetVar(const AgString & id, Context* con,
                                                     const AgString & name, OutgoingAggreGateCommandPtr ans)
{
    DataTablePtr result;

    if(Log::CLIENTS.isDebugEnabled())
    {
        LOG_CLIENTS_DEBUG("Getting variable '" + name.toUtf8() + "' from context '" + con->getPath().toUtf8() + "'");
    }

    VariableDefinitionPtr vd = getVariableDefinition(con, name);

    if (vd == NULL) {
        ans->constructReply(id, AggreGateCommand::REPLY_CODE_DENIED(), Cres::get()->getString("conVarNotAvail") + name);
        return;
    }

    result = getVariable(con, name);

    ans->constructReply(id, AggreGateCommand::REPLY_CODE_OK());

    ans->addParam(result->encode(createClassicEncodingSettings(vd->getFormat().get() != NULL)));
}

DataTablePtr DefaultClientController::getVariable(Context* con, const AgString & name)
{
    return con->getVariable(name, getCallerController());
}

void DefaultClientController::processOperationSetVar(const AgString & id, Context* con, const AgString & name, const AgString & encodedValue, OutgoingAggreGateCommandPtr ans)
{
    if(Log::CLIENTS.isDebugEnabled())
    {
        LOG_CLIENTS_DEBUG("Setting variable '" + name.toUtf8() + "' of context '" + con->getPath().toUtf8() + "'");
    }

    VariableDefinitionPtr vd = getVariableDefinition(con, name);
    if (vd == NULL) {
        ans->constructReply(id, AggreGateCommand::REPLY_CODE_DENIED(), Cres::get()->getString("conVarNotAvail") + name);
        return;
    }

    ClassicEncodingSettingsPtr settings = ClassicEncodingSettingsPtr(new ClassicEncodingSettings(false, vd->getFormat()));

    DataTablePtr value = DataTablePtr(new DataTable(encodedValue, settings, true));

    setVariable(con, name, value);

    ans->constructReply(id, AggreGateCommand::REPLY_CODE_OK());
}

VariableDefinitionPtr DefaultClientController::getVariableDefinition(Context* con, const AgString & name)
{
    return con->getVariableDefinition(name);
}

void DefaultClientController::setVariable(Context* con, const AgString & name, DataTablePtr value)
{
    con->setVariable(name, static_cast< CallerControllerPtr >(getCallerController()), value);
}

void DefaultClientController::processOperationCallFunction(const AgString & id, Context* con, const AgString & name, const AgString & encodedParameters, OutgoingAggreGateCommandPtr ans)
{
    DataTablePtr result = DataTablePtr();

    if(Log::CLIENTS.isDebugEnabled())
    {
        LOG_CLIENTS_DEBUG("Calling function '" + name.toUtf8() + "' of context '" + con->getPath().toUtf8() + "'");
    }

    FunctionDefinitionPtr fd = getFunctionDefinition(con, name);

    if (fd == NULL) {
        ans->constructReply(id, AggreGateCommand::REPLY_CODE_DENIED(), Cres::get()->getString("conFuncNotAvail") + name);
        return;
    }

    ClassicEncodingSettingsPtr settings = ClassicEncodingSettingsPtr(new ClassicEncodingSettings(false, fd->getInputFormat()));

    DataTablePtr parameters = DataTablePtr(new DataTable(encodedParameters, settings, true));

    result = callFunction(con, name, parameters);

    ans->constructReply(id, AggreGateCommand::REPLY_CODE_OK());
    ans->addParam(result->encode(createClassicEncodingSettings(fd->getOutputFormat().get() != NULL)));
}

FunctionDefinitionPtr DefaultClientController::getFunctionDefinition(Context* con, const AgString & name)
{
    return con->getFunctionDefinition(name);
}

DataTablePtr DefaultClientController::callFunction(Context* con, const AgString & name, DataTablePtr parameters)
{
    return con->callFunction(name, static_cast< CallerControllerPtr >(getCallerController()), parameters);
}

bool DefaultClientController::addNormalListener(const AgString &context,const AgString &name, ContextEventListenerPtr cel)
{
    // Distributed: ok, because remote events will be redirected to this server
    Context* con = getContext(context);
    if (con != NULL)
    {
        return con->addEventListener(name, cel, true);
    }else {
        return false;
    }
}

void DefaultClientController::processOperationAddEventListener(const AgString & id, const AgString & context,
                                                               const AgString & name, int  listener, const AgString & filter,
                                                               OutgoingAggreGateCommandPtr ans)
{
    if(Log::CLIENTS.isDebugEnabled())
    {
        LOG_CLIENTS_DEBUG("Adding listener for event '" + name.toUtf8() + "' of context '" + context.toUtf8() + "'");
    }

    ContextEventListenerPtr cel = createListener(listener, !filter.empty() ? ExpressionPtr(new Expression(filter)) : ExpressionPtr());
    addMaskListener(context, name, cel, true);
    ans->constructReply(id, AggreGateCommand::REPLY_CODE_OK());
}

void DefaultClientController::processOperationRemoveEventListener(const AgString & id, const AgString & context, const AgString & name, int  listenerHashCode, const AgString & filter, OutgoingAggreGateCommandPtr ans)
{
    if(Log::CLIENTS.isDebugEnabled())
    {
        LOG_CLIENTS_DEBUG("Removing listener for event '" + name.toUtf8() + "' of context '" + context.toUtf8() + "'");
    }

    ContextEventListenerPtr cel = createListener(listenerHashCode, !filter.empty() ? ExpressionPtr(new Expression(filter)) : ExpressionPtr());
    removeMaskListener(context, name, cel);
    ans->constructReply(id, AggreGateCommand::REPLY_CODE_OK());
}

void DefaultClientController::processMessageStart(IncomingAggreGateCommandPtr cmd, OutgoingAggreGateCommandPtr ans) 
{
    int version = AgString(cmd->getParameter(AggreGateCommand::INDEX_PROTOCOL_VERSION)).toInt();


    LOG_CLIENTS_DEBUG("Processing start command, client protocol version: " + std::to_string(version));

    if (version == AggreGateCommandUtils::CLIENT_PROTOCOL_VERSION) {
        ans->constructReply(cmd->getId(), AggreGateCommand::REPLY_CODE_OK());
        ans->addParam(AgString::fromInt(AggreGateCommandUtils::CLIENT_PROTOCOL_VERSION));
        startMessageReceived = true;
    }else {
        ans->constructReply(cmd->getId(), AggreGateCommand::REPLY_CODE_DENIED());
        LOG_CLIENTS_DEBUG("Rejecting client connection with unsupported version '");
    }
}

void DefaultClientController::processMessageOperation(IncomingAggreGateCommandPtr cmd, OutgoingAggreGateCommandPtr ans) 
{
    AgString operation = cmd->getParameter(AggreGateCommand::INDEX_OPERATION_CODE);
    AgString context = cmd->getParameter(AggreGateCommand::INDEX_OPERATION_CONTEXT);
    AgString target = cmd->getParameter(AggreGateCommand::INDEX_OPERATION_TARGET);

    AgString originalThreadName = "";
    if (operation.length() > 1)
    {
        throw SyntaxErrorException(Cres::get()->getString("clInvalidOpcode") + operation);
    }

    if(Log::CLIENTS.isDebugEnabled())
    {
        LOG_CLIENTS_DEBUG("Processing message, context '" + context.toUtf8() + "', target '" + target.toUtf8() + "', operation '" + operation.toUtf8() + "'");
    }

    switch (operation.at(0))
    {
    case AggreGateCommand::COMMAND_OPERATION_ADD_EVENT_LISTENER: {
        AgString listenerStr = cmd->getParameter(AggreGateCommand::INDEX_OPERATION_LISTENER_CODE);
        int listener = listenerStr.length() > 0 ? listenerStr.toInt() : 0;
        AgString filter = cmd->hasParameter(AggreGateCommand::INDEX_OPERATION_FILTER)
                ? TransferEncodingHelper::decode(cmd->getParameter(AggreGateCommand::INDEX_OPERATION_FILTER)) : "";
        processOperationAddEventListener(cmd->getId(), context, target, listener, filter, ans);
        return;
    }

    case AggreGateCommand::COMMAND_OPERATION_REMOVE_EVENT_LISTENER: {
        AgString listenerStr = cmd->getParameter(AggreGateCommand::INDEX_OPERATION_LISTENER_CODE);
        int listener = listenerStr.length() > 0 ? listenerStr.toInt() : 0;
        AgString filter = cmd->hasParameter(AggreGateCommand::INDEX_OPERATION_FILTER)
                ? TransferEncodingHelper::decode(cmd->getParameter(AggreGateCommand::INDEX_OPERATION_FILTER)) : "";
        processOperationRemoveEventListener(cmd->getId(), context, target, listener, filter, ans);
        return;
    }
    }

    Context* con = getContext(context);
    if (con == NULL) {
        throw new ContextException(Cres::get()->getString("conNotAvail") + context);
    }

    if (addNormalListener(con->getPath(), AbstractContextStrings::E_DESTROYED, defaultEventListener))
    {
        addNormalListener(con->getPath(), AbstractContextStrings::E_CHILD_ADDED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_CHILD_REMOVED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_VARIABLE_ADDED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_VARIABLE_REMOVED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_FUNCTION_ADDED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_FUNCTION_REMOVED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_EVENT_ADDED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_EVENT_REMOVED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_INFO_CHANGED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_ACTION_ADDED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_ACTION_REMOVED, defaultEventListener);
        addNormalListener(con->getPath(), AbstractContextStrings::E_ACTION_STATE_CHANGED, defaultEventListener);

        addCustomListeners(con);
    }

    switch (operation.at(0))
    {
    case AggreGateCommand::COMMAND_OPERATION_GET_VAR:
        processOperationGetVar(cmd->getId(), con, target, ans);
        break;

    case AggreGateCommand::COMMAND_OPERATION_SET_VAR:
        processOperationSetVar(cmd->getId(), con, target, cmd->getEncodedDataTableFromOperationMessage(), ans);
        break;

    case AggreGateCommand::COMMAND_OPERATION_CALL_FUNCTION:
        processOperationCallFunction(cmd->getId(), con, target, cmd->getEncodedDataTableFromOperationMessage(), ans);
        break;

    default:
        throw SyntaxErrorException(Cres::get()->getString("clInvalidOpcode") + AgString(operation.at(0)));
    }
}

void DefaultClientController::addCustomListeners(Context* /*con*/)
{
}

void DefaultClientController::processMessage(IncomingAggreGateCommandPtr cmd, OutgoingAggreGateCommandPtr ans) 
{
    AgString messageCode = cmd->getMessageCode();

    if (messageCode.length() > 1) {
        throw SyntaxErrorException(Cres::get()->getString("clInvalidMsgCode") + messageCode);
    }

    char code = (char)messageCode.at(0);

    if ((code != AggreGateCommand::MESSAGE_CODE_START) && (!startMessageReceived))
    {
        LOG_CLIENTS_DEBUG("Can't process message: start message was not received yet");
        ans->constructReply(cmd->getId(), AggreGateCommand::REPLY_CODE_DENIED());
        return;
    }

    switch (code)
    {
    case AggreGateCommand::MESSAGE_CODE_START:
        processMessageStart(cmd, ans);
        break;

    case AggreGateCommand::MESSAGE_CODE_OPERATION:
        processMessageOperation(cmd, ans);
        break;

    default:
        throw SyntaxErrorException(Cres::get()->getString("clInvalidMsgCode") + AgString(messageCode.at(0)));
    }
}

OutgoingAggreGateCommandPtr DefaultClientController::processCommand(IncomingAggreGateCommandPtr cmd) 
{
    OutgoingAggreGateCommandPtr ans = OutgoingAggreGateCommandPtr(new OutgoingAggreGateCommand());

    try
    {
        AgString commandCode = cmd->getParameter(AggreGateCommand::INDEX_COMMAND_CODE);

        if (commandCode.length() > 1) {
            throw new AggreGateException(Cres::get()->getString("clInvalidCmdCode") + commandCode);
        }

        switch (commandCode.at(0))
        {
        case AggreGateCommand::COMMAND_CODE_MESSAGE:
            processMessage(cmd, ans);
            break;

        default:
            throw AggreGateException(Cres::get()->getString("clInvalidCmdCode") + AgString(commandCode[0]));
        }
    }catch (ContextSecurityException ex) {
        if(Log::CLIENTS.isDebugEnabled())
        {
            LOG_CLIENTS_DEBUG("Access denied while processing command '" + cmd->toString().toUtf8() + "': " + ex.getMessage().toUtf8());
        }
        ans->constructReply(cmd->getId(), AggreGateCommand::REPLY_CODE_DENIED(), ex.getMessage());
    }catch (AggreGateException/*Throwable*/ ex) {
        if(Log::CLIENTS.isDebugEnabled())
        {
            LOG_CLIENTS_DEBUG("Error processing command '" + cmd->toString().toUtf8() + "': " + ex.getMessage().toUtf8());
        }
        ans->constructReply(cmd->getId(), AggreGateCommand::REPLY_CODE_ERROR(), ex.getMessage(), getErrorDetails(ex));
    }

    return ans;
}

void DefaultClientController::shutdown()
{
    try
    {
        LOG_CLIENTS_DEBUG("Shutting down client controller");

        shutDown = true;

        if (dataChannel != NULL) {
            dataChannel->close();
        }

        if (getContextManager()->getRoot()->getFunctionDefinition(RootContextConstants::F_LOGOUT) != NULL) {
            getContextManager()->getRoot()->callFunction(RootContextConstants::F_LOGOUT, getCallerController());
        }

        AbstractClientController::shutdown();
    }catch (AggreGateException ex){
        LOG_CLIENTS_WARN("Error shutting down client controller: "+ ex.getMessage().toUtf8());
    }
}

bool DefaultClientController::run()
{
    try {
        runImpl();
        return true;
    }catch (IOException ex) {
        AgString msg = "I/O error while communicating with client";
        LOG_CLIENTS_INFO(msg.toUtf8() + ": " + ex.getMessage().toUtf8());
        return false;
    }catch (DisconnectionException ex) {
        LOG_CLIENTS_INFO("Client disconnected: " + ex.getMessage().toUtf8());
        return false;
    }catch (AggreGateException/*Exception*/ ex) {
        LOG_CLIENTS_WARN("Error in client controller " + ex.getMessage().toUtf8());
        return false;
    }
}

void DefaultClientController::runImpl() 
{
    try
    {
        IncomingAggreGateCommandPtr command = boost::static_pointer_cast<IncomingAggreGateCommand>(commandParser->readCommand());

        if (command != NULL) {
            if(Log::CLIENTS.isDebugEnabled())
            {
                LOG_CLIENTS_DEBUG("Received: " + command->toString().toUtf8());
            }

            try {
                ProcessCommandTask *pct = new ProcessCommandTask(command, this);
                commandExecutionService->submit(pct);
                delete pct;
            }catch (RejectedExecutionException ex) {
                OutgoingAggreGateCommandPtr reply = OutgoingAggreGateCommandPtr(new OutgoingAggreGateCommand());
                reply->constructReply(command->getId(), AggreGateCommand::REPLY_CODE_ERROR(), Cres::get()->getString("devServerOverloaded"));
                sendCommand(reply);
            }
        }
    }
    catch (DisconnectionException ex)
    {
        throw DisconnectionException(ex);
    }
}

AgString DefaultClientController::getErrorDetails(AggreGateException/*Throwable*/ error)
{
    UNUSED(error);
    //TODO:
    return "";
}

void DefaultClientController::sendCommand(OutgoingAggreGateCommandPtr cmd) 
{
    cmd->send(dataChannel);

    if(Log::CLIENTS.isDebugEnabled())
    {
        LOG_CLIENTS_DEBUG("Sent: " + cmd->toString().toUtf8());
    }
}

ContextEventListenerPtr DefaultClientController::createListener(int  listenerHashCode, ExpressionPtr filter)
{
    if (listenerHashCode == 0) {
        return defaultEventListener;
    }

    return ContextEventListenerPtr(new ForwardingEventListener(listenerHashCode, filter, this));
}

// run in thread
void DefaultClientController::processPendingEvents()
{
    while (true)
    {
        OutgoingAggreGateCommandPtr current;

        {
            boost::lock_guard<boost::mutex> lock(pendingCommandsQueueMutex);

            if (pendingCommandsQueue.empty())
            {
                return;
            }
            current = pendingCommandsQueue.front();
            pendingCommandsQueue.pop();
        }

        try {
            sendCommand(current);
        }catch (DisconnectionException ex) {
            LOG_CLIENTS_DEBUG("Disconnection detected while forwarding event to client: " + ex.getMessage().toUtf8());
        }catch (AggreGateException/*Throwable*/ ex) {
            AgString msg = "Exception while forwarding event to client: ";
            LOG_CLIENTS_INFO(msg.toUtf8() + ex.getMessage().toUtf8());
        }
    }
}

Context* DefaultClientController::getContext(const AgString &path)
{
    return getContextManager()->get(path, getCallerController());
}

ContextEventListenerPtr DefaultClientController::getDefaultEventListener()
{
    return defaultEventListener;
}

ClassicEncodingSettingsPtr DefaultClientController::createClassicEncodingSettings(bool useFormatCache)
{
    UNUSED(useFormatCache);
    return ClassicEncodingSettingsPtr(new ClassicEncodingSettings(false));
}

OutgoingAggreGateCommandPtr DefaultClientController::constructEventCommand(EventPtr event, int  listenerCode)
{
    OutgoingAggreGateCommandPtr cmd = OutgoingAggreGateCommandPtr(new OutgoingAggreGateCommand());
    cmd->constructEvent(event->getContext(), event->getName(),
                        event->getLevel(), event->getData()->encode(createClassicEncodingSettings(true)),
                        event->getId(), event->getCreationtime(), listenerCode);
    return cmd;
}

bool DefaultClientController::isConnected()
{
    return dataChannel->isOpen();
}

//Not usage
AgString DefaultClientController::getAddress()
{
    assert(0);
    return "";
}
