#include "stdafx.h"
#include "LogServer.h"
#include <ostream>
#include <utility>

#pragma warning (push, 0)
#include "Task.pb.h"
#pragma warning (pop)
using boost::asio::ip::tcp;

namespace MP
{
  LogServer::RawData serialize(const TaskEvent &ev, Serialization::TaskEvent::EventType type)
  {
    Serialization::TaskEvent ser;
    ser.set_id(ev.id);
    ser.set_name(ev.name);
    ser.set_priority(ev.priority);
    ser.set_createtime(ev.createTime);
    ser.set_threadid(ev.threadID);
    ser.set_time(ev.time);
    ser.set_type(type);

    LogServer::RawData data(ser.ByteSize());
    ser.SerializeToArray(&data[0], data.size());
    return data;
  }

  LogServer::LogServer(unsigned short port) 
    : m_port(port), m_acceptor(0) {}

  void LogServer::accept()
  {
      tcp::endpoint ep(tcp::v4(), m_port);
      tcp::acceptor acc(m_service);
      acc.open(ep.protocol());
      acc.set_option(tcp::acceptor::reuse_address(true));
      acc.bind(ep);
      acc.listen();
      m_acceptor = &acc;
      while(isRunning())
      { 
        try
        {
          std::auto_ptr<Connection> conn(new Connection(m_service));
          if(conn->accept(acc) && conn->send(m_threadInfoMsg))
          {
            CriticalSection::scoped_lock lock(m_connLock);
            m_clients.insert(conn.release());
            SPAR_LOG(LogTypes.GENERAL, "Log server: Client connected");
          }          
        }
        catch(std::exception &ex)
        {
          SPAR_LOG(LogTypes.GENERAL_ERROR, "Log server: Error while accepting client: %s", ex.what());
        }
      }
  }

  LogServer::~LogServer()
  {
    stop();
    if(m_acceptor)
    {
      m_acceptor->close();
      m_acceptorThread.join();
    }
    for(ConnectionSet::iterator it = m_clients.begin(); it != m_clients.end(); ++it)
      delete *it;
    m_service.stop();
  }

  void LogServer::onTaskStarted(const TaskEvent &ev)
  {
    push(serialize(ev, Serialization::TaskEvent::STARTED));
  }

  void LogServer::onTaskEnded(const TaskEvent &ev)
  {
    push(serialize(ev, Serialization::TaskEvent::ENDED));
  }

  void LogServer::open(DWORD mainThread, const std::vector<DWORD> &workerThreads)
  {
    GOOGLE_PROTOBUF_VERIFY_VERSION;

    const char * SERVER_VERSION = "2012.2.28.0"; // TODO: Move to config or something

    Serialization::ThreadInfo threadInfo;
    threadInfo.set_version(SERVER_VERSION);
    threadInfo.set_main(mainThread);
    for(std::vector<DWORD>::const_iterator it = workerThreads.begin(); it != workerThreads.end(); ++it)
    {
      threadInfo.add_workers(*it);
    }

    m_threadInfoMsg.resize(threadInfo.ByteSize());
    if(!m_threadInfoMsg.empty())
      threadInfo.SerializeToArray(&m_threadInfoMsg[0], m_threadInfoMsg.size());

    start("Log Server Sender");
    {
      m_acceptorThread = boost::thread(boost::bind(&LogServer::accept, this));
      HANDLE handle = m_acceptorThread.native_handle();
      SetThreadName(GetThreadId(handle), "Log Server Acceptor");
      SetThreadPriority(handle, THREAD_PRIORITY_LOWEST);
    }
    
    SPAR_LOG(LogTypes.GENERAL, "Log server open for business on port %d", static_cast<int>(m_port));
  }

  void LogServer::handle(const RawData &data) const
  {
    CriticalSection::scoped_lock lock(m_connLock);
    ConnectionSet deleted;
    for(ConnectionSet::iterator it = m_clients.begin(); it != m_clients.end(); ++it)
    {
      ConnectionPtr c = *it;
      if(!c->send(data))
        deleted.insert(c);
    }
    for(ConnectionSet::iterator it = deleted.begin(); it != deleted.end(); ++it)
    {
      delete *it;
      m_clients.erase(*it);
    }
  }

  LogServer::Connection::~Connection()
  {
    try
    {
      boost::system::error_code err;
      m_sck.shutdown(boost::asio::ip::tcp::socket::shutdown_both, err);
    }
    catch(std::exception &ex)
    {
      SPAR_LOG(LogTypes.GENERAL_ERROR, "Log server: Error when closing socket: %s", ex.what());
    }
  }

  bool LogServer::Connection::send(const RawData & data)
  {
    try
    {
      boost::system::error_code err;
      size_t len = htonl(data.size());
      
      ::boost::asio::write(m_sck, boost::asio::buffer(&len, sizeof(len)), boost::asio::transfer_all(), err);
      if(!data.empty() && !err)
      {
        ::boost::asio::write(m_sck, boost::asio::buffer(data), boost::asio::transfer_all(), err);
        return !err;
      }
      return data.empty();
    }
    catch(std::exception &ex)
    {
      SPAR_LOG(LogTypes.GENERAL_ERROR, "Log server: Error when sending data: %s", ex.what());
      return false;
    }
  }

  bool LogServer::Connection::accept(boost::asio::ip::tcp::acceptor & acc)
  {
    boost::system::error_code err;
    acc.accept(m_sck, err);
    if(err != 0)
    {
      SPAR_LOG(LogTypes.GENERAL_ERROR, "Log server : Connection was not accepted: %s",  err.message().c_str());
      return false;
    }
    return true;
  }
}