/////////////////////////////////////////////////////////////////////
// MT1Q2.cpp - posting received messages in Core TestHarness       //
//                                                                 //
// Jim Fawcett, CSE687 - OOD : Midterm Exam, Fall 2018             //
/////////////////////////////////////////////////////////////////////
/*
*  - Solutions for both MT1Q2 and MT2Q3
*/
#include <iostream>
#include <conio.h>
#include <thread>
#include <vector>
#include "../CppCommWithFileXfer/Message/Message.h"
#include "../CppCommWithFileXfer/MsgPassingComm/Comm.h"
#include "../CppCommWithFileXfer/Cpp11-BlockingQueue/Cpp11-BlockingQueue.h"

using namespace MsgPassingCommunication;

namespace MTF18
{
  ///////////////////////////////////////////////////////////////////
  // Client class
  // - sends a few test request messages to TestHarness

  class Client
  {
  public:
    Client(EndPoint from, EndPoint to);
    ~Client();
    void start();
    void wait();
    void sendMessages();
  private:
    std::thread recv;
    Comm comm_;
    EndPoint to_;
    EndPoint from_;
  };
  //----< initializes endpoints and starts comm >--------------------

  Client::Client(EndPoint from, EndPoint to) : comm_(from, "sender"), to_(to), from_(from)
  {
    comm_.start();
  }
  //----< stops comm and detaches thread if needed >-----------------

  Client::~Client()
  {
    std::cout << "\n  Client Comm shutting down";
    comm_.stop();
    if (recv.joinable())
      recv.detach();
  }
  //----< creates thread to send test request messages >-------------

  void Client::start()
  {
    std::cout << "\n  starting Client's send thread listening on " + from_.toString();
    recv = std::thread([&]() { sendMessages(); });
  }
  //----< waits for all messages to be sent >------------------------

  void Client::wait()
  {
    if (recv.joinable())
      recv.join();
  }
  //----< send request messages >------------------------------------
  /*
  *  - send thread will shut down when this method completes
  *  - sleeps just for display so everything doesn't happen at once
  */
  void Client::sendMessages()
  {
    Message testReq;
    testReq.to(to_);
    testReq.from(from_);
    testReq.command("testReq");
    std::cout << "\n  Client sending msg: " + testReq.command();
    comm_.postMessage(testReq);
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    std::cout << "\n  Client sending msg: " + testReq.command();
    comm_.postMessage(testReq);
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    std::cout << "\n  Client sending msg: " + testReq.command();
    comm_.postMessage(testReq);
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
  }
  ///////////////////////////////////////////////////////////////////
  // ChildTester class
  // - pretends to execute tests when sent a request message
  // - sends a ready message on startup and at the end of each 
  //   mock test

  class ChildTester
  {
  public:
    ChildTester(EndPoint from, EndPoint to);
    ~ChildTester();
    void start();
    void wait();
    void doTests();
  private:
    std::thread recv;       // processing thread
    Comm comm_;
    EndPoint to_;           // TestHarness address
    EndPoint from_;         // my address
                            // Client address in msg.from()
  };
  //----< initialize endpoints and create comm >---------------------

  ChildTester::ChildTester(EndPoint from, EndPoint to) 
    : comm_(from, "tester"), from_(from), to_(to) {}

  //----< stop comm >------------------------------------------------

  ChildTester::~ChildTester()
  {
    std::cout << "\n      ChildTester destroyed";
    comm_.stop();
  }
  //----< start comm and testing thread >----------------------------

  void ChildTester::start()
  {
    comm_.start();
    std::cout << "\n      starting ChildTester thread listening on " + from_.toString();
    recv = std::thread([&]() { doTests(); });
    recv.detach();
  }
  //----< deQ request message and do mock test >---------------------
  /*
  *  - sends ready message when starting and when done with each test
  */
  void ChildTester::doTests()
  {
    Message ready;
    ready.from(from_);
    ready.to(to_);
    ready.command("ready");
    std::cout << "\n      Tester sending msg: " + ready.command();
    comm_.postMessage(ready);

    while (true)
    {
      Message msg = comm_.getMessage();
      std::cout << "\n      Tester " + from_.toString() + " received msg: " + msg.command();

      if (msg.command() == "done")
      {
        msg.from(from_);
        comm_.postMessage(msg);
        std::cout << "\n      ChildTester shutting down";
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        break;
      }
      // mock test
      std::cout << "\n      --- testing --- ";
      std::this_thread::sleep_for(std::chrono::milliseconds(1000));
      std::cout << "\n      Tester sending msg: " + ready.command();
      comm_.postMessage(ready);
    }
  }
  ///////////////////////////////////////////////////////////////////
  // TestHarness class
  // - manages thread pool's message flow
  // - you will replace child threads with child processes

