#include "../shared/SimpleLogFile.h"
#include "../shared/PollSet.h"
#include "../shared/GlobalConfigFile.h"

#include "CandleClient.h"


/////////////////////////////////////////////////////////////////////
// CandleClient::GetIntraday
/////////////////////////////////////////////////////////////////////

void CandleClient::GetIntraday::sendResult(std::string const &result)
{
  _result = result;
  _conditionVar.set();
}

std::string const &CandleClient::GetIntraday::getResult()
{
  _conditionVar.waitForever();
  return _result;
}


/////////////////////////////////////////////////////////////////////
// CandleClient::GetDaily
/////////////////////////////////////////////////////////////////////

void CandleClient::GetDaily::sendResult(std::string const &result)
{
  _result = result;
  _conditionVar.set();
}

std::string const &CandleClient::GetDaily::getResult()
{
  _conditionVar.waitForever();
  return _result;
}


/////////////////////////////////////////////////////////////////////
// CandleClient
/////////////////////////////////////////////////////////////////////

std::string CandleClient::makeStartList(AllRowTimes const &allRowTimes)
{
  std::string result;
  for (auto it = allRowTimes.begin(); it != allRowTimes.end(); it++)
  {
    if (!result.empty())
      result += ',';
    result += ntoa(it->start);
  }
  return result;
}

std::string CandleClient::makeEndList(AllRowTimes const &allRowTimes)
{
  std::string result;
  for (auto it = allRowTimes.begin(); it != allRowTimes.end(); it++)
  {
    if (!result.empty())
      result += ',';
    result += ntoa(it->end);
  }
  return result;
}

SingleCandle CandleClient::extractSingleCandle(std::string const &bytes)
{
  SingleCandle::ByStartTime result;
  const bool success = SingleCandle::unmarshal(bytes, result);
  if ((!success) || (result.size() > 1))
  { // This should never happen.
    TclList msg;
    msg<<FLF<<"INVALID message from server"<<bytes.size()<<result.size();
    sendToLogFile(msg);
  }
  if (result.size() == 1)
    return result.begin()->second;
  else
    return SingleCandle();
}

CandleClient::CandleClient() :
  ForeverServerConnection("CandleClient")
{
  getAndParseAddress("candle_server");
  start();
}

CandleClient::~CandleClient()
{ // Not implemented.  It's not really needed.
  assert(false);
}

void CandleClient::onNewConnection(TalkWithServer *talkWithServer)
{
  {
    TalkWithServer::Message message;
    message["command"] = "login";
    message["username"] = getConfigItem("candle_server_username");
    message["password"] = getConfigItem("candle_server_password");
    // TODO it's tempting to use the session id, like a normal client would.
    talkWithServer->sendMessage(message, this, ciLogin, true);
  }
  // Send/resend any old messages.
  {
    std::vector< TodaysCandles * > toResend;
    toResend.reserve(_todaysCandlesPending.size());
    for (auto it = _todaysCandlesPending.begin(); 
	 it != _todaysCandlesPending.end(); 
	 it++)
      toResend.push_back(it->second);
    _todaysCandlesPending.clear();
    for (auto it = toResend.cbegin(); it != toResend.end(); it++)
      sendRequestToServer(talkWithServer, *it);
  }
  {
    std::vector< GetIntraday * > toResend;
    toResend.reserve(_getIntradayPending.size());
    for (auto it = _getIntradayPending.begin(); 
	 it != _getIntradayPending.end(); 
	 it++)
      toResend.push_back(it->second);
    _getIntradayPending.clear();
    for (auto it = toResend.cbegin(); it != toResend.end(); it++)
      sendRequestToServer(talkWithServer, *it);
  }
  {
    std::vector< GetDaily * > toResend;
    toResend.reserve(_getDailyPending.size());
    for (auto it = _getDailyPending.begin(); 
	 it != _getDailyPending.end(); 
	 it++)
      toResend.push_back(it->second);
    _getDailyPending.clear();
    for (auto it = toResend.cbegin(); it != toResend.end(); it++)
      sendRequestToServer(talkWithServer, *it);
  }
}

void CandleClient::sendRequestToServer(TalkWithServer *connection, 
				       TodaysCandles *request)
{
  TalkWithServer::Message message;
  message["command"] = "get_todays_candles";
  message["symbol"] = request->getSymbol();
  TalkWithServer::CancelId messageId = 
    connection->sendMessage(message, this, ciTodaysCandles, false);
  _todaysCandlesPending[messageId] = request;
}

void CandleClient::onTodaysCandlesMessage(std::string const &bytes,
					  TalkWithServer::CancelId cancelId)
{
  const std::map< TalkWithServer::CancelId, TodaysCandles * >::iterator it =
    _todaysCandlesPending.find(cancelId);
  if (it == _todaysCandlesPending.end())
  { // This should never happen.
    TclList msg;
    msg<<FLF<<"UNEXPECTED message from server"<<bytes.size()<<cancelId;
    sendToLogFile(msg);
    return;
  }
  TodaysCandles *const request = it->second;
  _todaysCandlesPending.erase(it);
  request->callbackId = request->_finalCallbackId;
  const bool success = SingleCandle::unmarshal(bytes, request->data);
  request->_finalListener->newRequest(request);
  if (!success)
  { // This should never happen.
    TclList msg;
    msg<<FLF<<"INVALID message from server"<<bytes.size()<<cancelId;
    sendToLogFile(msg);
    return;
  }
}

