DataClientProxy.cpp

Go to the documentation of this file.

#include "DataClientProxy.hpp"
#include "ServiceMessages.hpp"
#include "DRCEResourceUsage.hpp"
#include "DRCEResourceExtractor.hpp"

namespace HCE {
  namespace handlers {

    DataClientProxy::DataClientProxy(std::string name, zmq::context_t& ctx, const std::string& clientIdentity, std::string connectionString, int64_t startedAt,
        unsigned char logPriority, const std::string& iniFile) :
        Handler(name, ctx, clientIdentity, connectionString, startedAt, logPriority, iniFile), DataNodeOptions(Handler::handlerProperties),
        _shiftCounter(), _clientsQueue(){

      setIniFile(iniFile),
      initialize();
    }

    DataClientProxy::~DataClientProxy(){
      deinitialize();
    }

    int DataClientProxy::initialize(void){
      logger.log(NODE_LOG_MODE_INFORMATION) << "Initialize client [" << getClientIdentity() << "] with connection [" << getConnectionString() << "]" << flush;

      _inprocProxySock = nullptr;
      _inprocReducerOutSock = nullptr;
      _dataClientSock = nullptr;
      _dataClientSockConnected = false;

      initNode(getIniFile(), logger);

      setHeartbeatAt(s_clock() + getHeartbeatDelay());
      setHeartbeatedAt(s_clock());

      if (!statProperties.isExist(HCE::handlers::properties::STAT_REQUESTS))
        setStatRequests(0);

      if (!statProperties.isExist(HCE::handlers::properties::STAT_RESPONSES))
        setStatResponses(0);

      rebuildAdminClientSocket();

      rebuildClientConnection(_inprocProxySock, NODE_CS_INPROC_PROXY, ZMQ_DEALER, getClientIdentity());

      rebuildClientConnection(_inprocReducerOutSock, NODE_CS_INPROC_REDUCER_OUT, ZMQ_DEALER);

      rebuildDataClientSocket();

      return true;
    }

    void DataClientProxy::deinitialize(void){
      if(_inprocProxySock){
        if(_inprocProxySock->connected()){
          _inprocProxySock->close();
        }
        delete _inprocProxySock;
        _inprocProxySock = NULL;
      }

      if(_inprocReducerOutSock){
        if(_inprocReducerOutSock->connected()){
          _inprocReducerOutSock->close();
        }
        delete _inprocReducerOutSock;
        _inprocReducerOutSock = NULL;
      }

      if(_dataClientSock){
        if(_dataClientSock->connected()){
          sendByeMessage(_dataClientSock, getClientIdentity());
          _dataClientSockConnected = false;
          _dataClientSock->close();
        }
        delete _dataClientSock;
        _dataClientSock = NULL;
      }

      HCE::drce::ResourceUsageExtractor::shutdown();
    }

    void DataClientProxy::setHeartbeatAt(int64_t heartbeatAt){
      handlerProperties.set(HCE::handlers::properties::HEARTBEAT_AT, heartbeatAt);
    }

    int64_t DataClientProxy::getHeartbeatAt(void){
      return handlerProperties.get<int64_t>(HCE::handlers::properties::HEARTBEAT_AT);
    }

    void DataClientProxy::setHeartbeatedAt(int64_t heartbeatedAt){
      handlerProperties.set(HCE::handlers::properties::HEARTBEATED_AT, heartbeatedAt);
    }

    int64_t DataClientProxy::getHeartbeatedAt(void){
      return handlerProperties.get<int64_t>(HCE::handlers::properties::HEARTBEATED_AT);
    }

    void DataClientProxy::setStatRequests(unsigned int statRequests){
      statProperties.set(HCE::handlers::properties::STAT_REQUESTS, statRequests);
    }

    unsigned int DataClientProxy::getStatRequests(void){
      return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_REQUESTS);
    }

    void DataClientProxy::setStatResponses(unsigned int statResponses){
      statProperties.set(HCE::handlers::properties::STAT_RESPONSES, statResponses);
    }

