#include "datatable/encoding/FormatCache.h"
#include "IllegalArgumentException.h"
#include "IllegalStateException.h"
#include "InterruptedException.h"
#include "util/Util.h"
#include "util/simpleobject/AgString.h"
#include <boost/thread/locks.hpp>
#include "datatable/TableFormat.h"
#include "datatable/DataRecord.h"
#include "protocol/AbstractAggreGateDeviceController.h"
#include "protocol/RemoteContextManager.h"
#include "context/Contexts.h"
#include "server/UtilitiesContextConstants.h"
#include "server/CommonServerFormats.h"
#include "datatable/encoding/ClassicEncodingSettings.h"
#include "util/Log.h"

const int FormatCache::RETRIES;
const int FormatCache::TIMEOUT;

FormatCache::FormatCache(AbstractAggreGateDeviceController* controller)
{
    this->controller = controller;
}

void FormatCache::put(int id, TableFormatPtr format)
{
    boost::lock_guard<boost::mutex> lock(cacheLock);
    if (format.get() == NULL)
    {
        throw IllegalArgumentException("Format is NULL");
    }

    cache[id] = format;

    //  AGDEBUG(DBG_DEBUG, "PROTOCOL_CACHING", "Cached format as #" + AgString::fromInt(id) + ": " + format->toString());
}

boost::shared_ptr<int> FormatCache::add(TableFormatPtr format)
{
    assert(0);
    boost::lock_guard<boost::mutex> lock(cacheLock);

    if (format.get() == NULL)
    {
        throw IllegalArgumentException("Format is NULL");
    }

    int id = format->hashCode();

    TableFormatPtr previous;
    std::map<int, TableFormatPtr>::iterator it = cache.find(id);
    if (it != cache.end())
    {
        previous = it->second;
    }

    cache[id] = format;

    if (previous.get() != NULL && !Util::equals(format, previous))
    {
        LOG_PROTOCOL_CACHING_WARN("Found two non-equal formats with equal ID (" + AgString::fromInt(id).toUtf8() + "): " + format->toString().toUtf8() + " and " + previous->toString().toUtf8());
    }

    if (!format->isImmutable())
    {
        LOG_PROTOCOL_CACHING_WARN("Cached mutable format as #" + AgString::fromInt(id).toUtf8() + ": " + format->toString().toUtf8());
    }

    if(Log::PROTOCOL_CACHING.isDebugEnabled())
    {
        LOG_PROTOCOL_CACHING_DEBUG("Cached format as #" + AgString::fromInt(id).toUtf8() + ": " + format->toString().toUtf8());
    }

    return boost::shared_ptr<int>(new int(id));
}

boost::shared_ptr<int> FormatCache::getId(TableFormatPtr format)
{
    assert(0);
    boost::lock_guard<boost::mutex> lock(cacheLock);
    int id = format->hashCode();

    std::map<int, TableFormatPtr>::iterator it = cache.find(id);
    if (it != cache.end())
    {
        return boost::shared_ptr<int>(new int(id));
    }
    else
    {
        return boost::shared_ptr<int>(new int(0));
    }
}

TableFormatPtr FormatCache::getCachedVersion(TableFormatPtr format)
{
    UNUSED(format);
    //not used
    assert(0);
    return TableFormatPtr();
}

