ReducerFunctionalObjectSphinx.hpp

Go to the documentation of this file.

#ifndef REDUCERFUNCTIONALOBJECT_HPP_
#define REDUCERFUNCTIONALOBJECT_HPP

#include "ReducingHandlerBuilder.hpp"
#include "PocoJSONReducingInputMessageConvertor.hpp"
#include <Poco/SharedPtr.h>

namespace HCE
{
namespace reduce
{

const int OK = 0;
const int WRONG_JSON_STRUCT_ERROR = 1;
const int NOT_FOUND_ITEM_ID_ERROR = 2;
const std::string WRONG_JSON_STRUCT_DESCR = "err string";
const std::string NOT_FOUND_ITEM_ID_DESCR = "err string2";

class ReducerFunctionalObjectSphinx{
  public:
    ReducerFunctionalObjectSphinx()
    {
      ReducingHandlerBuilder reducingHandlerBuilder;
      _reducingHandler = reducingHandlerBuilder.build();
      _rejectedMessages = 0;
    }

    void accumulate(std::string& jsonReducingInputMessage, unsigned long long itemId)
    {
      if( _reducingHandler->isTaskTerminatedByTTL( itemId ) ) {
          _rejectedMessages ++;
          return;
      } 
      
      std::string emptyMsg = "";
      Poco::SharedPtr<HCE::reduce_types::ReducingInputMessage>reducingInputMessage(
      new HCE::reduce_types::ReducingInputMessage(HCE::reduce_types::FAKE_MSG_TYPE, HCE::reduce_types::FAKE_TTL, emptyMsg));
      try{
        reducingInputMessage = _pocoJSONReducingInputMessageConvertor.convertToReducingInputMessageFrom(jsonReducingInputMessage);
        errCode = OK;
      }
      catch(HCE::sphinx::reduce_task::WrongJSONStructureException& e){
        errCode = WRONG_JSON_STRUCT_ERROR;
        errMsg  = WRONG_JSON_STRUCT_DESCR;
      }
      _reducingHandler->accumulateReducingData(itemId, reducingInputMessage);
    }

    bool isCompleteTask(unsigned long long itemId, int nodesCount, unsigned int ttl = USE_TASK_TTL_VALUE)
    {
      try{
        errCode = OK;
        return  _reducingHandler->isTaskExpired( itemId, ttl ) || _reducingHandler->getAccumulateReducingPartsNumberBy(itemId) == nodesCount;
      }
      catch(HCE::sphinx::reduce_task::NotFoundByKeyException& e){
        errCode = NOT_FOUND_ITEM_ID_ERROR;
        errMsg  = NOT_FOUND_ITEM_ID_DESCR;
      }
      return false;
    }

    Poco::SharedPtr<HCE::reduce_types::ReducingOutputMessage> reduce(unsigned long long itemId)
    {
      Poco::SharedPtr<HCE::reduce_types::ReducingOutputMessage>reducingOutputMessage = _reducingHandler->makeReducing(itemId);
      _reducingHandler->deleteReducingTaskBy(itemId);
      errCode = OK;
      return reducingOutputMessage;
    }
    
    unsigned int getTasksNumber(){
      return _reducingHandler->getTasksNumber();
    }

    int getErrCode(){
      return errCode;
    }

    std::string getErrMsg(){
      return errMsg;
    }
    
    int cleanupExpiredTasksByTTL(unsigned int maxRemoveTasks = 1, unsigned int ttl = USE_TASK_TTL_VALUE) {
      return _reducingHandler->cleanupExpiredTasksByTTL( maxRemoveTasks, ttl );
    }
    
    unsigned int cleanupExpiredTasksByTTLQueue( unsigned int maxRemoveItems, unsigned int minTerminateTime ) {
      return _reducingHandler->cleanupExpiredTasksByTTLQueue( maxRemoveItems, minTerminateTime );
    }
    
    void getExceededTTLTasks( unsigned int maxTaskNumber, std::vector<unsigned long long>& exceededTTLTasks,
                              unsigned int ttl = USE_TASK_TTL_VALUE) {
      _reducingHandler->getExceededTTLTasks( maxTaskNumber, exceededTTLTasks, ttl );
    }
    
    int getRejectedMessages() const {
      return _rejectedMessages;
    }

  private:
    Poco::SharedPtr<HCE::reduce::ReducingHandler<> >_reducingHandler;
    HCE::reduce_convertors::PocoJSONReducingInputMessageConvertor _pocoJSONReducingInputMessageConvertor;
    int errCode;
    std::string errMsg;
    int _rejectedMessages;
};

}
}

#endif