CCats/cli/src/ioman.cpp

#include "../include/ioman.h"
#include "../include/global.h"

#include <boost/asio.hpp>
#include <iostream>

#include <boost/algorithm/string.hpp>
#include <string>
#include <vector>

#include <readline/history.h>
#include <readline/readline.h>

#include <poll.h>

using std::string;
using std::vector;

using boost::asio::buffer;
using boost::asio::ip::address;
using boost::asio::ip::tcp;

/* this will be provided by main.cpp for the readline callback */
extern IoMan *gIOMAN;

void ioman_externalDebugPrint(string msg) {
  gIOMAN->printMessage(msg, gIOMAN->OutMsgType::debug);
}

IoMan::IoMan(char *ipcstring) : cmdman(fileman, &ioman_externalDebugPrint) {
  ipstring = std::string(ipcstring);
  port = 1234;
  tcpsock = new tcp::socket(ios);
  connected = false;

  /* to be put elsewhere */

  /* setup json stuff */
  Json::CharReaderBuilder rbuilder;
  wbuilder.settings_["indentation"] = "";
  reader = rbuilder.newCharReader();
  runnetwork = false;
  runinput = false;
  runresponse = false;
  versionstatus = off;
  loginstatus = off;
}

IoMan::~IoMan() {
  initcv.notify_all();
  if (runnetwork) {
    networkmutex.lock();
    runnetwork = false;
    networkmutex.unlock();
    tnetwork.join();
  }
  if (runinput) {
    inputmutex.lock();
    runinput = false;
    inputmutex.unlock();
    localcv.notify_all();
    tinput.join();
  }
  if (runresponse) {
    responsemutex.lock();
    runresponse = false;
    responsemutex.unlock();
    netcv.notify_all();
    tresponse.join();
  }

  if (connected) {
    disconnect();
  }

  delete tcpsock;
  delete reader;
}

void IoMan::printMessage(string nouse, OutMsgType nouse2) {}

vector<string> IoMan::tokenizeInput(string in) {
  size_t prev, index, quot;
  vector<string> args;
  /* tokenize string into command and arguments vector*/
  if ((index = in.find(" ")) == string::npos) {
    // only command no args
    args.push_back(in);
  } else {
    args.push_back(in.substr(0, index));
    index++;

    bool end_tokenizing = false;
    while (!end_tokenizing) {
      // find first char thats not a space
      while (in[index] == ' ') {
        index++;

        // bounds check
        if (index == in.size())
          end_tokenizing = true;
      }
      if (end_tokenizing)
        break;

      in = in.substr(index);

      if (in[0] == '\"') {
        // quoted string
        in = in.substr(1);
        index = in.find("\"");
        args.push_back(in.substr(0, index));
        index++;

        /*
        tokens.push_back(in.substr(0, ++index));
        */

        // char after closing quote should be space while within bounds
        if (index == in.size())
          end_tokenizing = true;
      } else {
        // non-quoted string
        index = in.find(" ");
        if (index == string::npos) { // no spaces, last arg
          args.push_back(in);
          end_tokenizing = true;
        } else {
          args.push_back(in.substr(0, index));
        }
      }
    }
  }
  return args;
}

bool IoMan::connect() {
  tcp::endpoint *ep = NULL;
  address addr;
  Json::Value root;
  root["command"] = "connect";
  root["address"] = ipstring;
  root["port"] = port;
  addr = address::from_string(ipstring, errcode);
  if (errcode) {
    root["error"] = errcode.message();
    connected = false;
  } else {
    // establish connection
    printMessage("IoMan::connect() connecting to " + ipstring, debug);
    ep = new tcp::endpoint(addr, port);
    tcpsock->connect(*ep, errcode);
    if (errcode) {
      root["error"] = errcode.message();
      connected = false;
    } else {
      connected = true;
      root["error"] = "";
    }
    delete ep;
  }
  root["accept"] = connected;
  printMessage(Json::writeString(wbuilder, root), normal);
  return connected;
}

