/////////////////////////////////////////////////////////////////////
//  socks.cpp   -   socket client and multi-threaded server        //
//                  prototype for project #2                       //
//  ver 1.9                                                        //
//  Language:       Visual C++, ver 6.0, SP 5                      //
//  Platform:       Dell Dimension 8100, Win 2000 Pro, SP 2        //
//  Application:    CSE691/791 - SWM&A Demonstration               //
//  Author:         Jim Fawcett, CST 2-187, Syracuse Univ          //
//                  (315) 443-3948, jfawcett@twcny.rr.com          //
/////////////////////////////////////////////////////////////////////

#include <iostream>
#include <process.h>
#include "socks.h"      // must put here due to dependencies in Win32 defs
#include "lockingPtr.h"
#include "sysUtils.h"
#include "msgHnd.h"
#include "connMgr.h"
#include "thrdsWithProc.h"
using namespace std;

const int recvQlock = 101;  // template argument for recvQ's GlobalMutex

//----< handle WinSock errors >--------------------------------------
//
//  Reports Win32 system error messages in one of three ways, to console,
//  to a frameWindow (not implemented yet), or as a thrown exception.  
//
//       enum reportMethod { console, window, exception };

void reportError(reportMethod rm, const char *msg) {

  safeOut sout(cout);
  sysError sysErr;
  int errType = 0;

  if(msg)
    sout << "\n  " << msg;
  switch(rm) {
    case console:
      errType = WSAGetLastError();
      sout << "\n  initialization error type: " << errType;
      sout << "\n  " << sysErr.getLastMsg() << "\n\n";
      break;
    case window :
      sout << "\n  window reporting not implemented yet\n\n";
      break;
    case exception:
      throw sysErr.getLastMsg();
    default :
      sout << "\n  invalid report method\n\n";
  }
}

//

//----< constructor >------------------------------------------------

sockClt::sockClt() {

  // load version 1.1 of winsock dll

  WORD wVersionRequested = MAKEWORD(1,1);   // requesting version 1.1
  WSAData wData;                            // startup data filled
                                            // by WSAStartup

  int err = WSAStartup(wVersionRequested, &wData);

  if(err == SOCKET_ERROR) {
    _good = false;
    reportError(console,"starting WinSock on client");
  }

  // create TCP/IP socket object

  _client = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  _good = true;
}

//----< attempt to connect to a server >-----------------------------

bool sockClt::connectTo(address &addr, int port) {

  // connect to server on specified address and port
  //   - note the IP address 127.0.0.1 refers to host machine

  _port = port;        // server port

  struct hostent *serverHostent;
  serverHostent = gethostbyname("Lusitania");
  
  SOCKADDR_IN tcpAddr;
  tcpAddr.sin_family = AF_INET;                       // TCP/IP
  tcpAddr.sin_addr.s_addr = inet_addr(addr.c_str());  // host machine
  tcpAddr.sin_port = htons(port);                     // server port

  int err = ::connect(_client, (sockaddr*)&tcpAddr, sizeof(tcpAddr));

  if(err == SOCKET_ERROR) {
    reportError(console,"client can't connect");
    return (_good = false);
  }
  return true;
}

//

//----< attempt to connect to a server >-----------------------------

bool sockClt::connectTo(const char *s, int port) {

  return connectTo(string(s), port);
}

//----< destructor >-------------------------------------------------

sockClt::~sockClt() {

  closesocket(_client);
  WSACleanup();
}

//----< insertion operator >-----------------------------------------

sockClt& sockClt::operator<<(std::string &s) {

  static char *msg = 
    "can't send all bytes in sockClt::operator<<(std::string &s)";
  int bytesSent = send(s.c_str());
  if(bytesSent != s.size())
    reportError(console,msg);
  return *this;
}
    
//----< insertion operator >-----------------------------------------

sockClt& sockClt::operator<<(const char *s) {

  return operator<<(string(s));
}
    
//----< extraction operator >----------------------------------------

sockClt& sockClt::operator>>(std::string &s) {

  recv(s);
  return *this;
}

