/////////////////////////////////////////////////////////////////////
// Comm.h - message-passing communication facility                 //
//                                                                 //
// Jim Fawcett, CSE687-OnLine Object Oriented Design, Fall 2017    //
/////////////////////////////////////////////////////////////////////

#include "Comm.h"
#include "../Logger/Logger.h"
#include "../Utilities/Utilities.h"
#include "../Cpp11-BlockingQueue/Cpp11-BlockingQueue.h"
#include <iostream>
#include <functional>

using namespace MsgPassingCommunication;
using namespace Sockets;

Receiver::Receiver(EndPoint ep) : listener(ep.port) 
{
  StaticLogger<1>::write("starting Receiver");
}

BlockingQueue<Message>& Receiver::queue()
{
  return rcvQ;
}

template<typename CallableObject>
void Receiver::start(CallableObject& co)
{
  listener.start(co);
}

void Receiver::stop()
{

}
Message Receiver::getMessage()
{
  return rcvQ.deQ();
}

Sender::~Sender()
{
  if (sendThread.joinable())
    sendThread.join();
}
void Sender::start()
{
  std::function <void()> threadProc = [&]() {
    Message msg = sndQ.deQ();
    std::string msgStr = msg.toString();
    StaticLogger<1>::write("\n  send thread dequeued message: " + msgStr);
    if (msg.body == "quit")
    {
      StaticLogger<1>::write("\n  send thread shutting down");
      return;
    }
    StaticLogger<1>::write("\n  sending " + Utilities::Converter<size_t>::toString(msgStr.length()));
    bool sendRslt = connecter.send(msgStr.length(), (Socket::byte*)msgStr.c_str());
  };
  std::thread t(threadProc);
  sendThread = std::move(t);
}

void Sender::stop()
{
  Message msg;
  msg.body = "quit";
  postMessage(msg);
  connecter.shutDown();
}
bool Sender::connect(EndPoint ep)
{
  return connecter.connect(ep.address, ep.port);
}

void Sender::postMessage(Message& msg)
{
  sndQ.enQ(msg);
}

bool Sender::sendFile(const std::string& fileName)
{
  return false;
}

class ClientHandler
{
public:
  ClientHandler(BlockingQueue<Message>& q) : q_(q) {
    StaticLogger<1>::write("\n  starting ClientHandler");
  }
  ~ClientHandler() { std::cout << "\n  ClientHandler destroyed;"; }

  std::string readMsg(Socket& socket)
  {
    std::string temp, msgString;
    while (socket.validState())
    {
      temp = socket.recvString('\n');
      msgString += temp;
      StaticLogger<1>::write("--" + temp);
      if (temp.length() < 2)
        break;
    }
    return msgString;
  }
  void operator()(Socket socket)
  {
    while (socket.validState())
    {
      std::string msgString = readMsg(socket);
      Message msg = Message::fromString(msgString);
      StaticLogger<1>::write("read message from rcvr's socket: " + msg.toString());
      q_.enQ(msg);
      std::cout << msgString;
      if (msg.body == "quit")
        break;
    }
  }
private:
  BlockingQueue<Message>& q_;
};

#ifdef TEST_COMM

int main()
{
  StaticLogger<1>::attach(&std::cout);
  StaticLogger<1>::start();
  SocketSystem ss;
  EndPoint ep;
  ep.port = 9090;
  ep.address = "localhost";
  Receiver rcvr(ep);
  BlockingQueue<Message>& Q = rcvr.queue();

  ClientHandler ch(Q);
  rcvr.start(ch);

  Sender sndr;
  sndr.start();
  bool connected = sndr.connect(ep);
  Message msg;
  msg.name = "msg #1";
  msg.to = EndPoint("localhost", 8080);
  msg.from = msg.to;
  msg.command = "do it";
  msg.body = "zzz";
  sndr.postMessage(msg);
  sndr.postMessage(msg);

  Message rcvdMsg = rcvr.getMessage();  // blocks until message arrives
  std::cout << "\n  received message:" << rcvdMsg.toString();
  std::cin.get();
  sndr.stop();
  std::cout << "\n\n";
  return 0;
}
#endif