void IoMan::disconnect() {
  printMessage("IoMan::disconnect()", debug);
  tcpsock->close();
  connected = false;
}

bool IoMan::init() {
  CmdMan::CmdRet ret;
  string work;
  Json::Value root;
  std::unique_lock<std::mutex> ulock;

  printMessage("IoMan::Init() begin", debug);

  if (!connect())
    return false;

  printMessage("IoMan::Init() versioncheck", debug);

  localmutex.lock();
  printMessage(string(__PRETTY_FUNCTION__) + string(" get localmutex"), debug);
  localinput.push_back("version " + protocolVersion);
  localmutex.unlock();
  printMessage(string(__PRETTY_FUNCTION__) + string(" release localmutex"),
               debug);
  localcv.notify_all();

  printMessage("IoMan::Init() begin", debug);

  runnetwork = true;
  runinput = true;
  runresponse = true;

  tnetwork = std::thread(&IoMan::networkMain, this);
  tinput = std::thread(&IoMan::inputMain, this);
  tresponse = std::thread(&IoMan::responseMain, this);

  ulock = std::unique_lock<std::mutex>(initmutex);
  while (versionstatus == off) {
    initcv.wait(ulock);
  }
  if (versionstatus == err) {
    runnetwork = false;
    runinput = false;
    runresponse = false;
    return false;
  }
  initmutex.unlock();
  initcv.notify_all();

  printWelcomeMessage();
  return true;
}

/* loop to fetch data from the network, doing light preprocessing on it to be
 * handled by responseMain */
void IoMan::networkMain() {
  vector<Json::Value> toput;
  char *recvjson;
  Json::Value root;
  unsigned int jsonsize, readsize;

  printMessage("IoMan::networkMain() begin", debug);
  networkmutex.lock();
  while (runnetwork) {
    networkmutex.unlock();
    /*
    read from network until \n
    try to parse json
    - output error if not ok
    store all ok json in local vector

    get networkmutex
    put all local jsons into network vector
    release networkmutex
    */

    // read from network
    readsize = boost::asio::read_until(*tcpsock, recvbuf, '\n', errcode);
    printMessage(string(__PRETTY_FUNCTION__) + string(" asio::read() ok ") +
                     std::to_string(readsize),
                 debug);
    // printMessage(string("have ") + std::to_string(toprocess.size()) +
    // string(" commands"), debug);
    if (errcode && errcode != boost::asio::error::eof) {
      printMessage("IoMan::networkMain() couldnt read json data\n" +
                       errcode.message(),
                   error);
      continue;
    }
    if (!readsize)
      break;
    recvjson = (char *)(boost::asio::buffer_cast<const char *>(recvbuf.data()));
    recvjson[readsize] = 0;
    while (strchr(recvjson, '\n')) {
      // parse
      jsonsize = strchr(recvjson, '\n') - recvjson + 1;

      printMessage(string(__PRETTY_FUNCTION__) + string(" found jsondata ") +
                       string(recvjson),
                   debug);

      if (!reader->parse(recvjson, recvjson + jsonsize, &root, &jsonerror)) {
        printMessage("IoMan::networkMain() couldnt parse json data: " +
                         jsonerror,
                     error);
        recvbuf.consume(jsonsize);
        continue;
      }
      recvbuf.consume(jsonsize);
      readsize -= jsonsize;

      printMessage(
          string(__PRETTY_FUNCTION__) + string(" remaining recvbuf ") +
              string(boost::asio::buffer_cast<const char *>(recvbuf.data())),
          debug);

      for (int i = 0; i < jsonsize; i++)
        recvjson++;
      // store locally
      toput.push_back(root);
    }

    if (toput.size()) {
      // put into global vector
      netmutex.lock();
      printMessage(string(__PRETTY_FUNCTION__) + string(" get netmutex"),
                   debug);
      netinput.insert(netinput.end(), toput.begin(), toput.end());
      netmutex.unlock();
      printMessage(string(__PRETTY_FUNCTION__) + string(" release netmutex"),
                   debug);
    }
    netcv.notify_all();

    // clean up local stuff
    toput = vector<Json::Value>();
    recvbuf.consume(readsize);
    networkmutex.lock();
  }
}

