#include "communication/AsyncCommandProcessor.h"
#include "communication/SocketException.h"
#include "util/TimeHelper.h"
#include "util/Log.h"
#include "util/MessageFormat.h"
#include "Cres.h"
#include "InterruptedException.h"
#include "IOException.h"
#include "ExecutionException.h"
#include <boost/thread/lock_guard.hpp>

AsyncCommandProcessor::AsyncCommandProcessor(AbstractDeviceController* controller)
    : AgThread()
{
    this->controller = controller;
    this->statistics = CommandProcessorStatisticsPtr(new CommandProcessorStatistics());
}

CommandPtr AsyncCommandProcessor::sendSyncCommand(CommandPtr cmd)
{
    ReplyMonitorPtr mon = sendCommand(cmd);
    CommandPtr reply = waitReplyMonitor(mon);
    statistics->update(mon);

    return reply;
}

void AsyncCommandProcessor::sendUnrepliedCommand(CommandPtr cmd)
{
    sendCommandImplementation(cmd);
}

void AsyncCommandProcessor::resetSentCommandTimeouts()
{
    boost::lock_guard<boost::mutex> lock(sentCommandsQueueMutex_);
    for (std::list<ReplyMonitorPtr>::iterator iter=sentCommandsQueue.begin(); iter!=sentCommandsQueue.end(); ++iter) {
        (*iter)->reset();
    }
}

void AsyncCommandProcessor::sendCommandImplementation(CommandPtr cmd)
{
    try {
        controller->send(cmd);
    }catch (InterruptedException ex) {
        throw IOException(ex.getMessage(), ex.getDetails());
    }
}

ReplyMonitorPtr AsyncCommandProcessor::sendCommand(CommandPtr cmd)
{
    boost::lock_guard<boost::mutex> lock(mutex_);
    ReplyMonitorPtr mon = ReplyMonitorPtr(new ReplyMonitor(cmd));
    addSentCommand(mon);
    sendUnrepliedCommand(cmd);

    return mon;
}

CommandPtr AsyncCommandProcessor::waitReplyMonitor(ReplyMonitorPtr mon) //throws IOException, InterruptedException
{
    if (mon->getReply() == NULL) {
        int64_t timeout = mon->getCommand()->getTimeout() != 0 ? mon->getCommand()->getTimeout() : controller->getCommandTimeout();

        mon->waitReply(timeout);
    }

    {
        boost::lock_guard<boost::mutex> lock(sentCommandsQueueMutex_);
        sentCommandsQueue.remove(mon);
    }

    if (mon->getReply() != NULL){
        return mon->getReply();
    }else {
        throw IOException(MessageFormat::format(Cres::get()->getString("cmdTimeout"), mon->getCommand()->toString()));
    }
}

void AsyncCommandProcessor::addSentCommand(ReplyMonitorPtr mon)
{
    boost::lock_guard<boost::mutex> lock(sentCommandsQueueMutex_);
    sentCommandsQueue.push_back(mon);
}

