/////////////////////////////////////////////////////////////////////////
// MessageMgr.cpp - Provides asynchronous message-passing IPC          //
// ver 1.2                                                             //
// Language:    Visual C++ 2008                                        //
// Platform:    Dell Precision T7400, Vista Ultimate, SP1              //
// Application: Help for CSE687 - Object Oriented Design, Project #2   //
// Author:      Jim Fawcett, CST 4-187, Syracuse University            //
//              (315) 443-3948, jfawcett@twcny.rr.com                  //
/////////////////////////////////////////////////////////////////////////

#include "MessageMgr.h"
#include "..\MMFmgr\MMFmgr.h"
#include "..\Threads\Threads.h"
#include "..\Threads\locks.h"

/////////////////////////////////////////////////////////////////////
// Thread Proc for Sender thread

class SenderThreadProc : public Thread_Processing
{
public:
  SenderThreadProc(BQueue<std::string>* pSQ, MMFmgr* pMMF) : pSQ_(pSQ), pMMF_(pMMF) {}
  SenderThreadProc* clone() const 
  { 
    return new SenderThreadProc(*this); 
  }
  void run()
  {
    while(true)
    {
      std::string msg = pSQ_->deQ();
      pMMF_->putString(msg);
      if(msg == "quit")
        break;
    }
  }
private:
  BQueue<std::string>* pSQ_;
  MMFmgr* pMMF_;
};
/////////////////////////////////////////////////////////////////////
// Thread Proc for Receiver thread

class RecverThreadProc : public Thread_Processing
{
public:
  RecverThreadProc(BQueue<std::string>* pRQ, MMFmgr* pMMF) : pRQ_(pRQ), pMMF_(pMMF) {}
  RecverThreadProc* clone() const 
  { 
    return new RecverThreadProc(*this); 
  }
  void run()
  {
    while(true)
    {
      std::string msg = pMMF_->getString();
      pRQ_->enQ(msg);
      if(msg == "quit")
        break;
    }
  }
private:
  BQueue<std::string>* pRQ_;
  MMFmgr* pMMF_;
};
//----< constructor used by both client and server >---------------------

MsgMgr::MsgMgr(MMFmgr::role r) : role_(r)
{
  // create read and write queues
  
  pOutQ = new BQueue<std::string>();
  pInQ = new BQueue<std::string>();

  // create two memory-mapped files:
  // - client MMF accepts inputs from client and supplies
  //   outputs to server
  // - server MMF accepts inputs from server and supplies
  //   outputs to client

  if(r == MMFmgr::client)
  {
    pOutMMF = new MMFmgr("client",4098,MMFmgr::client);
    pInMMF = new MMFmgr("server",4098,MMFmgr::server);
  }
  else if(r == MMFmgr::server)
  {
    pOutMMF = new MMFmgr("server",4098,MMFmgr::server);
    pInMMF = new MMFmgr("client",4098,MMFmgr::client);
  }
  else
  {
    throw std::exception("unknown role in MMFmgr ctor");
  }

  // Send thread reads from input queue and writes to MMF

  Thread<terminating_thread>* pSendThrd 
    = new Thread<terminating_thread>(*new SenderThreadProc(pOutQ,pOutMMF));
  sout << "\n  starting SenderThread";
  pSendThrd->start();

  // Recv thread reads from MMF and writes to output queue

  Thread<terminating_thread>* pRecvThrd 
    = new Thread<terminating_thread>(*new RecverThreadProc(pInQ,pInMMF));
  sout << "\n  starting RecverThread";
  pRecvThrd->start();

  // This event, used in conjunction with MessageMgr::wait() allows both
  // client and server to wait for the other to be created before starting

  hStarted_ = CreateEventA(0,TRUE,FALSE,"started");
  if(hStarted_ == NULL)
    throw std::exception("\n\ncan't create started event\n\n");
  else 
  {
    if(GetLastError() == ERROR_ALREADY_EXISTS)
    {
      SetEvent(hStarted_);  // make signaled so waiter will awake
    }
    else
    {
      ResetEvent(hStarted_); // make unsignaled so waiter will block
    }
  }
}

MsgMgr::~MsgMgr()
{
  delete pInQ;
  delete pOutQ;
  delete pInMMF;
  delete pOutMMF;
  CloseHandle(hStarted_);
}
//----< wait for companion process to start >----------------------------

void MsgMgr::wait() 
{ 
  WaitForSingleObject(hStarted_,INFINITE);
}
//----< enqueue message awaiting send thread >===------------------------

void MsgMgr::putMessage(const std::string& msg)
{
  pOutQ->enQ(msg);
}
//----< dequeue message enqueued by recv thread >------------------------

std::string MsgMgr::getMessage()
{
  std::string temp = pInQ->deQ();
  return temp;
}

#include <iostream>
#include <sstream>

template <typename T>
std::string ToString(T t)
{
  std::ostringstream out;
  out << t;
  return out.str();
}
//----< Test stub >------------------------------------------------------

#ifdef TEST_MESSAGEMGR

class threadProc : public Thread_Processing
{
public:
  threadProc(MsgMgr* pMsgMgr) : pMsgMgr_(pMsgMgr) {}
  threadProc* clone() const { return new threadProc(*this); }
  void run()
  {
    std::string msg;
    for(int i=0; i<20; ++i)
    {
      msg = "msg#";
      msg += ToString(i);
      msg += " - through MMF";
      pMsgMgr_->putMessage(msg);
      sout << locker << "\n  sending " << msg << unlocker;
      ::Sleep(10);
    }
    msg = "quit";
    pMsgMgr_->putMessage(msg);
  }
private:
  MsgMgr* pMsgMgr_;
};

void main()
{
  try
  {
    MsgMgr client(MMFmgr::client);
    MsgMgr server(MMFmgr::server);
    client.wait();
    server.wait();
    client.putMessage("1st ");
    sout << locker << "\n  " << server.getMessage().c_str() << "\n" << unlocker;


    threadProc tp(&server);
    Thread<default_thread> thrd(tp);
    thrd.start();
    std::string msg = "";
    while(msg != "quit")
    {
      msg = client.getMessage();
      sout << locker << "\n  " << msg.c_str() << " received" << unlocker;
    }
    sout << "\n\n";
    thrd.wait();
    sout << "\n  Press key to end process";
    sout.flush();
    std::cin.get();
  }
  catch(std::exception& ex)
  {
    sout << ex.what();
  }
}

#endif