SphinxTaskReducersManager.cpp

Go to the documentation of this file.

#include "SphinxTaskReducersManager.hpp"

namespace HCE {
namespace sphinx {
namespace reduce_task {

SphinxTaskReducersManager::SphinxTaskReducersManager(
   const Poco::SharedPtr<SphinxTaskReducersStorageIf>& sphinxTaskReducersStorage,
   const Poco::SharedPtr<SphinxTaskReducerFactoryIf>& sphinxTaskReducerFactory,
   const Poco::SharedPtr<SphinxResultJSONConvertorIf>& sphinxResultJSONConvertor)
{
  _sphinxTaskReducersStorage = sphinxTaskReducersStorage;
  _sphinxTaskReducerFactory = sphinxTaskReducerFactory;
  _sphinxResultJSONConvertor = sphinxResultJSONConvertor;
}

SphinxTaskReducersManager::~SphinxTaskReducersManager(){}


types::MessageType SphinxTaskReducersManager::getType()
{
  return types::MessageType::mtSphinx;
}

bool SphinxTaskReducersManager::isReducerExist(unsigned long long reduceTaskId)
{
  try{
    _sphinxTaskReducersStorage->findByKey(reduceTaskId);
    return true;
  }
  catch(HCE::reduce::NotFoundByKeyException& e){
    return false;
  }
}

void SphinxTaskReducersManager::createReducer(unsigned long long reduceTaskId)
{
  Poco::SharedPtr<SphinxTaskReducerIf> sphinxTaskReducer =
                                          _sphinxTaskReducerFactory->create();
  _sphinxTaskReducersStorage->addWithKey(reduceTaskId, sphinxTaskReducer);
}

void SphinxTaskReducersManager::addDataInReducer(unsigned long long reduceTaskId,
                                                     const std::string& json)
{
  Poco::SharedPtr<SphinxResultData> sphinxResult =
                     _sphinxResultJSONConvertor->convertToSphinxResultDataFrom(json);
  Poco::SharedPtr<SphinxTaskReducerIf> sphinxTaskReducer =
          _sphinxTaskReducersStorage->findByKey(reduceTaskId);
  sphinxTaskReducer->addData(*sphinxResult.get());
}

std::string SphinxTaskReducersManager::runReduceTaskForTaskId(unsigned long long
                                                                          reduceTaskId)
{
  Poco::SharedPtr<SphinxTaskReducerIf> sphinxTaskReducer =
                         _sphinxTaskReducersStorage->findByKey(reduceTaskId);
  sphinxTaskReducer->reduce(_sphinxTaskReducerFactory->getScheduleStrategy());
  Poco::SharedPtr<SphinxResultData> reduceResult = sphinxTaskReducer->getResult();
  return _sphinxResultJSONConvertor->convertToJSONFrom(
                                     transformSphinxResultData(reduceResult));
}

void SphinxTaskReducersManager::deleteReducerBy(unsigned long long reduceTaskId)
{
  _sphinxTaskReducersStorage->deleteByKey(reduceTaskId);
}

Poco::SharedPtr<SphinxResultData> SphinxTaskReducersManager::transformSphinxResultData(
                                         Poco::SharedPtr<SphinxResultData>& reduceResult)
{
  SphinxResultTruncaterIf& sphinxResultTrancater =  _sphinxTaskReducerFactory->
                                                      getSphinxResultTrancater();
  return sphinxResultTrancater.trancate(reduceResult);
}

}
}
}