//

//----< send a block of bytes through socket connection >------------
//
//  attempts to guarantee that all bytes are sent
//
int sockClt::send(const char *block) {

  const char *msg = "sockClt::send failed";

  int blockLen = strlen(block);
  int bytesSent;
  int bytesLeft = blockLen;
  int blockIndx = 0;
  int count = 0;
  while(bytesLeft > 0) {
    bytesSent = ::send(_client,&block[blockIndx],bytesLeft,0);
    if(bytesSent == SOCKET_ERROR) {
      reportError(console,msg);
      _good = false;
      return blockIndx;
    }
    bytesLeft -= bytesSent;
    blockIndx += bytesSent;
  }
  return blockIndx;
}

//----< receive block of bytes through socket connection >-----------
//
//  attempts to guarantee that all bytes are read
//
int sockClt::recv(std::string &s, int numBytes) {

  const char *msg = "sockClt::recv failed";
  char block[MAX_MSG_SIZE];

  if(numBytes > MAX_MSG_SIZE-1)
    numBytes = MAX_MSG_SIZE-1;

  int bytesRecvd, bytesLeft=numBytes, blockIndx=0, count=0;
  while(bytesLeft > 0) {
    bytesRecvd = ::recv(_client,&block[blockIndx],bytesLeft,0);
    if(bytesRecvd == SOCKET_ERROR) {
      reportError(console,msg);
      _good = false;
      return blockIndx;
    }
    bytesLeft -= bytesRecvd;
    blockIndx += bytesRecvd;
    if(numBytes == MAX_MSG_SIZE-1)
      break;
  }
  block[blockIndx] = '\0';
  s = block;
  return blockIndx;
}

//

//----< client will stop receiving >---------------------------------

void sockClt::shutDownRecv() { 
  
  _good = false;
  shutdown(_client,SD_RECEIVE); 
}

//----< client will stop sending >-----------------------------------

void sockClt::shutDownSend() { 
  
  _good = false;
  shutdown(_client,SD_SEND); 
}

//----< client will stop sending and receiving >---------------------

void sockClt::shutDown() { 

  _good = false;
  shutdown(_client,SD_BOTH); 
}

//----< close socket connection >------------------------------------
//
//  need to call this after client calls shutDown to return
//  socket resources
//
void sockClt::close() { 
  
  _good = false;
  closesocket(_client); 
}

//

//----< constructor >------------------------------------------------

sockSvr::sockSvr(int port) : _listeningPort(port) {

  // load version 1.1 of winsock dll

  WORD wVersionRequested = MAKEWORD(1,1);   // requesting version 1.1
  WSAData wData;                            // startup data filled
                                            // by WSAStartup

  int err = WSAStartup(wVersionRequested, &wData);
  if(err == SOCKET_ERROR) {
    reportError(console,"starting WinSock on server");
    _good = false;
  }
  else
    _good = true;
}

//----< destructor >-------------------------------------------------

sockSvr::~sockSvr() {

  _good = false;
  closesocket(_listener);
  WSACleanup();
}

//

//----< send a block of bytes through socket connection >------------
//
//  attempts to guarantee that all bytes are sent
//
int sockSvr::send(SOCKET sock, const char *block) {

  static char *msg = "sockSvr::send failed";

  int blockLen = strlen(block);
  int bytesSent;
  int bytesLeft = blockLen;
  int blockIndx = 0;
  int count = 0;
  while(bytesLeft > 0) {
    bytesSent = ::send(sock,&block[blockIndx],bytesLeft,0);

    if(bytesSent == SOCKET_ERROR) {
      reportError(console,msg);
      return blockIndx;
    }
    bytesLeft -= bytesSent;
    blockIndx += bytesSent;
  }
  return blockIndx;
}