    unsigned int DataClientProxy::getStatResponses(void){
      return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_RESPONSES);
    }

    void DataClientProxy::rebuildDataClientSocket(void){
      _dataClientSockConnected = rebuildClientConnection(_dataClientSock, getConnectionString(), ZMQ_DEALER, getClientIdentity(), 0, NODE_MSG_READY);
    }

    void DataClientProxy::processControlMessage(zmsg& msg){

      std::string addr((msg.address())?msg.address():"");
      if(addr.compare(0, NODE_MSG_HEARTBEAT.length(), NODE_MSG_HEARTBEAT)==0){
        logger.log(NODE_LOG_MODE_DEBUG) << "RECEIVED HEARTBEAT from server" << flush;

        setHeartbeatedAt(s_clock());

        if (getHeartbeatMode() == HEARBEAT_MODE_IMMEDIATE_AFTER_HB ||
            getHeartbeatMode() == HEARBEAT_MODE_MIXED){
          sendHeartbeat();
          logger.log(NODE_LOG_MODE_TRACE) << "HEARBEAT_MODE_IMMEDIATE_AFTER_HB || HEARBEAT_MODE_MIXED" << flush;
        }
      }else{
        logger.log(NODE_LOG_MODE_TRACE) << "ERROR: wrong control message received" << endl << msg.dump() << flush;
      }
    }

    void DataClientProxy::processDataMessage(zmsg& msg){
      unsigned int statRequests = getStatRequests();
      setStatRequests(++statRequests);

      logger.log(NODE_LOG_MODE_TRACE) << "MSG FROM dataclient" << endl << msg.dump() << flush;
      logger.log(NODE_LOG_MODE_DEBUG) << "RECEIVED REQUEST from upper server, n=" << getStatRequests() << flush;

      msg.send(*_inprocProxySock);

      logger.log(NODE_LOG_MODE_DEBUG) << "REQUEST SENT to inproc_proxy socket" << flush;

      if (getHeartbeatMode() == HEARBEAT_MODE_ADAPTIVE){
        setHeartbeatedAt(s_clock());
        _shiftCounter.reset();
        setHeartbeatAt(s_clock() + getHeartbeatDelay());
      }
    }

    void DataClientProxy::handleExternalMessage(void){
      zmsg clientMsg(*_dataClientSock);

      if(clientMsg.parts() == NODE_MSG_CMD_MIN_FIELDS){
        processControlMessage(clientMsg);
      }else{
        processDataMessage(clientMsg);
      }

      setHeartbeatedAt(s_clock());
    }

    void DataClientProxy::handleInternalMessage(void){
      zmsg msg(*_inprocReducerOutSock);

      if (msg.parts() == NODE_MSG_CMD_MIN_FIELDS)
      {
        char* addrp = msg.address();
        std::string addr ((addrp)?addrp:"");
        if (addr.compare(0, NODE_MSG_ROUTES.length(), NODE_MSG_ROUTES)==0)
        {
          s_send(*_dataClientSock, addr);  // resend message to server
        }
      }
      else
      {
        unsigned int statResponses = getStatResponses();
        setStatResponses(++statResponses);

        logger.log(NODE_LOG_MODE_TRACE) << "MSG FROM inproc_reducer_out" << endl << msg.dump() << flush;
        logger.log(NODE_LOG_MODE_DEBUG) << "RESPONSE SENT to dataclient socket from proxy node to upper server, n=" << getStatResponses() << flush;

        msg.send(*_dataClientSock);
      }
    }

    void DataClientProxy::handleRouteMessage(void){

      zmsg msg(*_inprocProxySock);
      logger.log(NODE_LOG_MODE_TRACE) << " MSG FROM inproc_proxy" << endl << msg.dump() << flush;

      std::string addr((msg.address())?msg.address():"");

      if (addr.compare(0, NODE_MSG_ROUTES.length(), NODE_MSG_ROUTES)==0)
      {
        RouteMessage routeMessage(_clientsQueue);
        routeMessage.extract(NODE_MSG_ROUTES, addr);
        if (routeMessage.isError())
          logger.log(NODE_LOG_MODE_ERROR) << "Operation extract ROUTE MESSAGE has error: " << routeMessage.getErrorMsg() << flush;
        else
        {
          logger.log(NODE_LOG_MODE_TRACE) << "Operation extract ROUTE MESSAGE was made SUCCESSFULLY" << flush;
        }
      }
    }

    void DataClientProxy::heartbit(void){
      if (getHeartbeatMode() == HEARBEAT_MODE_FIXED_TIME ||
          getHeartbeatMode() == HEARBEAT_MODE_MIXED){
        if(s_clock() > getHeartbeatAt() && _dataClientSockConnected){
          sendHeartbeat();

          logger.log(NODE_LOG_MODE_DEBUG) << "SENT HEARTBIT from client [" << getClientIdentity() << "] to server" << flush;
        }
      }else if (getHeartbeatMode() == HEARBEAT_MODE_ADAPTIVE){
        _shiftCounter.responceTimeStamp(getHeartbeatedAt());
        _shiftCounter.recount(getHeartbeatTimeout());

        logger.log(NODE_LOG_MODE_TRACE) << "DataClientProxy::heartbit  HEARBEAT_MODE_ADAPTIVE  delta = " << (s_clock()-getHeartbeatedAt()+getHeartbeatDelay())
            << " _shiftCounter.getShift()= " << _shiftCounter.getShift() << flush;

        size_t delta = s_clock()-getHeartbeatAt()+getHeartbeatDelay();
        if (delta > (getHeartbeatTimeout()+_shiftCounter.getShift()) && _dataClientSockConnected){
          
          sendHeartbeat();
          _shiftCounter.reset();

          logger.log(NODE_LOG_MODE_DEBUG) << "SENT HEARTBIT from client [" << getClientIdentity() << "] to server getHeartbeatAt() = '"
              << Poco::DateTimeFormatter::format(Poco::LocalDateTime(Poco::Timestamp(Poco::Timestamp::fromEpochTime(getHeartbeatAt()/1000))), "%Y-%m-%d %H:%M:%S") << "'" << flush;
        }
        setHeartbeatedAt(s_clock());
      }
    }

    void DataClientProxy::sendHeartbeat(void){
      HCE::drce::ResourceUsageExtractor::getInstance().update();
      HCE::drce::ResourceUsage resourceUsage = HCE::drce::ResourceUsageExtractor::getInstance().extract();
      HCE::drce::ResourceUsageConverter resourceUsageConverter(resourceUsage);
      HCE::handlers::HeartbeatMessage heartbeatMessage(_clientsQueue, resourceUsageConverter.toString());
      std::string message = heartbeatMessage.build(NODE_MSG_HEARTBEAT);

      logger.log(NODE_LOG_MODE_TRACE) << "SENT HEARTBIT MESSAGE: " << message << flush;
      s_send(*_dataClientSock, message);

      setHeartbeatAt(s_clock() + getHeartbeatDelay());
    }

    void DataClientProxy::externalConnectionRefresh(void){

      if (getHeartbeatMode() != HEARBEAT_MODE_ADAPTIVE){

        if(((s_clock() - getHeartbeatedAt()) > getHeartbeatTimeout()) && getInProgress() && _dataClientSockConnected){
          logger.log(NODE_LOG_MODE_INFORMATION) << "RECONNECT client [" << getClientIdentity() << "] because HEARTBIT not received during time of "
              << (getHeartbeatTimeout()+_shiftCounter.getShift()) << " msec" << flush;

          rebuildDataClientSocket();

          setHeartbeatedAt(s_clock());

          setHeartbeatAt(s_clock() + getHeartbeatDelay());
        }
      }else{ // HEARBEAT_MODE_ADAPTIVE

        _shiftCounter.recount(getHeartbeatTimeout());

        logger.log(NODE_LOG_MODE_TRACE) << "DataClientProxy::externalConnectionRefresh  (s_clock()-getHeartbeatedAt())= " << (s_clock()-getHeartbeatedAt()) <<
            " getHeartbeatTimeout()= " << getHeartbeatTimeout() << " _shiftCounter.getShift()= " << _shiftCounter.getShift() << flush;

        if(((s_clock() - getHeartbeatedAt()) > (getHeartbeatTimeout()+_shiftCounter.getShift())) && getInProgress() && _dataClientSockConnected){
          logger.log(NODE_LOG_MODE_INFORMATION) << "RECONNECT client [" << getClientIdentity() << "] because HEARTBIT not received during time of "
              << (getHeartbeatTimeout()+_shiftCounter.getShift()) << " msec" << flush;

          rebuildDataClientSocket();

          //Reset time mark
          setHeartbeatedAt(s_clock());

          setHeartbeatAt(s_clock() + getHeartbeatDelay());
        }
      }
    }

    std::string DataClientProxy::getPropertyInformation(void){
      std::stringstream result;
      result << NODE_ADMIN_ERROR_OK
          << NODE_ADMIN_COMMAND_DELIMITER << getProperties() << "requests=" << getStatRequests()
          << NODE_ADMIN_COMMAND_DELIMITER << "responses=" << getStatResponses()
          << NODE_ADMIN_COMMAND_DELIMITER << "clientIdentity=" << getClientIdentity()
          << NODE_ADMIN_COMMAND_DELIMITER << "heartbeatAt=" << static_cast<uint64_t>(getHeartbeatAt())
          << NODE_ADMIN_COMMAND_DELIMITER << "heartbeatedAt=" << static_cast<uint64_t>(getHeartbeatedAt());
      return result.str();
    }

    std::string DataClientProxy::processAdminCommands(std::string& command, std::string& parameters){
      std::string ret = NODE_ADMIN_ERROR_OK;

      logger.log(NODE_LOG_MODE_DEBUG) << "Admin command requested [" << command << "], parameters [" << parameters << "]" << flush;

      if(command == NODE_MSG_STAT){
        std::stringstream result;
        result << NODE_ADMIN_ERROR_OK << NODE_ADMIN_COMMAND_DELIMITER << "requests=" << getStatRequests() << ",responses=" << getStatResponses();
        ret = result.str();
      }else if(command == NODE_MSG_REBUILD_CLIENT_CONNECTION){
        rebuildDataClientSocket();
      }else if(command == NODE_MSG_DISCONNECT_CLIENT_CONNECTION){
        _dataClientSock->close();
        _dataClientSockConnected = false;
      }else if(command == NODE_MSG_UPDATE_SCHEMA){
        setConnectionString(parameters);
      }else if(command == NODE_MSG_PROPERTIES){
        ret = getPropertyInformation();
      }else if(command == NODE_MSG_HEARTBEAT_DELAY_GET){
        ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_DELAY + "=" + std::to_string(getHeartbeatDelay()/1000);
      }else if(command == NODE_MSG_HEARTBEAT_DELAY_SET){
        unsigned int heartbeatDelay = HEARTBEAT_DELAY;
        std::istringstream(parameters) >> heartbeatDelay;
        ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_DELAY + "=" + std::to_string(getHeartbeatDelay()/1000);
        setHeartbeatDelay(heartbeatDelay*1000);
      }else if(command == NODE_MSG_HEARTBEAT_TIMEOUT_GET){
        ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_TIMEOUT + "=" + std::to_string(getHeartbeatTimeout()/1000);
      }else if(command == NODE_MSG_HEARTBEAT_TIMEOUT_SET){
        int hbTimeout = HEARTBEAT_TIMEOUT;
        std::istringstream(parameters) >> hbTimeout;
        ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_TIMEOUT + "=" + std::to_string(getHeartbeatTimeout()/1000);
        setHeartbeatTimeout(static_cast<unsigned int>(hbTimeout*1000));
      }else if(command == NODE_MSG_HEARTBEAT_MODE_GET){
        ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_MODE + "=" + std::to_string(getHeartbeatMode());
      }else if(command == NODE_MSG_HEARTBEAT_MODE_SET){
        int hbMode = HEARBEAT_MODE_DEFAULT;
        std::istringstream(parameters) >> hbMode;
        ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_MODE + "=" + std::to_string(getHeartbeatMode());
        setHeartbeatMode(static_cast<unsigned int>(hbMode));
      }else{
        ret = NODE_ADMIN_ERROR_ERROR + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_ERROR_UNSUPPORTED_COMMAND;
      }

      return ret;
    }

    void* DataClientProxy::process(void){
      logger.log(NODE_LOG_MODE_DEBUG) << NODE_ADMIN_ERROR_MAIN_MSG_LOOP_STARTED << flush;

      while(getInProgress()){
        try {
          std::vector<zmq::pollitem_t> items;
          if(_dataClientSockConnected){
            items ={{*_dataClientSock, 0, ZMQ_POLLIN, 0},{*_inprocReducerOutSock, 0, ZMQ_POLLIN, 0},{*_inprocAdminSock, 0, ZMQ_POLLIN, 0},{*_inprocProxySock, 0, ZMQ_POLLIN, 0}}; //add new socket
          } else{
            items ={{*_inprocReducerOutSock, 0, ZMQ_POLLIN, 0},{*_inprocAdminSock, 0, ZMQ_POLLIN, 0},{*_inprocProxySock, 0, ZMQ_POLLIN, 0}}; //add new socket inproc_proxy
          }

          zmq::poll(&items[0], items.size(), getPollTimeout());

          if(_dataClientSockConnected){
            if(items[0].revents & ZMQ_POLLIN){
              handleExternalMessage();
            }
          }

          if(items[items.size() - 3].revents & ZMQ_POLLIN){
            handleInternalMessage();
          }

          if(items[items.size() - 2].revents & ZMQ_POLLIN){
            handleAdminMessage();
          }

          if(items[items.size() - 1].revents & ZMQ_POLLIN){
            handleRouteMessage();
          }

          heartbit();

          sendPeriodicProperties();

          dumpRunTimeVariables();

          externalConnectionRefresh();
        }catch(zmq::error_t& error){
          if (EINTR!=error.num()){
            std::cout << "DataClientProxy E: " << error.what() << std::endl;
          }
        }catch(std::exception& e){
          std::cout << "DataClientProxy E: " << e.what() << std::endl;
        }catch(...){
          std::cout << "DataClientProxy E: Unknown exception" << std::endl;
        }
      }
      logger.log(NODE_LOG_MODE_INFORMATION) << NODE_ADMIN_ERROR_MAIN_MSG_LOOP_STOPPED << flush;

      return NULL;
    }

  }
}