void  AsyncCommandProcessor::run()
{
    CommandPtr cmd;

    try {
        while (bRunning) {
            try {
                cmd = controller->getCommandParser()->readCommand();

                if (cmd != NULL) {
                    if(Log::AG.isDebugEnabled())
                    {
                        LOG_AG_DEBUG("Received command: " +  cmd->toString().toUtf8());
                    }
                    if (cmd->isAsync()) {
                        controller->processAsyncCommand(cmd);

                       // AgPerformanceCounter::get("AsyncProcessor", 10)->count(1);

                    } else {
                        bool found = false;
                        boost::posix_time::ptime thisTime = boost::posix_time::microsec_clock::local_time();

                        {
                            boost::lock_guard<boost::mutex> lock(sentCommandsQueueMutex_);
                            std::list<ReplyMonitorPtr>::iterator iter=sentCommandsQueue.begin();
                            while (iter!=sentCommandsQueue.end()) {

                                ReplyMonitorPtr cur = *iter;

                                if (cur->getCommand()->getId() == cmd->getId()) {
                                    cur->setReply(cmd);
                                    iter = sentCommandsQueue.erase(iter);
                                    found = true;
                                    continue;
                                }else if ( (thisTime - cur->getTime()) > boost::posix_time::milliseconds(TimeHelper::DAY_IN_MS())) {
                                    if(Log::AG.isInfoEnabled())
                                    {
                                        LOG_AG_INFO("Removing expired reply monitor from queue: " + cur->toString().toUtf8());
                                    }
                                    iter = sentCommandsQueue.erase(iter);
                                    continue;
                                }

                                iter++;
                            }
                        }

                        if (!found) {
                            if(Log::AG.isDebugEnabled())
                            {
                                std:: string str;
                                for(std::list<ReplyMonitorPtr>::iterator iter = sentCommandsQueue.begin(); iter != sentCommandsQueue.end(); ++iter)
                                    str.append((*iter)->toString().toUtf8() + ", ");
                                LOG_AG_DEBUG("Reply cannot be matched to a sent command: "
                                             + cmd->toString().toUtf8() + ", commands in progress:" + str);
                            }
                        }
                    }
                }
            } catch(boost::thread_interrupted const&) {
                LOG_AG_ERROR("Async processor interrupted");
                break;
            } catch(...) {
                LOG_AG_ERROR("Async processor interrupted. Unknown error.");
                break;
            }
        }

        controller->disconnectImpl();

    } catch (DisconnectionException ex) {
        processError(log4cpp::Priority::DEBUG, "Disconnection of peer detected in async processor", ex);
        bRunning = false;
    } catch (SocketException ex) {
        processError(log4cpp::Priority::DEBUG, "Socket error in async processor", ex);
    } catch (AggreGateException ex) {
        processError(log4cpp::Priority::ERROR, "Error in async processor", ex);
    } catch(...) {
        LOG_AG_ERROR("Unknown error in async processor");
    }

    boost::lock_guard<boost::mutex> lock(sentCommandsQueueMutex_);
    {
        if (sentCommandsQueue.size() > 0)
        {
            for (std::list<ReplyMonitorPtr>::iterator rm = sentCommandsQueue.begin(); rm != sentCommandsQueue.end(); ++rm) {
                (*rm)->terminate();
            }
            sentCommandsQueue.clear();
        }
    }
}


void AsyncCommandProcessor::processError(int priority, const AgString & message, const AggreGateException& ex)
{
    UNUSED(priority);
    UNUSED(message);
    UNUSED(ex);

    switch(priority)
    {
    case log4cpp::Priority::DEBUG :
        LOG_AG_DEBUG(message.toUtf8() + ": " + ex.getMessage().toUtf8());
    case log4cpp::Priority::ERROR :
        LOG_AG_ERROR(message.toUtf8() + ": " + ex.getMessage().toUtf8());
    case log4cpp::Priority::INFO :
        LOG_AG_INFO(message.toUtf8() + ": " + ex.getMessage().toUtf8());
    case log4cpp::Priority::WARN :
        LOG_AG_WARN(message.toUtf8() + ": " + ex.getMessage().toUtf8());
    default :;
    }

    try
    {
        controller->disconnectImpl();
    }catch (AggreGateException e) {
        LOG_AG_DEBUG("Disconnection error: " + e.getMessage().toUtf8());
    }
}

bool AsyncCommandProcessor::isActive()
{
    boost::lock_guard<boost::mutex> lock(sentCommandsQueueMutex_);
    return !sentCommandsQueue.empty();
}

void AsyncCommandProcessor::start()
{
    bRunning = true;
    AgThread::start();
}

void AsyncCommandProcessor::finishThread()
{
    bRunning = false;
    controller->getCommandParser()->close();
    waitForFinished();
}

std::list<ReplyMonitorPtr> AsyncCommandProcessor::getActiveCommands()
{
    std::list<ReplyMonitorPtr> res;

    boost::lock_guard<boost::mutex> lock(sentCommandsQueueMutex_);
    std::copy(sentCommandsQueue.begin(), sentCommandsQueue.end(), std::back_inserter(res));

    return res;
}

AbstractDeviceController* AsyncCommandProcessor::getController()
{
    return controller;
}

CommandProcessorStatisticsPtr AsyncCommandProcessor::getStatistics()
{
    return statistics;
}