//----< receive block of bytes through socket connection >-----------
//
//  attempts to guarantee that all bytes are read
//
int sockSvr::recv(SOCKET sock, std::string &s, int numBytes) {

  const char *msg = "sockSvr::recv failed";
  char block[MAX_MSG_SIZE];

  if(numBytes > MAX_MSG_SIZE-1)
    numBytes = MAX_MSG_SIZE-1;

  int bytesRecvd, bytesLeft=numBytes, blockIndx=0, count=0;
  while(bytesLeft > 0) {

    // recv reads all available bytes up to bytesLeft.
    // If no data is available it blocks until data becomes
    // available or the connection closes.

    bytesRecvd = ::recv(sock,&block[blockIndx],bytesLeft,0);
    if(bytesRecvd == 0)
      break;
    if(bytesRecvd == SOCKET_ERROR) {
      reportError(console,msg);
      return blockIndx;
    }
    bytesLeft -= bytesRecvd;
    blockIndx += bytesRecvd;
    if(numBytes == MAX_MSG_SIZE-1)
      break;
  }
  block[blockIndx] = '\0';
  s = block;
  return blockIndx;
}

//

//----< start up server >--------------------------------------------

void sockSvr::start() {

  // create TCP/IP socket object to listen for connection requests

  _listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

  // bind listner socket to specific port and network interface

  SOCKADDR_IN tcpAddr;
  tcpAddr.sin_family = AF_INET;             // use TCP/IP protocol
  tcpAddr.sin_port = htons(_listeningPort); // listening port
  tcpAddr.sin_addr.s_addr = INADDR_ANY;     // listen over all NICs
  int err = bind(_listener, (SOCKADDR*)&tcpAddr, sizeof(tcpAddr));

  if(err == SOCKET_ERROR) {
    reportError(console,"failed to bind");
    _good = false;
    return;
  }

  // listen for incoming connection requests

  int backLog = 10;
  err = ::listen(_listener, backLog);

  if(err == SOCKET_ERROR) {
    reportError(console,"failed to start listening");
    _good = false;
    return;
  }

//

  int size = sizeof(tcpAddr);
  safeOut sout(cout);

  while(1) {

    sout << "\n  waiting for incoming request";

    // block until a connection is established

    SOCKET toClient = accept(_listener, (SOCKADDR*)&tcpAddr, &size);      

    if(toClient == INVALID_SOCKET) {
      reportError(console,"could not create socket for new client");
      break;
    }
    sout << locker 
         << "\n\n  established connection with: "
         << inet_ntoa(tcpAddr.sin_addr) << " : " 
         << ntohs(tcpAddr.sin_port) << "\n"
         << unlocker;

    // got connection so spawn thread to read transferred data

    unsigned int threadID;
    HANDLE hThread = 
      (HANDLE)_beginthreadex(
        NULL, 0, threadProc, (void*)toClient, 0, &threadID
      );
    CloseHandle(hThread);   // listener doesn't need to communicate
  }                         // with thread so close thread handle
}

//
//----< send queue thread processing >-------------------------------
//
//  Extracts messages from a send queue dedicated to a single client
//  and sends them back to the client via a duplex socket.
//
typedef lockingPtr< connectionMgr, GlobalMutex<connMgrLock> > connMgrlp;
typedef lockingPtr< mtqueue<message>, Mutex > sendQlp;

class svrSendProc : public processing {

  public:
    svrSendProc(connectionMgr &cm, SOCKET sock, int clientID) 
      : _cm(cm), _client(sock), _myID(clientID) { }
    
    ~svrSendProc() { }

    void run() {
      GlobalMutex<connMgrLock> cmlocker;
      mtqueue<message> &sendQ = connMgrlp(_cm,cmlocker)->sendQ(_myID);
      Mutex &sQlocker = connMgrlp(_cm,cmlocker)->locker(_myID);
      safeOut sout(cout);
      bool sentQuit = false;
      int lenSQ;

      do {  // look for message to send back to client
        
        lenSQ = sendQlp(sendQ,sQlocker)->length();
        if(lenSQ == 0) {
          ::Sleep(20);
          continue;
        }
        do {  // found message(s) in queue, so process

          message sndMsg = sendQlp(sendQ,sQlocker)->deQ();
          if(!sockSvr::send(_client,sndMsg.serialize().c_str())) {
            reportError(console,"\n  send error in server");
          }
          if(sndMsg.body() == "quit") {
            sout << locker << "\n  quitting: there are " << lenSQ-1 
                 << " messages in sendQ" << unlocker;
            sentQuit = true;
            break;
          }
          lenSQ = sendQlp(sendQ,sQlocker)->length();
        } while(lenSQ > 0);
      } while(!sentQuit);
      sout << "\n  server send thread shutting down";
    }
  private:
    connectionMgr &_cm;
    SOCKET _client;
    int _myID;
};

