DataReducerProxy.hpp

Go to the documentation of this file.

#ifndef __HCE_DATAREDUCER_PROXY_INCLUDED__
#define __HCE_DATAREDUCER_PROXY_INCLUDED__

#include <sstream>
#include <map>
#include <iterator>
#include <Poco/Checksum.h>

#include "Handler.hpp"

//Sphinx functional object dependencies
#include "SphinxInputJsonMessage.hpp"
#include "SphinxOutputJsonMessage.hpp"
#include "SphinxSearcher.hpp"
#include "SphinxError.hpp"
//Reducer functional object dependencies
#include "ReducerFunctionalObject.hpp"
//Message cover object
#include "JsonMessageCover.hpp"

namespace HCE {
  namespace handlers {

    class DataReducerProxy: public HCE::handlers::Handler {
      public:
        const int TTL_TASKS_CLEANUP_INTERVAL_DEFAULT = 10000;
        const int TTL_TASKS_CLEANUP_NUMBER_DEFAULT = 100;
        const int TTL_TASKS_QUEUE_CLEANUP_NUMBER_DEFAULT = 1000;
        const int TTL_TASKS_QUEUE_CLEANUP_TIME = 200;

        DataReducerProxy(std::string name, zmq::context_t& ctx, const std::string& clientIdentity, unsigned char nodeMode, int64_t startedAt, unsigned char logPriority, const std::string& iniFile);
        ~DataReducerProxy();

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

        void handleInternalMessage(void);
        std::string processAdminCommands(std::string& command, std::string& parameters);

        void reducerInit(void);

        void defaultReducerResponseItemCreate(std::string& messageId, std::string& messageBody);
        static void defaultReducerResponseItemAccumulate(std::map<std::string, DefaultReducerResponseItem*>::iterator& DefaultReducerResponseItemIterator, std::string& messageBody);
        std::string defaultReducerResponseItemReduce(std::map<std::string, DefaultReducerResponseItem*>::iterator& DefaultReducerResponseItemIterator);
        void defaultReducerReducePlainText(std::string& messageId, std::string& messageBody);
        void defaultReducerReduceJson(std::string& messageId, std::string& messageBody);

        void reduceJson(std::string& messageId, std::string& messageBody);
        void reducerTasksCleanupByTtl(void);

        std::string getPropertyInformation(void);

        zmq::socket_t* _inprocReducerInSock;
        zmq::socket_t* _inprocReducerOutSock;

        void setStatRequests(unsigned int statRequests);
        unsigned int getStatRequests(void);

        void setStatResponses(unsigned int statResponses);
        unsigned int getStatResponses(void);

        void setClientsNumber(unsigned int clientsNumber);
        unsigned int getClientsNumber(void);

        void setNodeMode(unsigned char nodeMode);
        unsigned char getNodeMode(void);

        int64_t _reducerTtlTasksCleanupAt;

        //Sphinx reducer FO
        HCE::reduce::ReducerFunctionalObject* _extendedReducerFO;

        //TODO: only for model purposes, need to be removed in final version
        std::map<std::string, DefaultReducerResponseItem*> _responses;
    };

  }
}

#endif /* __HCE_DATAREDUCER_PROXY_INCLUDED__ */