DRCETaskRequestGetDataExecutor.cpp

Go to the documentation of this file.

#include <string.h>
#include <errno.h>

#include "DRCEFileStream.hpp"
#include "DRCEOutputJsonMessage.hpp"
#include "DRCEResultDataSerializator.hpp"
#include "DRCEFileExtractor.hpp"

#include "DRCETaskRequestGetDataExecutor.hpp"
#include "DRCETaskRequestGetData.hpp"

namespace HCE
{
namespace drce
{
//-----------------------------------------------------------------------------
DRCETaskRequestGetDataExecutor::DRCETaskRequestGetDataExecutor(DRCEAsyncTasksQueue& asyncTaskQueue_,
    DRCENodeOptions& nodeOptions_, CustomMessage& message_, DRCEResourceMonitor& resourceMonitor_)
: inherited(asyncTaskQueue_, nodeOptions_, message_, resourceMonitor_)
{
}
//-----------------------------------------------------------------------------
DRCEResultDataItem DRCETaskRequestGetDataExecutor::execute(DRCETaskRequest* pTaskRequest, DRCEInputJsonMessage& inputJsonMessage)
{
  DRCEResultDataItem resultDataItem = getResultDataItem(*pTaskRequest, DRCETaskRequest::TaskState::UNDEFINED);

  try
  {
    DRCETaskRequestGetData* pTaskRequestGetData = dynamic_cast<DRCETaskRequestGetData*>(pTaskRequest);
    if (!pTaskRequestGetData)
      throw Poco::Exception(message(message_const::WRONG_TYPE_TASK_REQUEST), ERROR_BAD_COMMAND_PARAMS);

    if (pTaskRequestGetData->getFetchType()!=DRCETaskRequestGetData::FetchType::ftDeleteDataAfterFetch &&
        pTaskRequestGetData->getFetchType()!=DRCETaskRequestGetData::FetchType::ftNotDeleteDataAfterFetch)
      throw Poco::Exception(message(message_const::WRONG_FETCH_TYPE, static_cast<unsigned int>(pTaskRequestGetData->getFetchType())), ERROR_BAD_COMMAND_PARAMS);

    if (asyncTaskQueue.isExistAsyncTask(pTaskRequestGetData->getTaskId()))
    {
      DRCETaskRequest::TaskState taskState = asyncTaskQueue.getAsyncTask(pTaskRequestGetData->getTaskId()).state;
      if (taskState == DRCETaskRequest::TaskState::IN_PROGRESS ||
          taskState == DRCETaskRequest::TaskState::SET_AS_NEW ||
          taskState == DRCETaskRequest::TaskState::QUEUED_TO_RUN)
      {
        resultDataItem.setState(taskState);
      }
    }
    else
    {
      const std::string dataFileName =  FileStream::getDataFileName(nodeOptions.getTasksDataDir(), pTaskRequestGetData->getTaskId());

      resultDataItem.setState(static_cast<unsigned int>(DRCETaskRequest::TaskState::NOT_FOUND));

      AsyncTaskLocker lock(asyncTaskQueue, pTaskRequestGetData->getTaskId());
      // read data file content
      DataFileExtractor dataFileExtractor(message);
      resultDataItem = dataFileExtractor.extract(dataFileName);

      // delete data file after fetch if need
      if (pTaskRequestGetData->getFetchType() == DRCETaskRequestGetData::FetchType::ftDeleteDataAfterFetch)
      {
        DRCECommonTask::cleanup(pTaskRequestGetData->getTaskId(), nodeOptions, message, true);
      }
    }
  }
  catch(Poco::Exception& e)
  {
    resultDataItem.setErrorCode(e.code());
    resultDataItem.setErrorMessage(e.message());
    resultDataItem.setState(static_cast<unsigned int>(DRCETaskRequest::TaskState::CRASHED));
  }
  catch(std::exception& e)
  {
    resultDataItem.setErrorCode(ERROR_EXECUTE_TASK);
    resultDataItem.setErrorMessage(e.what());
    resultDataItem.setState(static_cast<unsigned int>(DRCETaskRequest::TaskState::CRASHED));
  }
  return resultDataItem;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
} // end namespace drce 
} // end namespace HCE