//
//----< receive queue thread processing >----------------------------
//
//  This thread extracts messages from a socket connected to a client
//  and posts them to a receive queue.  Note that all clients use this
//  same receive queue, and are served by this same processing function.
//
//  Queues are made available by a connectionMgr object created by this
//  function.  The connectionMgr is responsible for helping server to route
//  responses to messages drawn from recvQ back to sendQ for sending client.
//
unsigned int __stdcall sockSvr::threadProc(void *pArgs) {

  SOCKET sock = (SOCKET)pArgs;

  GlobalMutex<connMgrLock> cmlocker;
  GlobalMutex<svrRecvQlock> rQlocker;

  connectionMgr cm;
  int myID = connMgrlp(cm,cmlocker)->openConnection(sock);
  mtqueue<message> &recvQ = connMgrlp(cm,cmlocker)->recvQ();
  Mutex &sQlocker = connMgrlp(cm,cmlocker)->locker(myID);
  safeOut sout(cout);
  const int bufSize = MAX_MSG_SIZE;

  // start message sending thread

  svrSendProc proc(cm,sock,myID);
  thread send(proc);
  send.start();

  // start pulling messages out of socket and processing them
  
  int bytes;
  bool gotQuit = false;
  do {
    message rcvd;
    if(!gotQuit) {

      // read head of message from socket so we can tell
      // how much more there is to read

      string rcvStr, head, remainder;
      bytes = recv(sock,head,message::headSize);
      if(bytes == SOCKET_ERROR) {
        reportError(console,"\n  client socket shutting down");
        break;
      }
      // now read the rest of message

      int bytesLeft = message::bytesToRead(head);
      bytes = recv(sock,remainder,bytesLeft);
      rcvStr = head + remainder;
      rcvd = message(rcvStr);
      if(bytes == SOCKET_ERROR) {
        reportError(console,"\n  client socket shutting down");
        break;
      }
    }
//
    if(bytes > 0 && !gotQuit) {

      // we've got a valid message so process it
      // enqueue clientID message and recvd client message on recv queue

      message id(message::id,TtoStr<int>(myID).convert());
      rQlocker.lock();    // Make sure that these messages stay together
      recvQ.enQ(id);      // in recvQ by locking both enQ instructions.
      recvQ.enQ(rcvd);    // This prevents another server client thread
      rQlocker.unlock();  // from squeezing in with its messages.

      if(rcvd.body() == "quit")
        break;
    }
    else {
      ::Sleep(20);
    }
  } while(bytes != 0);

  send.wait();
  sout << "\n  server closing client thread\n";
  shutdown(sock,SD_SEND);
  connMgrlp(cm,cmlocker)->closeConnection(myID);
  if(closesocket(sock) != 0)
    reportError(console,"error closing client socket in server");
  return 0;
}

//
#ifdef TEST_SOCKS_SERVER

//----< test stub >--------------------------------------------------

///////////////////////////////////////////////////////////////
//  define processing for server's processing thread
//  - simply starts server
//
class serverProc : public processing {

  public:
    serverProc() { }
    virtual void run() {
      safeOut sout(cout);
      sout << "\n  running server\n";
      sockSvr server(1024);
      server.start();
    }
};