TableFormatPtr FormatCache::get(int id)
{
    TableFormatPtr result;
    int retry = 0;

    do
    {
        {
            boost::lock_guard<boost::mutex> lock(cacheLock);
            std::map<int, TableFormatPtr>::iterator it = cache.find(id);
            if (it != cache.end())
            {
                result = it->second;
            }
        }

        if (result.get() == NULL)
        {
            if (controller != NULL)
            {
                try
                {
                    if(Log::PROTOCOL_CACHING.isDebugEnabled())
                    {
                        LOG_PROTOCOL_CACHING_DEBUG("Requesting remote format #" + AgString::fromInt(id).toUtf8());
                    }

                    DataTablePtr output;

                    Context* rootContext = controller->getContextManager() != NULL ? controller->getContextManager()->get(Contexts::CTX_ROOT) : NULL;
                    Context* utilitiesContext = controller->getContextManager() != NULL ? controller->getContextManager()->get(Contexts::CTX_UTILITIES) : NULL;
                    std::list<AgObjectPtr> listParam;
                    listParam.push_back( AgObjectPtr(new AgInteger(id)) );
                    if (rootContext != NULL && rootContext->getFunctionDefinition(UtilitiesContextConstants::F_GET_FORMAT).get() != NULL)
                    {
                        output = rootContext->callFunction(UtilitiesContextConstants::F_GET_FORMAT, listParam);
                    }
                    else if(utilitiesContext != NULL && utilitiesContext->getFunctionDefinition(UtilitiesContextConstants::F_GET_FORMAT) != NULL){
                        // TODO: remove in v6
                        // backward compatibility:
                        output = utilitiesContext->callFunction(UtilitiesContextConstants::F_GET_FORMAT, listParam);
                    }
                    else
                    {
                        // TODO: change Contexts.CTX_UTILITIES to Contexts.CTX_ROOT in v6
                        // Also, getFormat function should be removed from UtilitiesContext in v6. Check it out, please.
                        DataRecordPtr dr = DataRecordPtr(new DataRecord(CommonServerFormats::getInstance().getFIFT_GET_FORMAT(), AgObjectPtr(new AgInteger(id))));
                        DataTablePtr input = dr->wrap();
                        output = controller->callRemoteFunction(Contexts::CTX_UTILITIES, UtilitiesContextConstants::F_GET_FORMAT, CommonServerFormats::getInstance().getFOFT_GET_FORMAT(), input);
                    }

                    AgString formatData = output->rec()->getString(UtilitiesContextConstants::FOF_GET_FORMAT_DATA);
                    result = TableFormatPtr(new TableFormat(formatData, ClassicEncodingSettingsPtr(new ClassicEncodingSettings(false))));

                    if(Log::PROTOCOL_CACHING.isDebugEnabled())
                    {
                        LOG_PROTOCOL_CACHING_DEBUG("Received explicitely requested remote format #" + AgString::fromInt(id).toUtf8() + ": " + result->toString().toUtf8());
                    }

                    {
                        boost::lock_guard<boost::mutex> lock(cacheLock);
                        cache[id] = result;
                    }
                }
                catch (AggreGateException &ex)
                {
                    throw IllegalStateException("Error obtaining format #" + AgString::fromInt(id) + ": " + ex.getMessage(), ex.getDetails());
                }
            }
            else
            {
                if(Log::PROTOCOL_CACHING.isDebugEnabled())
                {
                    LOG_PROTOCOL_CACHING_DEBUG("Waiting for format #" + AgString::fromInt(id).toUtf8());
                }
                boost::this_thread::sleep(boost::posix_time::millisec(TIMEOUT));
            }
        }

        retry++;
    }
    while (result.get() == NULL && retry < RETRIES);

    if (result.get() == NULL)
    {
        LOG_PROTOCOL_CACHING_WARN("Timeout while getting format #" + AgString::fromInt(id).toUtf8() + ", cache size: " + AgString::fromInt(cache.size()).toUtf8());
        dump();
    }

    return result;
}

void FormatCache::clear()
{
    boost::lock_guard<boost::mutex> lock(cacheLock);
    cache.clear();
}

void FormatCache::dump()
{
    boost::lock_guard<boost::mutex> lock(cacheLock);
    for (std::map<int, TableFormatPtr>::iterator entry = cache.begin(); entry != cache.end(); ++entry)
    {
        if(Log::PROTOCOL_CACHING.isDebugEnabled())
        {
            LOG_PROTOCOL_CACHING_DEBUG("Format cache entry with ID #" + AgString::fromInt(entry->first).toUtf8() + ": " + entry->second->toString().toUtf8());
        }
    }
}
