DRCETaskReducersManager.cpp

Go to the documentation of this file.

#include "DRCETaskReducersManager.hpp"
#include "Exceptions.hpp"
#include "DRCEResultDataSerializator.hpp"

using HCE::reduce::NotFoundByKeyException;
using HCE::reduce::WrongJSONStructureException;
using HCE::drce::DRCEResultDataSerializator;

namespace HCE {
namespace drce{
namespace reduce_task {


DRCETaskReducersManager::~DRCETaskReducersManager(){}

types::MessageType DRCETaskReducersManager::getType()
{
  return HCE::types::MessageType::mtDrce;
}

bool DRCETaskReducersManager::isReducerExist(unsigned long long reduceTaskId)
{
  return taskReduceResults_.find( reduceTaskId ) != taskReduceResults_.end();
}

void DRCETaskReducersManager::createReducer(unsigned long long reduceTaskId)
{
  auto iter = taskReduceResults_.find( reduceTaskId );
  
  if( iter == taskReduceResults_.end() ) {
    taskReduceResults_[ reduceTaskId ] = new DRCEResultData();
  }
}

void DRCETaskReducersManager::addDataInReducer(unsigned long long reduceTaskId, const std::string& json)
{
  auto iter = taskReduceResults_.find( reduceTaskId );
  if( iter == taskReduceResults_.end() ) {
    throw NotFoundByKeyException( "Not exist task id " + std::to_string( reduceTaskId ) );
  }
  
  DRCEResultData tmpResult;
  
  DRCEResultDataSerializator resultSerializator( tmpResult );
  
  if( resultSerializator.unserialize( json ) ) {
  
        Poco::SharedPtr< DRCEResultData > resultData = iter->second;
        
        for( size_t i = 0, n = tmpResult.getItemsCount(); i < n; i++ ) {
      resultData->addDataItem( tmpResult.getDataItem( i ) );
        } 
  } else {
    throw WrongJSONStructureException( resultSerializator.getErrorMsg() );
  }
}

std::string DRCETaskReducersManager::runReduceTaskForTaskId(unsigned long long reduceTaskId)
{
  auto iter = taskReduceResults_.find( reduceTaskId );
  
  if( iter == taskReduceResults_.end() ) {
    throw NotFoundByKeyException( "Not exist task id " + std::to_string( reduceTaskId ) );
  } 
  
  Poco::SharedPtr< DRCEResultData > resultData = iter->second;
    
  DRCEResultDataSerializator resultSerializator( *resultData );
  std::string json;
            
    if( resultSerializator.serialize( json ) ) {
      return json;
        }
        
        throw NotFoundByKeyException( resultSerializator.getErrorMsg() );
}

void DRCETaskReducersManager::deleteReducerBy(unsigned long long reduceTaskId)
{
  if( taskReduceResults_.erase( reduceTaskId ) != 1 ) {
    throw NotFoundByKeyException( "Not exist task id " + std::to_string( reduceTaskId ) );
  }
}

}
}
}