/* loop to handle input from the user and responseMain, sending data via network
 * if required */
void IoMan::inputMain() {
  vector<string> toprocess;
  string command;
  vector<string> args;
  CmdMan::CmdRet cmdret;
  std::unique_lock<std::mutex> ulock;

  printMessage("IoMan::inputMain() begin", debug);
  inputmutex.lock();
  while (runinput) {
    inputmutex.unlock();
    /*
    get inputmutex
    read all input vector into local vector
    release inputmutex

    process inputs
    send to server if required
    */

    // read into local vector
    ulock = std::unique_lock<std::mutex>(localmutex);
    while (!localinput.size() && runinput) {
      localcv.wait(ulock);
    }
    printMessage(string(__PRETTY_FUNCTION__) + string(" has localmutex"),
                 debug);
    toprocess = vector<string>(localinput);
    localinput = vector<string>();
    localmutex.unlock();
    printMessage(string(__PRETTY_FUNCTION__) + string(" release localmutex"),
                 debug);
    localcv.notify_all();

    if (!runinput)
      return;

    // printMessage(string("have ") + std::to_string(toprocess.size()) +
    // string(" commands"), debug);

    // process
    for (string cmd : toprocess) {
      args = tokenizeInput(cmd);
      command = args.front();
      args.erase(args.begin());
      cmdret = cmdman.execute(command, args);

      // determine wether to send something and do so if required
      if (cmdret.type & CmdMan::rettype::print) {
        printMessage(Json::writeString(wbuilder, cmdret.msg), normal);
      }
      if (cmdret.type & CmdMan::rettype::send) {
        printMessage("IoMan::inputMain() sending json \"" +
                         Json::writeString(wbuilder, cmdret.msg) + "\"",
                     debug);
        boost::asio::write(
            *tcpsock, buffer(Json::writeString(wbuilder, cmdret.msg) + "\n"),
            errcode);
        if (errcode) {
          printMessage("IoMan::inputMain() couldnt send json data\n" +
                           errcode.message() + "\n",
                       error);
          continue;
        }
      }
      if (cmdret.type & CmdMan::rettype::error) {
        printMessage(Json::writeString(wbuilder, cmdret.msg), error);
      }
      if(cmdret.type & CmdMan::rettype::close) {
        /* TODO i dunno */
        mainmutex.lock();
        runmain = false;
        mainmutex.unlock();
      }
    }

    // clean up local stuff
    toprocess = vector<string>();
    inputmutex.lock();
  }
}

/* loop to handle responses that have been fetched by netMain and possibly add
 * new commands to be handled by inputMain */
