/////////////////////////////////////////////////////////////////////
// Communicator.cpp - communication channel based on sockets       //
//                                                                 //
// Lanaguage:     Visual C++, VS2005                               //
// Platfrom:      Dell Dimension 9150, Windows XP Pro, SP 2.0      //
// Application:   Demonstration for CSE775 - Distributed Objects   //
// Author:        Jim Fawcett, CST 2-187, Syracuse University      //
//                (315) 443-3980, jfawcett@twcny.rr.com            //
/////////////////////////////////////////////////////////////////////

#include "Communicator.h"

/////////////////////////////////////////////////////////////////////
// SendProc class - send thread processing

class SendProc : public processing
{
public:
  SendProc(Socket* pSock, BQueue<std::string>* pQ)
    : pSendSock(pSock), pSendQ(pQ), sendStop(false) {}
  void run()
  {
    do
    {
      if(sendStop)
        break;
      std::string msg = pSendQ->deQ();
      pSendSock->send(msg);
//      sout << "\n  send message";
    } while(true);  // !sendStop
  }
private:
  Socket* pSendSock;
  BQueue<std::string>* pSendQ;
  volatile bool sendStop;
};
//
/////////////////////////////////////////////////////////////////////
// RecvProc class - recv thread processing

class RecvProc : public processing
{
public:
  RecvProc(Socket* pSock, BQueue<std::string>* pQ)
    : pRecvSock(pSock), pRecvQ(pQ), recvStop(false) {}
  void run()
  {
    do
    {
      if(recvStop)
        break;
//      sout << "\n  starting to get message from socket";
      pRecvQ->enQ(pRecvSock->recv());
//      sout << "\n  received message";
    } while(true);  // !recvStop
  }
private:
  Socket* pRecvSock;
  BQueue<std::string>* pRecvQ;
  volatile bool recvStop;
};

/////////////////////////////////////////////////////////////////////
// ListenProc class - listener thread processing
//
//   Makes connected socket for receiving, then dies
//
class ListenProc : public processing
{
public:
  ListenProc(Communicator* pCommunicator) : pComm(pCommunicator) {}
  void run()
  {
    pComm->pRecvQ = new BQueue<message>;
    pComm->pRecvSock = new Socket(pComm->pListener->waitForConnect());
    pComm->pRecvSock->setTerminator("</msg>");
    pComm->pRecvThread 
      = CreateThreadOnHeap(new RecvProc(pComm->pRecvSock,pComm->pRecvQ));
    pComm->pRecvThread->start();
  }
private:
  Communicator* pComm;
};
//
//----< ctor - zero pointers for safe deletion >---------------------

Communicator::Communicator() 
  : pListener(0), 
    pRecvSock(0), pRecvQ(0), recvrStop(false), 
    pSendSock(0), pSendQ(0), sendrStop(false) {}

//---< detor: delete heap objects - safe even if not constructed >---

Communicator::~Communicator()
{
  delete pListener;
  delete pRecvSock;
  delete pSendSock;
  delete pSendQ;
  delete pRecvQ;
}
//----< spawn listener on its own thread >---------------------------
//
//  Note:  objects not created for send only
//  To Do: provide mechanism to gracefully shutdown listener
//
void Communicator::listen(int port)
{
  recvrStop = false;
  delete pListener;
  pListener = new SocketListener(port);
  pListenProc = new ListenProc(this);
  pListenThread = CreateThreadOnHeap(pListenProc);
  pListenThread->start();
}
//----< connect to another communicator >----------------------------
//
//  Note:  objects not created for receive only
//  To Do: provide mechanism to gracefully shutdown connection
//         and reconnect to new receiver
//
bool Communicator::connect(const std::string &url, int port)
{
  sendrStop = false;
  pSendSock = new Socket();
  pSendSock->setTerminator("</msg>");
  if(pSendSock->connect(url,port))
  {
    if(!pSendQ)
      pSendQ = new BQueue<std::string>;
    pSendThread = CreateThreadOnHeap(new SendProc(pSendSock,pSendQ));
    pSendThread->start();
    return true;
  }
  return false;
}
//
//----< pull message from recv queue >-------------------------------

