ReducingHandler.cpp

Go to the documentation of this file.

#include "ReducingHandler.hpp"

namespace HCE{
namespace reduce{

ReducingHandler::ReducingHandler(
    const Poco::SharedPtr<ReducingPartsCountersStorageIf>& reducingPartsCountersStorage,
    const Poco::SharedPtr<ReduceAdditionsStorageIf>& reduceAdditionsStorage,
    const Poco::SharedPtr<TaskReducersManagerFactoryIf> &taskReducersManagerFactory,
    const Poco::SharedPtr<ReducingOutputMessageBuilderIf>& reducingOutputMessageBuilder,
    const Poco::SharedPtr<SupportNotifierIf>& supportNotifier)
{
    _reducingPartsCountersStorage = reducingPartsCountersStorage;
    _reduceAdditionsStorage = reduceAdditionsStorage;
    _taskReducersManagerFactory = taskReducersManagerFactory;
    _reducingOutputMessageBuilder = reducingOutputMessageBuilder;
    _supportNotifier = supportNotifier;
  }

ReducingHandler::~ReducingHandler(){}

void ReducingHandler::accumulateReducingData(unsigned long long itemId,
                    Poco::SharedPtr<reduce_types::ReducingInputMessage>& reducingInputMessage)
{
  try{
    if (not(_reducingPartsCountersStorage->findBy(itemId))){
      createNewTaskReducersManager(reducingInputMessage->getProcessingType());
      _reduceAdditionsStorage->addWith(itemId,
                      _reduceAdditionsStorage->findBy(reducingInputMessage->getProcessingType()));
      _reducingPartsCountersStorage->addReducingPartsCounterWithKey(itemId);
      _reduceAdditionsStorage->findBy(reducingInputMessage->getProcessingType())->createReducer(itemId);
    }
    _reducingPartsCountersStorage->incrementBy(itemId);
    _taskTTLManager.addTaskTTL( itemId, reducingInputMessage->getTTL() );
    _reduceAdditionsStorage->findBy(itemId)->addDataInReducer(itemId, reducingInputMessage->getBody());
  }
  catch(NotFoundByKeyException& e){
    _supportNotifier->notifyNotFoundItemException(e);
  }
  catch(WrongJSONStructureException& e){
    _supportNotifier->notifyWrongJSONStructure(itemId, reducingInputMessage);
  }
}

Poco::SharedPtr<reduce_types::ReducingOutputMessage>ReducingHandler::makeReducing(unsigned long long itemId)
{
  try{
     unsigned int taskTTL = _taskTTLManager.getTaskTTL( itemId );
     //remove the task to avoid the task will be moved to terminated task queue if the reducing task spends too long time
     _taskTTLManager.removeTaskTTL( itemId );
     types::MessageType msgTypes = _reduceAdditionsStorage->findBy(itemId)->getType();
     std::string reducingJSON =_reduceAdditionsStorage->findBy(itemId)->runReduceTaskForTaskId(itemId);
     return _reducingOutputMessageBuilder->build(msgTypes, taskTTL, reducingJSON);
  }
  catch(NotFoundByKeyException& e){
    _supportNotifier->notifyNotFoundItemException(e);
    return _reducingOutputMessageBuilder->build(e);
  }
}

void ReducingHandler::deleteReducingTaskBy(unsigned long long itemId)
{
  try{
    _reducingPartsCountersStorage->deleteBy(itemId);
    _reduceAdditionsStorage->findBy(itemId)->deleteReducerBy(itemId);
    _reduceAdditionsStorage->deleteBy(itemId);
  }
  catch(NotFoundByKeyException& e){
    _supportNotifier->notifyNotFoundItemException(e);
  }
}

int ReducingHandler::getAccumulateReducingPartsNumberBy(unsigned long long itemId)
{
  return _reducingPartsCountersStorage->getAccumulateReducingParts(itemId);
}

unsigned int ReducingHandler::getTasksNumber()
{
  return _reducingPartsCountersStorage->getTotalTasksNumber();
}

bool ReducingHandler::isTaskTerminatedByTTL( unsigned long long taskId ) const
{
  return _taskTTLManager.isTaskTerminatedByTTL( taskId );
}

bool ReducingHandler::isTaskExpired( unsigned long long taskId, unsigned int ttl)
{
  return _taskTTLManager.isTaskExpired( taskId, ttl );
}

int ReducingHandler::cleanupExpiredTasksByTTL(unsigned int maxRemoveTasks,
                                              unsigned int ttl)
{
  return _taskTTLManager.cleanupExpiredTasksByTTL(  maxRemoveTasks, ttl );
}

unsigned int ReducingHandler::cleanupExpiredTasksByTTLQueue(unsigned int maxRemoveItems,
                                                            unsigned int minTerminateTime )
{
  return _taskTTLManager.cleanupExpiredTasksByTTLQueue(maxRemoveItems, minTerminateTime );
}

void ReducingHandler::getExceededTTLTasks(unsigned int maxTaskNumber,
                                                  std::vector<unsigned long long>& exceededTTLTasks,
                                          unsigned int ttl) const
{
  _taskTTLManager.getExceededTTLTasks( maxTaskNumber, exceededTTLTasks, ttl );
}

void ReducingHandler::createNewTaskReducersManager(HCE::types::MessageType reducerType)
{
  try{
    _reduceAdditionsStorage->findBy(reducerType);
  }
  catch(NotFoundByKeyException &e){
    _reduceAdditionsStorage->addWith(reducerType,
                                     _taskReducersManagerFactory->build(reducerType));
  }
}

}
}