void CandleClient::sendRequestToServer(TalkWithServer *connection,
				       GetIntraday *request)
{
  TalkWithServer::Message message;
  message["command"] = "get_intraday_candle";
  message["symbol"] = request->getSymbol();
  message["start"] = request->getStart();
  message["end"] = request->getEnd();
  TalkWithServer::CancelId messageId = 
    connection->sendMessage(message, this, ciIntradayCandle, false);
  _getIntradayPending[messageId] = request;
}

void CandleClient::onIntradayMessage(std::string const &bytes,
				     TalkWithServer::CancelId cancelId)
{
  const std::map< TalkWithServer::CancelId, GetIntraday * >::iterator it =
    _getIntradayPending.find(cancelId);
  if (it == _getIntradayPending.end())
  { // This should never happen.
    TclList msg;
    msg<<FLF<<"UNEXPECTED message from server"<<bytes.size()<<cancelId;
    sendToLogFile(msg);
    return;
  }
  GetIntraday *const request = it->second;
  _getIntradayPending.erase(it);
  request->sendResult(bytes);
}

/*
static std::string debugDump(TalkWithServer::Message const &message)
{
  TclList result;
  for (TalkWithServer::Message::const_iterator it = message.begin();
       it != message.end();
       it++)
    result<<(TclList()<<it->first<<it->second);
  return result;
}
*/

void CandleClient::sendRequestToServer(TalkWithServer *connection,
				       GetDaily *request)
{
  TalkWithServer::Message message;
  message["command"] = "get_daily_candle";
  message["symbol"] = request->getSymbol();
  message["date"] = request->getDate();
  //std::cout<<"CandleClient::sendRequestToServer(daily)"
  //	   <<debugDump(message)<<std::endl;
  TalkWithServer::CancelId messageId = 
    connection->sendMessage(message, this, ciDailyCandle, false);
  _getDailyPending[messageId] = request;
}

void CandleClient::onDailyMessage(std::string const &bytes,
				  TalkWithServer::CancelId cancelId)
{
  const std::map< TalkWithServer::CancelId, GetDaily * >::iterator it =
    _getDailyPending.find(cancelId);
  if (it == _getDailyPending.end())
  { // This should never happen.
    TclList msg;
    msg<<FLF<<"UNEXPECTED message from server"<<bytes.size()<<cancelId;
    sendToLogFile(msg);
    return;
  }
  GetDaily *const request = it->second;
  _getDailyPending.erase(it);
  request->sendResult(bytes);
}

void CandleClient::onMessage(std::string bytes, int clientId,
			     TalkWithServer::CancelId cancelId)
{
  //std::cout<<(std::string)(TclList()<<FLF<<bytes<<clientId<<cancelId)<<std::endl;
  ThreadMonitor &tm = ThreadMonitor::find();
  switch (clientId)
  {
  case ciLogin:
    break;
  case ciTodaysCandles:
    tm.increment("ciTodaysCandles");
    onTodaysCandlesMessage(bytes, cancelId);
    break;
  case ciIntradayCandle:
    tm.increment("ciIntradayCandle");
    onIntradayMessage(bytes, cancelId);
    break;
  case ciDailyCandle:
    tm.increment("ciDailyCandle");
    onDailyMessage(bytes, cancelId);
    break;
  default:
    ForeverServerConnection::onMessage(bytes, clientId, cancelId);
    break;
  }
}

void CandleClient::requestTodaysCandles(std::string const &symbol,
					RequestListener *listener, 
					int callbackId)
{
  TodaysCandles *request = new TodaysCandles(symbol, listener, callbackId);
  doWorkInThread([this, request]() { sendRequestToServer(getConnection(), 
							 request); });
}

SingleCandle
CandleClient::getIntraday(std::string const &symbol, time_t start, time_t end)
{
  GetIntraday request(symbol, start, end);
  doWorkInThread([this, &request]() { sendRequestToServer(getConnection(),
							  &request); });
  return extractSingleCandle(request.getResult());
}

SingleCandle CandleClient::getDaily(std::string const &symbol, time_t date)
{
  GetDaily request(symbol, date);
  //std::cout<<"GetDaily *request = new GetDaily(symbol, date);"<<std::endl;
  doWorkInThread([this, &request]() { sendRequestToServer(getConnection(),
							  &request); });
  return extractSingleCandle(request.getResult());
}

std::string CandleClient::getIntraday(std::string const &symbol,
				      AllRowTimes const &allRowTimes)
{
  GetIntraday request(symbol, allRowTimes);
  doWorkInThread([this, &request]() { sendRequestToServer(getConnection(),
                                                          &request); });
  return request.getResult();
}

std::string CandleClient::getDaily(std::string const &symbol,
				   AllRowTimes const &allRowTimes)
{
  GetDaily request(symbol, allRowTimes);
  doWorkInThread([this, &request]() { sendRequestToServer(getConnection(),
                                                          &request); });
  return request.getResult();
}

static bool candleClientInitialized = false;
static CandleClient *instance = NULL;
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

CandleClient *CandleClient::instance()
{
  if (!candleClientInitialized)
  {
    pthread_mutex_lock(&mutex);
    if (!candleClientInitialized)
    {
      if (getConfigItem("use_candle_server") == "1")
	::instance = new CandleClient();
      else 
	::instance = NULL;
      candleClientInitialized = true;
    }
    pthread_mutex_unlock(&mutex);
  }
  return ::instance;
}