DRCEFunctionalObject.cpp

Go to the documentation of this file.

#include <unistd.h>
#include <fstream>
#include <Poco/Logger.h>

#include "DRCEResultDataSerializator.hpp"
#include "DRCEFunctionalObject.hpp"
#include "DRCEDefaultJSON.hpp"
#include "DRCEMessageConst.hpp"
#include "DRCETaskRequestExecutor.hpp"
#include "DRCETaskRequestFactory.hpp"
#include "DRCETaskRequestExecutorFactory.hpp"
#include "DRCENotificationExecutor.hpp"
#include "DRCETaskRequestSetExecute.hpp"
#include "DRCEReadProcessData.hpp"
#include "DRCEListAllTasks.hpp"
#include "DRCEResourceExtractor.hpp"
#include "DRCEStatVariablesConverter.hpp"
#include "DRCEAvgCountsConverter.hpp"

namespace HCE
{
namespace drce
{
//-----------------------------------------------------------------------------
DRCEFunctionalObject::DRCEFunctionalObject(const std::string& nodeName_, const std::string& homeDir_,
    const std::string& tasksDataDir_, const std::string& tasksStatusDir_,
    const std::string& nodeHost_, const std::string& nodePort_)
: DRCEFunctionalObjectBase(), DRCENodeOptions(nodeName_, "", tasksDataDir_, tasksStatusDir_, nodeHost_, nodePort_),
  messagesCollection(message_const::messages), message(messagesCollection), notificationExecutor(),
  asyncTaskQueue(*this, notificationExecutor, message), resourceMonitor(asyncTaskQueue, message), requestTime(0)
{
  setHomeDir(homeDir_);
  HCE::drce::ResourceUsageExtractor::getInstance();
  HCE::drce::DRCEReadProcessData::getInstance();
}
//-----------------------------------------------------------------------------
DRCEFunctionalObject::~DRCEFunctionalObject(void)
{
  HCE::drce::ResourceUsageExtractor::shutdown();
  HCE::drce::DRCEReadProcessData::shutdown();
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::setResourceExtractorMaxThreadsCount(size_t maxThreadsCount_)
{
  HCE::drce::ResourceUsageExtractor::getInstance().setMaxThreadsCount(maxThreadsCount_);
}
//-----------------------------------------------------------------------------
size_t DRCEFunctionalObject::getResourceExtractorMaxThreadsCount(void) const
{
  return HCE::drce::ResourceUsageExtractor::getInstance().getMaxThreadsCount();
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::setResourceExtractorMaxProcessesCount(size_t maxProcessesCount_)
{
  HCE::drce::ResourceUsageExtractor::getInstance().setMaxProcessesCount(maxProcessesCount_);
}
//-----------------------------------------------------------------------------
size_t DRCEFunctionalObject::getResourceExtractorMaxProcessesCount(void) const
{
  return HCE::drce::ResourceUsageExtractor::getInstance().getMaxProcessesCount();
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::setResourceExtractorMaxDiskSize(size_t maxDiskSize_)
{
  HCE::drce::ResourceUsageExtractor::getInstance().setMaxDiskSize(maxDiskSize_);
}
//-----------------------------------------------------------------------------
size_t DRCEFunctionalObject::getResourceExtractorMaxDiskSize(void) const
{
  return HCE::drce::ResourceUsageExtractor::getInstance().getMaxDiskSize();
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::setResourceExtractorMaxVramSize(size_t maxVramSize_)
{
  HCE::drce::ResourceUsageExtractor::getInstance().setMaxVramSize(maxVramSize_);
}
//-----------------------------------------------------------------------------
size_t DRCEFunctionalObject::getResourceExtractorMaxVramSize(void) const
{
  return HCE::drce::ResourceUsageExtractor::getInstance().getMaxVramSize();
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::setResourceExtractorMaxRramSize(size_t maxRramSize_)
{
  HCE::drce::ResourceUsageExtractor::getInstance().setMaxRramSize(maxRramSize_);
}
//-----------------------------------------------------------------------------
size_t DRCEFunctionalObject::getResourceExtractorMaxRramSize(void) const
{
  return HCE::drce::ResourceUsageExtractor::getInstance().getMaxRramSize();
}
//-----------------------------------------------------------------------------
std::string DRCEFunctionalObject::makeErrorMessage(const std::string& inputMessage,
    unsigned int state,
    unsigned int errorCode,
    const std::string& errorMessage)
{
  HCE::drce::DRCEInputJsonMessage inputJsonMessage(inputMessage);
  HCE::drce::DRCEResultDataItem resultDataItem;
  resultDataItem.setRequestId(inputJsonMessage.getRequestId());
  resultDataItem.setState(state);
  resultDataItem.setErrorCode(errorCode);
  resultDataItem.setErrorMessage(errorMessage);

  HCE::drce::DRCEResultData resultData;
  resultData.addDataItem(std::forward<HCE::drce::DRCEResultDataItem>(resultDataItem));

  HCE::drce::DRCEOutputJsonMessage outputJsonMessage;
  outputJsonMessage.setResultData(std::forward<HCE::drce::DRCEResultData>(resultData));

  std::string resultJsonMessage;
  if (!outputJsonMessage.serialize(resultJsonMessage))
  {
    std::stringstream errMsg;
    errMsg << message_const::serializationOutputJsonMessageError << ": " << outputJsonMessage.getErrorMsg();
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, errMsg.str(), Poco::Message::Priority::PRIO_ERROR));
  }
  return resultJsonMessage;
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::setNotificationFunctor(std::function<std::string(const std::string&)> notificationFunctor_)
{
  notificationExecutor.setNotificationFunctor(notificationFunctor_);
}
//-----------------------------------------------------------------------------
bool DRCEFunctionalObject::setHomeDir(const std::string& homeDir_)
{
  resetError();
  if (chdir(homeDir_.c_str())==0)
  {
    homeDir=homeDir_;
  }
  else
  {
    isError(true);
    setErrorCode(ERROR_SET_HOME_DIR);
    setErrorMsg(message(message_const::SET_HOME_DIR)+message(message_const::ERROR, strerror(errno)));
  }
  return !_isError;
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::resetError(void)
{
  isError(false);
  setErrorCode(NO_ERROR);
  setErrorMsg("");
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::setMaxThreadCount(unsigned int threadCount)
{
  asyncTaskQueue.setMaxThreadCount(threadCount);
}
//-----------------------------------------------------------------------------
unsigned int DRCEFunctionalObject::getMaxThreadCount(void) const
{
  return asyncTaskQueue.getMaxThreadCount();
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
void DRCEFunctionalObject::saveTaskQueue(void)
{
  asyncTaskQueue.saveTasksQueue();
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
void DRCEFunctionalObject::loadTaskQueue(void)
{
  asyncTaskQueue.loadTasksQueue();
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::setTasksQueueDumpPeriod(unsigned int periodMs)
{
  asyncTaskQueue.setTasksQueueDumpPeriod(periodMs);
}
//-----------------------------------------------------------------------------
unsigned int DRCEFunctionalObject::getTasksQueueDumpPeriod(void)
{
  return asyncTaskQueue.getTasksQueueDumpPeriod();
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
void DRCEFunctionalObject::setResourceMonitorTimePeriod(unsigned int timePeriod_)
{
  resourceMonitor.setTimePeriod(timePeriod_);
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
unsigned int DRCEFunctionalObject::getResourceMonitorTimePeriod(void)
{
  return resourceMonitor.getTimePeriod();
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
void DRCEFunctionalObject::setResetErrorCodeStateNotification(bool resetErrorCodeStateNotification_)
{
  asyncTaskQueue.setResetErrorCodeStateNotification(resetErrorCodeStateNotification_);
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
bool DRCEFunctionalObject::getResetErrorCodeStateNotification(void)
{
  return asyncTaskQueue.getResetErrorCodeStateNotification();
}
//-----------------------------------------------------------------------------
DRCEFunctionalObject::Dictionary DRCEFunctionalObject::dumpResourceMonitor(void)
{
  StatVariables statVariables;
  AvgCounts avgCounts;
  resourceMonitor.dump(statVariables, avgCounts);

  StatVariablesConverter statVariablesConverter(statVariables);
  Dictionary statVars = statVariablesConverter.toDictionary();

  AvgCountsConverter avgCountsConverter(avgCounts);
  Dictionary avgVars = avgCountsConverter.toDictionary();

  statVars.insert(avgVars.begin(), avgVars.end());
  return statVars;
}
//-----------------------------------------------------------------------------
void DRCEFunctionalObject::restoreResourceMonitor(const Dictionary& dictionary)
{
  StatVariablesConverter statVariablesConverter(dictionary);
  StatVariables statVariables = statVariablesConverter.toObject();

  AvgCountsConverter avgCountsConverter(dictionary);
  AvgCounts avgCounts = avgCountsConverter.toObject();

  resourceMonitor.restore(statVariables, avgCounts);
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
StatVariables DRCEFunctionalObject::getStatVariables(void)
{
  return resourceMonitor.getStatVariables();
}
//-----------------------------------------------------------------------------
size_t DRCEFunctionalObject::getAsyncTasksCount(void)
{
  return asyncTaskQueue.getAsyncTasksCount();
}
//-----------------------------------------------------------------------------
bool DRCEFunctionalObject::hasTask(unsigned int taskId)
{
  return asyncTaskQueue.hasTask(taskId);
}
//-----------------------------------------------------------------------------
std::string DRCEFunctionalObject::getCurrentTasksQueue(void)
{
  return asyncTaskQueue.getCurrentTasksQueueAsString();
}
//-----------------------------------------------------------------------------
std::string DRCEFunctionalObject::getListInfoAllTasks(void)
{
  std::string result;
  DRCEListAllTasks listAllTasks;
  if (listAllTasks.rebuild(getTasksStatusDir()))
  {
    result = listAllTasks.toString();
  }

  if (listAllTasks.isError())
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName,
        message(message_const::GET_LIST_ALL_TASKS_ERROR, listAllTasks.getErrorMsg()),
        Poco::Message::Priority::PRIO_ERROR));
  }
  return result;
}
//-----------------------------------------------------------------------------
std::string DRCEFunctionalObject::getHostResourceUsage(void)
{
  std::string result;
  HCE::drce::ResourceUsageExtractor::getInstance().update();
  try
  {
    HCE::drce::ResourceUsage resourceUsage = HCE::drce::ResourceUsageExtractor::getInstance().extract();
    HCE::drce::ResourceUsageSerializator resourceUsageSerializator(resourceUsage);
    resourceUsageSerializator.serialize(result);
    if (resourceUsageSerializator.isError())
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName,
            message(message_const::GET_RESOURCE_USAGE_ERROR, resourceUsageSerializator.getErrorMsg()),
            Poco::Message::Priority::PRIO_ERROR));
    }
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName,
            message(message_const::GET_RESOURCE_USAGE_ERROR, e.message()),
            Poco::Message::Priority::PRIO_ERROR));
  }
  return result;
}
//-----------------------------------------------------------------------------
std::string DRCEFunctionalObject::getDefaultJSON(const std::string& jsonErrorMessage, unsigned int jsonErrorCode)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::CREATE_DEFAULT_JSON), Poco::Message::Priority::PRIO_TRACE));

  DRCEDefaultJSON defaultJson;
  defaultJson.setJsonErrorMessage(jsonErrorMessage);
  defaultJson.setJsonErrorCode(jsonErrorCode);
  defaultJson.setJsonNodeName(nodeName);
  defaultJson.setJsonNodeHost(nodeHost);
  defaultJson.setJsonNodePort(nodePort);
  defaultJson.makeJSON();

  return defaultJson.getJSON();
}
//-----------------------------------------------------------------------------
std::string DRCEFunctionalObject::Process(const std::string& json)
{
  resetError();
  std::string resJson;

  try
  {
    DRCEInputJsonMessage inputJsonMessage(json);
    if (inputJsonMessage.isError())
      throw Poco::Exception(inputJsonMessage.getErrorMsg(), PARSE_ERROR);

    DRCEOutputJsonMessage outputMessage = Process(inputJsonMessage);

    if (!outputMessage.serialize(resJson))
      throw Poco::Exception(outputMessage.getErrorMsg(), outputMessage.getErrorCode());
  }
  catch(Poco::Exception& e)
  {
    isError(true);
    errorMsg = e.message();
    errorCode = e.code();

    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::ERROR_MSG, errorMsg)+
        message(message_const::ERROR_CODE, errorCode),
        Poco::Message::Priority::PRIO_ERROR));
  }
  return (resJson.empty())?getDefaultJSON(errorMsg, errorCode):resJson;
}
//-----------------------------------------------------------------------------
DRCEOutputJsonMessage DRCEFunctionalObject::Process(DRCEInputJsonMessage& inputJsonMessage)
{
  resetError();
  requestTime = 0;
  Poco::Timestamp tsStart;
  DRCEOutputJsonMessage outputMessage;
  DRCEResultData resultData;
  DRCEResultDataItem resultDataItem;

  if (!process(inputJsonMessage, resultDataItem))
  {
    resultDataItem = DRCETaskRequestExecutor::getResultDataItem(inputJsonMessage.getRequestId(),
        inputJsonMessage.getRequestType(),
        *this,
        DRCETaskRequest::TaskState::UNDEFINED);
  }
  resultData.addDataItem(std::forward<DRCEResultDataItem>(resultDataItem));
  outputMessage.setResultData(resultData);

  requestTime = getElapsedTimeMsec(tsStart);

  return outputMessage;
}
//-----------------------------------------------------------------------------
bool DRCEFunctionalObject::process(DRCEInputJsonMessage& inputJsonMessage, DRCEResultDataItem& resultDataItem)
{
  bool result = false;
  try
  {
    Poco::SharedPtr<DRCETaskRequest> pTaskRequest = inputJsonMessage.getTaskRequest();
    if (pTaskRequest.isNull())
      throw Poco::Exception(message(message_const::WRONG_TYPE_TASK_REQUEST));

    Poco::SharedPtr<DRCETaskRequestExecutor> pTaskRequestExecutor = DRCETaskRequestExecutorFactory::create(inputJsonMessage.getRequestType(),
        asyncTaskQueue, *this, message, resourceMonitor);

    if (pTaskRequestExecutor.isNull())
      throw Poco::Exception(message(message_const::WRONG_TYPE_TASK_REQUEST));

    pTaskRequest->setTaskId(inputJsonMessage.getRequestId());

    if (pTaskRequest->getRequestType() == DRCETaskRequest::RequestType::rtSetTaskExecute)
      applyEnvironments(pTaskRequest.get());

    resultDataItem = pTaskRequestExecutor->execute(pTaskRequest.get(), inputJsonMessage);

    result = true;
  }
  catch(Poco::Exception& e)
  {
    isError(true);
    setErrorCode(e.code());
    setErrorMsg(e.message());
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    isError(true);
    setErrorCode(ERROR_FUNCTIONAL_OBJECT);
    setErrorMsg(e.what());
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.what(), Poco::Message::Priority::PRIO_ERROR));
  }

  return result;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
} // namespace drce
} // namespace HCE