#include <assert.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <unistd.h>
#include <fcntl.h>
#include <poll.h>
#include <errno.h>

#include "MiscSupport.h"
#include "ZLibMalloc.h"
#include "SimpleLogFile.h"
#include "ThreadMonitor.h"
#include "GlobalConfigFile.h"

#include "TalkWithServer64.h"


/////////////////////////////////////////////////////////////////////
//
/////////////////////////////////////////////////////////////////////

enum class DumpAll {Yes, No, Unknown};
static DumpAll dumpAll = DumpAll::Unknown;

static void initDumpAll()
{
  if (dumpAll == DumpAll::Unknown)
    // getConfigItem: dump_all=1 means send a copy of every message in or out
    // getConfigItem:   of the system to the log file.  This can be useful in
    // getConfigItem:   debugging, but it makes really big log files very
    // getConfigItem:   quickly.  This setting is shared by several libraries.
    dumpAll = (getConfigItem("dump_all") == "1")?DumpAll::Yes:DumpAll::No;
}


/////////////////////////////////////////////////////////////////////
// TalkWithServer64
/////////////////////////////////////////////////////////////////////

TalkWithServer64::MessageId TalkWithServer64::_lastMessageId = 0;

TalkWithServer64::MessageId TalkWithServer64::getNextMessageId()
{ // For simplicity make this atomic.  A __thread variable would probably
  // have worked, too.  Each client probably stays in one thread.  But it
  // was just as easy to do it this way and have fewer assumptions.
  return __sync_add_and_fetch(&_lastMessageId, 1);
}

TalkWithServer64::TalkWithServer64(std::string const &name) :
  _name(name),
  _uniqueId(getNextMessageId()),
  _tcpIpConnection(name),
  _status(NotYetOpen),
  _allResponsesListener(NULL)
{
  initDumpAll();
}

TalkWithServer64::~TalkWithServer64()
{
  assert(_atServer.empty() && _localQueue.empty());
}

void TalkWithServer64::connect(std::string address, std::string port, 
			       bool asyncConnect)
{
  assert(_status == NotYetOpen);
  _tcpIpConnection.connect(address, port, asyncConnect);
  _status = Open;
  _tcpIpConnection.toSend =
    "command=set_output&mode=zlib64\r\ncommand=set_input&mode=zlib\r\n";
  checkForClosed();
}

void TalkWithServer64::checkForClosed()
{
  if (_tcpIpConnection.closed())
    disconnect();
}

void TalkWithServer64::disconnect()
{
  if (_status != Closed)
  {
    _status = Closed;
    _tcpIpConnection.close();
    for (auto it = _atServer.begin(); it != _atServer.end(); it++)
    {
      LocalWrapper localWrapper;
      localWrapper.messageId = it->first;
      localWrapper.success = false;
      _localQueue.push_back(localWrapper);
    }
  }
}

bool TalkWithServer64::disconnected()
{
  return _status == Closed;
}

void TalkWithServer64::cancel(MessageId id)
{
  _atServer.erase(id);
}

void TalkWithServer64::cancelAll()
{
  _atServer.clear();
}

void
TalkWithServer64::setAllResponsesListener(IAllResponsesListener *newListener)
{
  _allResponsesListener = newListener;
  if (newListener == NULL)
    _allResponses.clear();
}

int TalkWithServer64::pendingResponseCount()
{
  return _localQueue.size();
}

void TalkWithServer64::doResponses()
{ // It is possible that during this one of these callbacks, new callbacks
  // will be added to the list.  It's is unlikely that a valid response will
  // appear in the list, but it is certaily possible for someone to disconnect
  // the session in a callback, so the list will have new error responses.
  // It is also possible for a request to be canceled while in a callback.
  //
  // Strange.  The comment above comes from TalkWithServer.C.  But that file
  // used iterators to walk through the list.  An iterator could be invalidated
  // if the list grows.  I wonder why I made that comment but didn't do
  // anything about it.
  for (size_t i = 0; i < _localQueue.size(); i++)
  {
    LocalWrapper const &localWrapper = _localQueue[i];
    //TclList msg;
    //msg<<FLF<<"messageId"<<localWrapper.messageId
    //   <<"body"<<localWrapper.body
    //   <<"success"<<localWrapper.success;
    const auto it = _atServer.find(localWrapper.messageId);
    if (it != _atServer.end())
    { // Not yet canceled.
      ServerWrapper const &serverWrapper = it->second;
      //msg<<"clienId"<<serverWrapper->clientId
      //   <<"streaming"<<serverWrapper->streaming;
      if (localWrapper.success)
      {
        serverWrapper.listener->onMessage(localWrapper.body,
					  serverWrapper.clientId,
					  localWrapper.messageId);
	if (!serverWrapper.streaming)
	  // Clean up. Without this we would have a memory leak.
	  _atServer.erase(it);
      }
      else
      {
        serverWrapper.listener->onAbort(serverWrapper.clientId,
					localWrapper.messageId);
	_atServer.erase(it);
      }
    }
    //sendToLogFile(msg);
  }
  _localQueue.clear();

  if (_allResponsesListener)
  { // It seems unlikely that this list would change during the callbacks.
    // But it's possible.  I could list out the conditions involved -- things
    // to avoid in a callback -- but as long as we're discussing it, let's just
    // make this bulletproof.
    std::vector< AllResponsesWrapper > toReport;
    _allResponses.swap(toReport);
    for (auto it = toReport.begin(); it != toReport.end(); it++)
      _allResponsesListener->onResponse(it->body, it->messageId);
  }
  else  // No listener
    _allResponses.clear();
}