///////////////////////////////////////////////////////////////
//  define processing for server's main thread
//
//  Simply extracts messages from shared receive queue
//  and sends them back to the sending client.
//
//  Uses the services of a connection manager to figure out
//  how to contact the right client, using a clientID message
//  from server.
//
void main() {

  safeOut sout(cout);
  sout << "\n  starting server thread\n";

  GlobalMutex<connMgrLock> cmlocker;
  GlobalMutex<recvQlock> rQlocker;
  typedef lockingPtr< connectionMgr, GlobalMutex<connMgrLock> > connMgrlp;
  typedef lockingPtr< mtqueue<message>, GlobalMutex<recvQlock> > recvQlp;
  typedef lockingPtr< mtqueue<message>, Mutex > sendQlp;

  serverProc proc;
  thread serverThrd(proc);
  serverThrd.start();

  connectionMgr cm;
  int clientID = 0;
  int numMsg = 0;
  while(1) {
    int len = connMgrlp(cm,cmlocker)->recvQ().length();
    if(numMsg != len) {
      sout << locker << "\n  there are " << len 
           << " messages in recv queue" << unlocker;
      numMsg = len;
    }
    if(len == 0) {
      ::Sleep(50);
      continue;
    }
//
    while(len > 0) {

      // server assumes that all commands set the client id

      message fromClient = connMgrlp(cm,cmlocker)->recvQ().deQ();
      if(fromClient.type() == message::id) {
        clientID = StrToT<int>(fromClient.body()).convert();
      }
      else {

        // Not a command so process as a message.
        // This demo simply loops message back to client's sendQ

        mtqueue<message> &sendQ = connMgrlp(cm,cmlocker)->sendQ(clientID);
        Mutex &sQlocker = connMgrlp(cm,cmlocker)->locker(clientID);
        sendQlp(sendQ,sQlocker)->enQ(fromClient);

        sout << locker << "\n  sending back: " 
             << fromClient.serialize() << unlocker;
      }
      len = connMgrlp(cm,cmlocker)->recvQ().length();
    }
  }
  serverThrd.wait();
}

#endif

//

#ifdef TEST_SOCKS_CLIENT
#include "thrdsWithProc.h"
#include "sysUtils.h"
#include "msgHnd.h"

///////////////////////////////////////////////////////////////
//  define processing for client's receive thread
//
//  simply reads and announces anything coming back from server
//
class clientRcvrProc : public processing {

  public:
    clientRcvrProc(sockClt &client) : _client(client) { }
    virtual void run() {
      string rcvMsg;
      safeOut sout(cout);
      do {
        string rcvMsg, head, remainder;
        _client.recv(head,message::headSize);
        int bytesLeft = message::bytesToRead(head);
        _client.recv(remainder,bytesLeft);
        rcvMsg = head + remainder;
        message recvd(rcvMsg);
        if(_client.good())
          sout << locker << "\n  message from server: " 
               << recvd.body() << unlocker;
        if(recvd.body() == "quit")
          break;
      } while(_client.good());
      sout << "\n  client receiver thread shutting down";
    }

  private:
    sockClt &_client;
};

//

///////////////////////////////////////////////////////////////
//  start running the client

void main(int argc, char *argv[]) {

  safeOut sout(cout);
  sout << "\n  running client\n";

  sockClt client;
  if(!client.connectTo("127.0.0.1", 1024)) {
    reportError(console,"client could not connect");
    return;
  }
  else
    sout << "\n  connected to server";

  clientRcvrProc proc(client);
  thread rcvThrd(proc);
  rcvThrd.start();

  int i;
  message::msgType type = message::text;
  for(i=0; i<50; i++) {
    string body = string("message #") += TtoStr<int>(i).convert();
    message sndMsg(type,body);
    sout << locker << "\n  sending to server:   " 
         << sndMsg.serialize() << unlocker;
    client << sndMsg.serialize();
  }

  message::msgType endType = message::command;
  message endMsg(type,"quit");
  sout << locker << "\n  sending to server:   " 
       << endMsg.serialize() << unlocker;
  client << endMsg.serialize();

  rcvThrd.wait();

  sout << "\n  client main thread shutting down";
  client.shutDown();
  client.close();
  sout << "\n\n";
}

#endif