std::string Communicator::getMessage()
{
  if(!pRecvQ)
    throw std::exception("attempt to dequeue when not listening");
  return pRecvQ->deQ();
}
//----< push message onto send queue >-------------------------------

void Communicator::postMessage(const std::string &msg)
{
  if(!pSendQ)
    throw std::exception("attempt to enqueue when not connected");
  pSendQ->enQ(msg);
}
//----< any messages in receive queue? >-----------------------------

bool Communicator::hasMessage()
{
  return (pRecvQ->size() > 0);
}
//
//----< test stub >--------------------------------------------------

#ifdef TEST_COMMUNICATOR

#include <iostream>
#include <iomanip>

void main()
{
  std::cout << "\n  Testing Communicator, ListenProc, RecvProc, and SendProc";
  std::cout << "\n ===========================================================\n";

  std::cout << "\n  simple back and forth messaging";
  std::cout << "\n ---------------------------------";

  try
  {
    Communicator clientComm, serverComm;

    clientComm.listen(4096);
    serverComm.listen(4097);
    clientComm.connect("localhost",4097);
    serverComm.connect("localhost",4096);

    clientComm.postMessage("<msg>hi from client</msg>");
    message srMsg = serverComm.getMessage();
    std::cout << "\n  server received " << srMsg;
    serverComm.postMessage("<msg>hello from server</msg>");
    message crMsg = clientComm.getMessage();
    std::cout << "\n  client received " << crMsg;

    clientComm.postMessage("<msg>hi from client</msg>");
    srMsg = serverComm.getMessage();
    std::cout << "\n  server received " << srMsg;
    serverComm.postMessage("<msg>hello from server</msg>");
    crMsg = clientComm.getMessage();
    std::cout << "\n  client received " << crMsg;
    std::cout << "\n\n";

    std::cout << "\n  Create child thread to send messages";
    std::cout << "\n --------------------------------------";

    Communicator parentComm, childComm;
    parentComm.listen(5098);
    childComm.listen(5099);

//
///////////////////////////////////////////////////////////
//  No need to connect if you only receive
//
    if(!parentComm.connect("localhost",5099))
      std::cout << "\n  failed to connect to child";

//////////////////////////////////////////
//  No need to listen if you only send
//
    if(!childComm.connect("localhost",5098))
      std::cout << "\n  failed to connect to parent";

    class MessageProc : public processing
    {
    public:
      MessageProc(Communicator* pCommunicator) : pComm(pCommunicator) {} 
      void run()
      {
        try
        {
          for(int i=0; i<20; ++i)
          {
            message msg = std::string("<msg>message #") + convert(i) + "</msg>";
            pComm->postMessage(msg);
            if(pComm->hasMessage())
            {
              sout << locker << "\n.." << pComm->getMessage() << unlocker;
            }
            else
            {
              pComm->postMessage("<msg>no message</msg>");
              ::Sleep(100);
            }
          }
          message msg = "<msg>quit</msg>";  // stop the parent message stream
          pComm->postMessage(msg);

          // this call blocks
          sout << locker << "\n.." << pComm->getMessage() << unlocker;

          // these won't block because of the test
          while(pComm->hasMessage())
            sout << locker << "\n.." << pComm->getMessage() << unlocker;
        }
        catch(std::exception& e)
        {
          sout << locker << "\n\n  " << e.what() << "\n\n" << unlocker;
        }
      }
    private:
      Communicator* pComm;
    };
//
    // child thread starts sending messages

    MessageProc childMessageProc(&childComm);
    thread child(&childMessageProc);
    parentComm.postMessage("<msg>start your messages</msg>");
    child.start();

    // parent thread starts collecting messages
    message msg;
    do
    {
      msg = parentComm.getMessage();
      sout << locker << "\n  " << msg << unlocker;
      message sendMsg = "<msg>got your message</msg>";
      parentComm.postMessage(sendMsg);
    } while(msg != "<msg>quit</msg>");

    child.wait();
    sout << "\n\n";
  }
  catch(std::exception& e)
  {
    sout << locker << "\n  " << e.what() << "\n\n" << unlocker;
  }
}
#endif