void TalkWithServer64::sendMessage(Message const &message)
{
  if (!disconnected())
    _outgoing.push_back(message);
}

void TalkWithServer64::sendMessageWithId(Message message,
                                         IMessageListener *listener,
                                         int64_t clientId,
                                         MessageId messageId,
                                         bool streaming)
{
  assert(listener);
  assert(!_atServer.count(messageId));
  ServerWrapper serverWrapper;
  serverWrapper.listener = listener;
  message["message_id"] = ntoa(messageId);
  serverWrapper.streaming = streaming;
  serverWrapper.clientId = clientId;
  _atServer[messageId] = serverWrapper;
  if (disconnected())
  { // Fail immediately.
    LocalWrapper localWrapper;
    localWrapper.messageId = messageId;
    localWrapper.success = false;
    _localQueue.push_back(localWrapper);
  }
  else
    _outgoing.push_back(message);
}

TalkWithServer64::MessageId
TalkWithServer64::sendMessage(Message message,
                              IMessageListener *listener, int64_t clientId,
                              bool streaming)
{
  const MessageId messageId = getNextMessageId();
  sendMessageWithId(message, listener, clientId, messageId, streaming);
  return messageId;
}

bool TalkWithServer64::wantsRead()
{
  return _tcpIpConnection.wantsRead();
}

bool TalkWithServer64::wantsWrite()
{
  return
    // The TalkWithServer64 has something in its outgoing buffer and the
    // TcpIpConnection is ready to accept it.
    (_tcpIpConnection.readyToSend() &&!_outgoing.empty())
    // Or the TcpIpConnection has something in its outgoing buffer.
    || _tcpIpConnection.wantsWrite();
}

int TalkWithServer64::getHandle()
{
  return _tcpIpConnection.getHandle();
}

void TalkWithServer64::addStringWithSize(std::string toCompress)
{
  addInt32(toCompress.size());
  _zLibOutputStream.add(toCompress);
}

void TalkWithServer64::addInt32(int32_t toCompress)
{
  _zLibOutputStream.add(&toCompress, 4);
}

static const std::string s_marshal = "marshal";
static const std::string s_tcp_ip = "TCP/IP";
static const std::string s_unmarshal = "unmarshal";

