ReducerFunctionalObject.cpp

Go to the documentation of this file.

#include "ReducerFunctionalObject.hpp"

namespace HCE{
namespace reduce{

ReducerFunctionalObject::ReducerFunctionalObject()
{
  ReducingHandlerBuilder reducingHandlerBuilder;
  _reducingHandler = reducingHandlerBuilder.build();
  _rejectedMessages = 0;
  errCode = 0;
}

ReducerFunctionalObject::~ReducerFunctionalObject(){}

void ReducerFunctionalObject::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::types::MessageType::mtNull, HCE::reduce_types::FAKE_TTL, emptyMsg));
  try{
    reducingInputMessage = _pocoJSONReducingInputMessageConvertor.
                          convertToReducingInputMessageFrom(jsonReducingInputMessage);
    errCode = OK;
  }
  catch(HCE::reduce::WrongJSONStructureException& e){
    errCode = WRONG_JSON_STRUCT_ERROR;
    errMsg  = WRONG_JSON_STRUCT_DESCR;
  }
  _reducingHandler->accumulateReducingData(itemId, reducingInputMessage);
}

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

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

unsigned int ReducerFunctionalObject::getTasksNumber()
{
  return _reducingHandler->getTasksNumber();
}

int ReducerFunctionalObject::getErrCode() const
{
  return errCode;
}

std::string ReducerFunctionalObject::getErrMsg() const
{
  return errMsg;
}

int ReducerFunctionalObject::cleanupExpiredTasksByTTL(unsigned int maxRemoveTasks,
                                                      unsigned int ttl)
{
  return _reducingHandler->cleanupExpiredTasksByTTL( maxRemoveTasks, ttl );
}

unsigned int ReducerFunctionalObject::cleanupExpiredTasksByTTLQueue(
                                      unsigned int maxRemoveItems,
                                      unsigned int minTerminateTime )
{
  return _reducingHandler->cleanupExpiredTasksByTTLQueue( maxRemoveItems, minTerminateTime );
}

void ReducerFunctionalObject::getExceededTTLTasks(unsigned int maxTaskNumber,
                                     std::vector<unsigned long long>& exceededTTLTasks,
                                     unsigned int ttl)
{
  _reducingHandler->getExceededTTLTasks( maxTaskNumber, exceededTTLTasks, ttl );
}

int ReducerFunctionalObject::getRejectedMessages() const
{
  return _rejectedMessages;
}

}
}