void IoMan::responseMain() {
  vector<Json::Value> toprocess;
  vector<string> toput;
  CmdMan::CmdRet cmdret;
  std::unique_lock<std::mutex> ulock;

  printMessage("IoMan::responseMain() begin", debug);
  responsemutex.lock();
  while (runresponse) {
    responsemutex.unlock();
    /*
    get networkmutex
    read all network vector into local vector
    release networkmutex

    process all jsons
    process putdata
    process getdata
    process listdata

    get inputmutex
    place new commands into input vector
    release inputmutex
    */

    // read into local vector
    ulock = std::unique_lock<std::mutex>(netmutex);
    while (!netinput.size() && runresponse) {
      netcv.wait(ulock);
    }
    printMessage(string(__PRETTY_FUNCTION__) + string(" get netmutex"), debug);
    toprocess = vector<Json::Value>(netinput);
    netinput = vector<Json::Value>();
    netmutex.unlock();
    printMessage(string(__PRETTY_FUNCTION__) + string(" release netmutex"),
                 debug);
    netcv.notify_all();

    if (!runresponse)
      return;

    // process jsons
    for (Json::Value root : toprocess) {
      cmdret = cmdman.handle(root);
      if (cmdret.type & CmdMan::rettype::close) {
        /* TODO i dunno */
        mainmutex.lock();
        runmain = false;
        mainmutex.unlock();
      }
      if (cmdret.type & CmdMan::rettype::error) {
        initmutex.lock();
        if (versionstatus == off)
          versionstatus = err;
        else if (loginstatus == off)
          loginstatus = err;
        initmutex.unlock();
        initcv.notify_all();
        printMessage(Json::writeString(wbuilder, cmdret.msg), error);
      }
      if (cmdret.type & CmdMan::rettype::seton) {
        initmutex.lock();
        if (versionstatus == off)
          versionstatus = on;
        else
          loginstatus = on;
        initmutex.unlock();
        initcv.notify_all();
      }
      if (cmdret.type & CmdMan::rettype::print) {
        printMessage(Json::writeString(wbuilder, cmdret.msg), normal);
      }
      if (cmdret.type & CmdMan::rettype::send) {
        if (cmdret.nextcommand.size()) {
          toput.push_back(cmdret.nextcommand);
        }
      }
    }

    if (toput.size()) {
      // put new commands into global vector
      localmutex.lock();
      printMessage(string(__PRETTY_FUNCTION__) + string(" get localmutex"),
                   debug);
      localinput.insert(localinput.end(), toput.begin(), toput.end());
      localmutex.unlock();
      printMessage(string(__PRETTY_FUNCTION__) + string(" release localmutex"),
                   debug);
    }
    localcv.notify_all();

    // clean up local stuff
    toprocess = vector<Json::Value>();
    toput = vector<string>();
    responsemutex.lock();
  }
}

/* this is the handler that readlines alternative interface will use to process
 * user input */
void ioman_readlineHandler(char *line) {
  vector<string> tokens;
  if (!line) {
    printf("\nNULLBURGER\n");
    gIOMAN->mainmutex.lock();
    gIOMAN->runmain = false;
    gIOMAN->mainmutex.unlock();
  } else {
    // split input line into tokens
    boost::algorithm::split(tokens, std::string(line),
                            boost::algorithm::is_any_of(" "),
                            boost::algorithm::token_compress_on);
    if (strlen(line) && tokens.size()) {
      add_history(line);

      gIOMAN->localmutex.lock();
      gIOMAN->printMessage(string(__PRETTY_FUNCTION__) +
                               string(" get localmutex"),
                           gIOMAN->debug);
      gIOMAN->localinput.push_back(line);
      gIOMAN->localmutex.unlock();
      gIOMAN->printMessage(string(__PRETTY_FUNCTION__) +
                               string(" release localmutex"),
                           gIOMAN->debug);
      gIOMAN->localcv.notify_all();
    }
    free(line);
  }
}

/* main user input loop */
void IoMan::run() {
  printMessage("IoMan::run() begin", debug);
  struct pollfd inPipeStatus;
  inPipeStatus.fd = STDIN_FILENO;
  inPipeStatus.events = POLLIN;

  runmain = true;

  // Install readline handler
  rl_callback_handler_install(getCmdPrompt().c_str(),
                              (rl_vcpfunc_t *)&ioman_readlineHandler);

  mainmutex.lock();
  while (runmain) {
    mainmutex.unlock();

    poll(&inPipeStatus, 1, 100);

    if (inPipeStatus.revents & POLLIN) {
      rl_callback_read_char();
    }
    if (!connected)
      break;
    if(loginstatus == err) break;
    mainmutex.lock();
  }
  mainmutex.unlock();

  // Remove the handler
  rl_callback_handler_remove();
}