void TalkWithServer64::wakeUp()
{
  TclList msg;
  if (dumpAll == DumpAll::Yes)
    msg<<FLF<<_name<<"outgoing message count"<<_outgoing.size();
  ThreadMonitor::SetState state(s_marshal);
  if (_tcpIpConnection.readyToSend() && !_outgoing.empty())
  { // Copy outgoing messages from here to the tcp/ip connection.
    // Don't do this before we have to.  Compression will work best if
    // we don't flush the buffer until the last moment.
    for (std::vector< Message >::const_iterator messageIt =
           _outgoing.begin();
         messageIt != _outgoing.end();
         messageIt++)
    {
      for (Message::const_iterator fieldIt = messageIt->begin();
           fieldIt != messageIt->end();
           fieldIt++)
      {
        addStringWithSize(fieldIt->first);
        addStringWithSize(fieldIt->second);
      }
      addInt32(0);
    }
    if (dumpAll == DumpAll::Yes)
    {
      for (std::vector< Message >::const_iterator messageIt =
	     _outgoing.begin();
	   messageIt != _outgoing.end();
	   messageIt++)
      {
	TclList msg;
	msg<<FLF<<_name<<"dump_all sending";
	TclList fields;
	for (Message::const_iterator fieldIt = messageIt->begin();
	     fieldIt != messageIt->end();
	     fieldIt++)
	  msg<<fieldIt->first<<fieldIt->second;
	sendToLogFile(msg);
      }
    }
    _outgoing.clear();
    _zLibOutputStream.flush();
    if (dumpAll == DumpAll::Yes)
      msg<<"previous outgoing buffer size"<<_tcpIpConnection.toSend.length()
	 <<"additional outgoing bytes"<<_zLibOutputStream.output.length();
    _tcpIpConnection.toSend += _zLibOutputStream.output;
    _zLibOutputStream.output.clear();
  }
  if (!_zLibOutputStream.valid())
  { // This should never happen!  Maybe out of memory?  Or a code error on
    // our part.
    TclList msg;
    msg<<FLF<<_name<<"!_zLibOutputStream.valid()";
    sendToLogFile(msg);
    disconnect();
    return;
  }
  state.setState(s_tcp_ip);
  _tcpIpConnection.wakeUp();
  if (dumpAll == DumpAll::Yes)
    msg<<"outgoing bytes not sent"<<_tcpIpConnection.toSend.length()
       <<"bytes read"<<_tcpIpConnection.received.length();
  state.setState(s_unmarshal);
  if (!_tcpIpConnection.received.empty())
  {
    //std::cout<<"before receive, _zLibInputStream.size() = "
    //       <<_zLibInputStream.output.size()<<std::endl;
    _zLibInputStream.add(_tcpIpConnection.received);
    //std::cout<<"received "<<_tcpIpConnection.received.size()
    //       <<" bytes of compressed data"<<std::endl
    //       <<"_zLibInputStream.size() = "<<_zLibInputStream.output.size()
    //       <<std::endl;
    _tcpIpConnection.received.clear();
    if (!_zLibInputStream.valid())
    { // This probably means the data got corrupted.  Note that there is no
      // way to end the stream in our protocol; it's made to go on forever.
      // The best thing to do is to close the connection and try again.
      // Note that we abort immediately.  We do not attempt to read any
      // data which might have been received before the bad data.  zlib's
      // ability to detect bad data is very limited.  Often it gives us
      // extra stuff after the last of the good bytes.  Often the extra
      // stuff is a copy of earlier good bytes, so it looks like a good
      // message, but it is not.  The safest thing is to stop immediately.
      TclList msg;
      msg<<FLF<<_name<<"!_zLibInputStream.valid()";
      sendToLogFile(msg);
      disconnect();
      return;
    }
    int consumed = 0;
    int remaining = _zLibInputStream.output.size();
    char const *const start = _zLibInputStream.output.data();
    while (true)
    { // id followed by body size followed by body.
      //std::cout<<"top:  consumed="<<consumed<<
      //  ", remaining="<<remaining<<std::endl;
      if (remaining < 12)
        // Not enough space for a header.
        break;
      const MessageId id = *(int64_t *)(start + consumed);
      //std::cout<<"id="<<id<<std::endl;
      const int size = *(int32_t *)(start + consumed + 8);
      //std::cout<<"size="<<size<<std::endl;
      if (size < 0)
      { // Invalid message.
        TclList msg;
        msg<<FLF<<_name<<"Invalid message header"<<id<<size;
        sendToLogFile(msg);
        disconnect();
        return;
      }
      if (remaining < 12 + size)
        // We have not loaded the entire message yet.
        break;
      const std::string body = std::string(start + 12 + consumed, size);
      if (id == (MessageId)-1)
      { // This message was from the micro_proxy and it is addressed to this
	// object.  It says that one of the servers on the far side of the
	// proxy closed its connection.
	//
	// Currently we are taking the simplest correct action.  We are
	// closing the connection between this object and the micro_proxy.
	//
	// This could be handled more delicately.  We could send notices
	// to only the listeners who were actually talking with the server
	// in question.  The body of this message gives us enough information
	// to do just that.  See OneServerDisconnected() in TalkWithServer.cs
	// for an example.
	//
	// One possible issue, if we wanted to handle this more delicately.
	// That protocol requires that all message ids be sent in order.
	// Currently ServerConnection64 mostly enforces that.  However,
	// ServerConnection64::onNewConnection() allows people to insert
	// new messages (like log in) before queued up messages (things which
	// assume we are already logged in).  Care would have to be taken here.
	// Note that onNewConnection() is a protected virtual method.  It's
	// pretty much assumed that anyone who uses ServerConnection64 will
	// override that method, so you'll have to search all of
	// cpp_alert_server/source/ for this problem.
	TclList msg;
        msg<<FLF<<_name<<"id == (MessageId)-1"<<body;
        sendToLogFile(msg);
        disconnect();
        return;
      }
      if (_atServer.count(id))
      {
        LocalWrapper localWrapper;
        localWrapper.messageId = id;
        localWrapper.body = body;
        localWrapper.success = true;
        _localQueue.push_back(localWrapper);
        //std::cout<<"cancelId="<<localWrapper.cancelId
        //<<", clientId="<<localWrapper.clientId
        //<<", body="<<tclQuote(localWrapper.body)<<std::endl;
      }
      if (_allResponsesListener)
      {
        AllResponsesWrapper wrapper;
        wrapper.messageId = id;
        wrapper.body = body;
        _allResponses.push_back(wrapper);
      }
      //if (dumpAll == DumpAll::Yes)
      //{
      //TclList msg;
      //msg<<FLF<<_name<<id<<body;
      //sendToLogFile(msg);
      //}
      // Skip past this message when we look for the next one.
      consumed += size + 12;
      remaining -= size + 12;
    }
    if (dumpAll == DumpAll::Yes)
      msg<<"bytes unmarshaled"<<consumed
	 <<"input remaining"<<(_zLibInputStream.output.length()-consumed);
    // Remove every message we've handled from the buffer.  Remove them all
    // at once, rather than one at a time, so we don't have an O(n*n)
    // algorithm.  Possibly leave a partial message in the buffer.
    _zLibInputStream.output.erase(0, consumed);
  }
  if (dumpAll == DumpAll::Yes)
  {
    msg<<"incomming messages waiting"
       <<_localQueue.size() + _allResponses.size();
    sendToLogFile(msg);
  }
  checkForClosed();
}