  class TestHarness
  {
  public:
    TestHarness(EndPoint from, EndPoint to);
    ~TestHarness();
    void start();
    void wait();
    void stop();
    void recvMessages();
    void dispatchMessages();
  private:
    BlockingQueue<Message> readyQ_;
    BlockingQueue<Message> requestQ_;
    std::thread recvr;
    std::thread dspat;
    Comm comm_;
    EndPoint to_;
    EndPoint from_;
    std::vector<ChildTester*> children_;
  };
  //----< starts comm and creates child threads >--------------------
  /*
  *  - you will create child processes instead of threads
  */
  TestHarness::TestHarness(EndPoint from, EndPoint to) 
    : comm_(from, "recvr"), to_(to), from_(from)
  {
    comm_.start();

    EndPoint ep1("localhost", 8081);  // testHarness
    EndPoint ep3("localhost", 8083);  // child tester
    EndPoint ep4("localhost", 8084);  // child tester

    ChildTester* pChild1 = new ChildTester(ep3, from);
    children_.push_back(pChild1);

    ChildTester* pChild2 = new ChildTester(ep4, from);
    children_.push_back(pChild2);
  }
  //----< deletes all child threads >--------------------------------

  TestHarness::~TestHarness()
  {
    std::cout << "\n    TestHarness Comm shutting down";
    std::cout << "\n    requestQ size = " + Utilities::Converter<size_t>::toString(requestQ_.size());
    std::cout << "\n    readyQ size   = " + Utilities::Converter<size_t>::toString(readyQ_.size());
    for (auto pChild : children_)
    {
      delete pChild;
    }
    comm_.stop();
  }
  //----< starts recvr, dspat, and child threads >-------------------

  void TestHarness::start()
  {
    ///////////////////////////////////////////////////////////////////
    // MT1Q2: Part B of answer

    std::cout << "\n    starting TestHarness receive thread listening on " << from_.toString();
    recvr = std::thread([&]() { recvMessages(); });
    recvr.detach();

    // MT1Q2: End of Part B of answer
    ///////////////////////////////////////////////////////////////////
    // MT2Q3: Part B of answer

    std::cout << "\n    starting TestHarness dispatch thread";
    dspat = std::thread([&]() { dispatchMessages(); });
    dspat.detach();

    // MT2Q3: End of Part B of answer
    ///////////////////////////////////////////////////////////////////
    
    for (auto pChild : children_)
    {
      pChild->start();
    }
  }
  //----< sends stop message to itself >-----------------------------
  /*
  *  - this is a place-holder for a more intelligent shutdown
  *  - not used in this demo
  */
  void TestHarness::stop()
  {
    Message stop;
    stop.command("TestHarnessStop");
    stop.to(from_);  // send to self
    stop.from(from_);
    comm_.postMessage(stop);
  }
  //----< waits for TestHarness to shutdown >------------------------
  /*
  *  - this is a place-holder for a more intellegent shutdown
  *  - not used in this demo
  */
  void TestHarness::wait()
  {
    if (recvr.joinable())
      recvr.join();
    if (dspat.joinable())
      dspat.join();
  }
  //----< receive and post messages to appropriate queues >----------

  ///////////////////////////////////////////////////////////////////
  // MT1Q2: Part A of answer

  void TestHarness::recvMessages()
  {
    while (true)
    {
      Message msg = comm_.getMessage();
      std::cout << "\n    TestHnRcv received msg: " + msg.command();
      if (msg.command() == "ready")
      {
        std::cout << "\n    TestHnRcv posting msg: " + msg.command();
        readyQ_.enQ(msg);
      }
      else if (msg.command() == "testReq")
      {
        std::cout << "\n    TestHnRcv posting msg: " + msg.command();
        requestQ_.enQ(msg);
      }
      else
      {
        std::cout << "\n    TestHarnessR - invalid message";
      }
    }
  }
  // MT1Q2: End of Part A of answer
  ///////////////////////////////////////////////////////////////////

  //----< deQ messages and send to testers >-------------------------

  ///////////////////////////////////////////////////////////////////
  // MT2Q3: Part A of answer

  void TestHarness::dispatchMessages()
  {
    while (true)
    {
      Message trMsg = requestQ_.deQ();
      std::cout << "\n    TestHnDsp deQ msg: " + trMsg.command();

      Message rdyMsg = readyQ_.deQ();
      std::cout << "\n    TestHnDsp deQ msg: " + rdyMsg.command();

      trMsg.to(rdyMsg.from());  // send request to ready child
      std::cout << "\n    TestHnDsp sending msg: " + trMsg.command();
      comm_.postMessage(trMsg);
    }
  }
  // MT2Q3: End of Part A of answer
  ///////////////////////////////////////////////////////////////////
}

using namespace MTF18;

int main()
{
  Utilities::Cosmetic cos;
  Utilities::Title("MT1Q2, MT2Q3 - posting incoming messages and sending to testers");

  EndPoint ep1("localhost", 8081);  // testHarness
  EndPoint ep2("localhost", 8082);  // client
  EndPoint ep3("localhost", 8083);  // child tester
  EndPoint ep4("localhost", 8084);  // child tester

  {
    TestHarness testHarness(ep1, ep3);
    testHarness.start();

    Client client(ep2, ep1);
    client.start();

    client.wait();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    testHarness.stop();
    Utilities::waitForKey();
  }
}