DataProcessorData.hpp

Go to the documentation of this file.

#ifndef __HCE_DATA_PROCESSOR_DATA_INCLUDED__
#define __HCE_DATA_PROCESSOR_DATA_INCLUDED__

#include <iostream>
#include <sstream>
#include <iterator>

#include "Handler.hpp"

//Node ini
#include <Poco/Util/IniFileConfiguration.h>
#include "Poco/String.h"

//Sphinx FO
#include "SphinxInputJsonMessage.hpp"
#include "SphinxOutputJsonMessage.hpp"
#include "SphinxFunctionalObject.hpp"
//DRCE FO
#include "DRCEDefaultJSON.hpp"
#include "DRCEInputJsonMessage.hpp"
#include "DRCEOutputJsonMessage.hpp"
#include "DRCEFunctionalObject.hpp"
//Reducer input object
#include "PocoJSONReducingInputMessageConvertor.hpp"
#include <Poco/SharedPtr.h>
#include "Exceptions.hpp"
//Message cover object
#include "JsonMessageCover.hpp"

//User home directory path
#include <unistd.h>
#include <sys/types.h>
#include <pwd.h>

//Task state notification
#include <Poco/UUID.h>

namespace HCE {
  namespace handlers {

    class DataProcessorData: public HCE::handlers::Handler {
      public:
        DataProcessorData(const std::string& nodeName, std::string name, zmq::context_t& ctx, int64_t startedAt, unsigned char logPriority, const std::string& iniFile);
        ~DataProcessorData();

      protected:
        int initialize(void);
        void deinitialize(void);
        void* process(void);

        void handleInternalMessage(void);
        std::string processDataRequest(const std::string& messageBody);
        std::string processAdminCommands(std::string& command, std::string& parameters);
        std::string processAdminCommandsSphinx(std::string& command);

        void setNodeName(const std::string& nodeName);
        std::string getNodeName(void);

        std::string processDataRequestPlainText(const std::string& messageBody);

        std::string processDataRequestJson(const std::string& messageBody);

        std::string processDataRequestSphinx(const std::string& messageBody);

        void dataStorageInitSphinx(const std::string& iniFile);

        void nodeInit(const std::string& iniFile);

        void checkInit(void);

        void saveNodeProperties(void);

        void saveSphinxProperties(void);

        void saveDRCEProperties(void);

        void restoreNodeProperties(void);

        void restoreSphinxProperties(void);

        void restoreDRCEProperties(void);

        static std::string getUserHomeDir(void);

        std::string getDataStorageSphinxProperties(void);

        std::string getDataStorageDRCEProperties(void);

        std::string processDataRequestDRCE(const std::string& messageBody);

        void dataStorageInitDRCE(const std::string& iniFile);

        bool rebuildTasksStateNotificationConnection();

        std::string getPropertyInformation(void);

        zmq::socket_t* _inprocDataSock;
        zmq::socket_t* _tasksStateNotificationSock;

        void setStatClientResponses(unsigned long long statClientResponses);
        unsigned long long getStatClientResponses(void);

        void setStatSphinxRequestsSearch(unsigned long long statSphinxRequestsSearch);
        unsigned long long getStatSphinxRequestsSearch(void);

        void setStatSphinxRequestsAdmin(unsigned long long statSphinxRequestsAdmin);
        unsigned long long getStatSphinxRequestsAdmin(void);

        void setStatSphinxTimeTotalSearch(unsigned long long statSphinxTimeTotalSearch);
        unsigned long long getStatSphinxTimeTotalSearch(void);

        void setStatSphinxTimeTotalAdmin(unsigned long long statSphinxTimeTotalAdmin);
        unsigned long long getStatSphinxTimeTotalAdmin(void);

        void setStatDRCERequests(unsigned long long statDRCERequests);
        unsigned long long getStatDRCERequests(void);

        void setStatDRCETimeTotal(unsigned long long statDRCETimeTotal);
        unsigned long long getStatDRCETimeTotal(void);

        HCE::sphinx::SphinxFunctionalObject* _sphinxFO;

        HCE::drce::DRCEFunctionalObject* _drceFO;

        void setDataProcessingMode(unsigned int dataProcessingMode);
        unsigned int getDataProcessingMode(void);

        void setDataProcessingFakeResults(unsigned int dataProcessingFakeResults);
        unsigned int getDataProcessingFakeResults(void);

        void setDataProcessingFakeResultsMax(unsigned int dataProcessingFakeResultsMax);
        unsigned int getDataProcessingFakeResultsMax(void);

        void setNodeNumber(unsigned int NodeNumber);
        unsigned int getNodeNumber(void);

        void setStateNotificationHost(const std::string& stateNotificationHost);
        std::string getStateNotificationHost(void);

        void setStateNotificationPort(const std::string& stateNotificationPort);
        std::string getStateNotificationPort(void);

        class TasksStateNotificationFunctor {
          public:
            TasksStateNotificationFunctor(zmq::socket_t* _ssock):_msgId(Poco::UUID::x500().toString()),_updateConnectionSock( _ssock){
              //std::cout << "\n\n***Constructor::TasksStateNotificationFunctor***\n\n";
              //_updateConnectionSock = _ssock;
              //_msgId = Poco::UUID::x500().toString();
            }

            ~TasksStateNotificationFunctor(){
              if(_updateConnectionSock){
                //std::cout << "\n\n***Destructor::TasksStateNotificationFunctor***\n\n";
              }
            }

            std::string operator()(const std::string& data){
//              std::cout << "!!!TasksStateNotificationFunctor!!!" << std::endl;
//              std::cout << data << std::endl;
//              std::cout << "!!!TasksStateNotificationFunctor!!!" << std::endl;

              if(_updateConnectionSock){
                //Create new message with Id
                zmsg msg(_msgId.c_str());
                msg.append(data.c_str());

                msg.send(*_updateConnectionSock);
              }

              return "";
            }

            //Message Id
            std::string _msgId;
            zmq::socket_t* _updateConnectionSock;
        };

        TasksStateNotificationFunctor* _tasksStateNotificationFunctor;
    };

  }
}

#endif /* __HCE_DATA_PROCESSOR_DATA_INCLUDED__ */