DRCECommonTask.cpp

Go to the documentation of this file.

#include <string.h>
#include <errno.h>
#include <list>
#include <Poco/Logger.h>
#include <Poco/File.h>

#include "DRCEMessageConst.hpp"
#include "DRCEEncodeDecodeBase64.hpp"
#include "DRCEError.hpp"
#include "DRCEFileStream.hpp"
#include "DRCECommonTask.hpp"
#include "DRCEFileExtractor.hpp"
#include "DRCEFileBuilder.hpp"
#include "DRCETaskRequestSetExecuteExecutor.hpp"

namespace HCE
{
namespace drce
{
//-----------------------------------------------------------------------------
DRCECommonTask::ApplyPidHandler::ApplyPidHandler(DRCEAsyncTasksQueue& asyncTasksQueue_, AsyncTasks& asyncTask_,
    DRCETaskRequest::TaskState allowedTaskState_, SessionOptions::ThreadMode threadMode_)
: asyncTasksQueue(asyncTasksQueue_), asyncTask(asyncTask_), allowedTaskState(allowedTaskState_), threadMode(threadMode_)
{
}
//-----------------------------------------------------------------------------
void DRCECommonTask::ApplyPidHandler::operator()(pid_t pid)
{
  if (threadMode == SessionOptions::ThreadMode::tmAsync)
  {
    if (asyncTasksQueue.isExistAsyncTask(asyncTask.taskId))
    {
      AsyncTasks task = asyncTasksQueue.getAsyncTask(asyncTask.taskId);
      if (task.state == allowedTaskState)
      {
        asyncTask.pid = pid;
        asyncTask.timeElapsed = DRCECommonTask::getElapsedTimeMsec(Poco::Timestamp::fromEpochTime(asyncTask.timeStart));
        asyncTasksQueue.safeChangeState(std::forward<AsyncTasks>(AsyncTasks(asyncTask)), nullptr);
      }
    }
  }
  else if (threadMode == SessionOptions::ThreadMode::tmSync)
  {
    asyncTask.pid = pid;
    asyncTask.timeElapsed = DRCECommonTask::getElapsedTimeMsec(Poco::Timestamp::fromEpochTime(asyncTask.timeStart));
  }
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
DRCECommonTask::DRCECommonTask(const std::string& taskName_, DRCENodeOptions& nodeOptions_, CustomMessage& message_,
    const DRCETaskRequestSetExecute& taskRequestSetExecute_, DRCEAsyncTasksQueue& asyncTasksQueue_,
    DRCEResourceMonitor& resourceMonitor_, const DRCEInputJsonMessage& inputJsonMessage_, unsigned int parentTaskId_)
: inherited(taskName_), nodeOptions(nodeOptions_), message(message_), taskRequestSetExecute(taskRequestSetExecute_),
  asyncTasksQueue(asyncTasksQueue_), resourceMonitor(resourceMonitor_), inputJsonMessage(inputJsonMessage_), parentTaskId(parentTaskId_)
{
}
//-----------------------------------------------------------------------------
void DRCECommonTask::makeFilesBefore(DRCEFilesList& filesList) throw (Poco::Exception)
{
  const size_t filesCount = filesList.getFilesCount();

  for (size_t i=0;i<filesCount;++i)
  {
    const std::string fileName = filesList.getFileItem(i).name;

    if ((filesList.getFileItem(i).actionType & FileItem::CREATE_BEFORE_EXEC_AND_WRITE_DATA)!=0)
    {
      std::ofstream ofs(fileName.c_str(), std::fstream::trunc|std::fstream::binary);
      if (!ofs)
        throw Poco::Exception(message(message_const::CREATE_FILE, fileName) , ERROR_CREATE_FILE_BEFORE_EXEC);

      ofs << (((filesList.getFileItem(i).actionType & FileItem::ENCODE_DECODE_CONTENT_BASE64)!=0)?DRCEDecodeBase64(filesList.getFileItem(i).data):filesList.getFileItem(i).data);
      ofs.close();
    }
    if ((filesList.getFileItem(i).actionType & FileItem::DELETE_BEFORE_EXEC)!=0)
    {
      if (remove(fileName.c_str())!=0)
        throw Poco::Exception(message(message_const::DELETE_FILE, fileName)+
            message(message_const::ERROR, strerror(errno)),
            ERROR_DELETE_FILE_BEFORE_EXEC);
    }
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::makeFilesAfter(DRCEFilesList& filesList, DRCEResultDataItem& resultDataItem)
{
  const size_t filesCount = filesList.getFilesCount();
  for (size_t i=0;i<filesCount;++i)
  {
    const std::string fileName = filesList.getFileItem(i).name;

    try
    {
      if ((filesList.getFileItem(i).actionType & FileItem::READ_AFTER_EXEC_AND_RETURN_RESULT)!=0)
      {
        std::ifstream ifs;
        ifs.open(fileName, std::fstream::binary);
        if ( (ifs.rdstate() & std::ifstream::failbit ) != 0 )
          throw Poco::Exception(message(message_const::READ_FILE, fileName), ERROR_READ_FILE_AFTER_EXEC);

        std::string fileContent;
        ifs.seekg(0, std::ios::end);
        fileContent.resize(ifs.tellg());
        ifs.seekg(0, std::ios::beg);
        ifs.read(const_cast<char*>(fileContent.c_str()), fileContent.size());
        ifs.close();

        FileItem fileItem;
        fileItem.actionType = filesList.getFileItem(i).actionType;
        fileItem.name = filesList.getFileItem(i).name;
        fileItem.data = (((filesList.getFileItem(i).actionType & FileItem::ENCODE_DECODE_CONTENT_BASE64)!=0)?DRCEEncodeBase64(fileContent):fileContent);
        resultDataItem.addFileItem(std::forward<FileItem>(fileItem));
      }
      if ((filesList.getFileItem(i).actionType & FileItem::DELETE_AFTER_EXEC)!=0)
      {
        if (remove(fileName.c_str())!=0)
          throw Poco::Exception(message(message_const::DELETE_FILE, fileName)+
              message(message_const::ERROR, strerror(errno)),
              ERROR_DELETE_FILE_AFTER_EXEC);
      }
    }
    catch(Poco::Exception& e)
    {
      std::stringstream errMsg(resultDataItem.getErrorMessage());
      if (!errMsg.str().empty())
        errMsg << ", ";
      errMsg << e.message();
      resultDataItem.setErrorMessage(errMsg.str());

      resultDataItem.setErrorCode(e.code());
    }
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::cleanup(bool needCleanup)
{
  cleanup(inputJsonMessage, nodeOptions, message, needCleanup);
}
//-----------------------------------------------------------------------------
void DRCECommonTask::cleanup(unsigned int taskId, DRCENodeOptions& nodeOptions, CustomMessage& message, bool needCleanup)
{
  Poco::File requetFile(FileStream::getRequestFileName(nodeOptions.getTasksDataDir(), taskId));
  if (requetFile.exists())
  {
    DRCEInputJsonMessage inputJsonMessage = DRCECommonTask::getRequestData(taskId, nodeOptions, message);
    DRCECommonTask::cleanup(inputJsonMessage, nodeOptions, message, needCleanup);
  }
  else
  {
    if (needCleanup)
    {
      DRCECommonTask::safeCleanup(FileStream::getDataFileName(nodeOptions.getTasksDataDir(), taskId));
      DRCECommonTask::safeCleanup(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), taskId));
      DRCECommonTask::safeCleanup(FileStream::getRequestFileName(nodeOptions.getTasksDataDir(), taskId));
    }
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::cleanup(DRCEInputJsonMessage& inputJsonMessage, DRCENodeOptions& nodeOptions, CustomMessage& message, bool needCleanup)
{
  try
  {
    DRCETaskRequestSetExecute* pTaskRequestSetExecute = dynamic_cast<DRCETaskRequestSetExecute*>(inputJsonMessage.getTaskRequest().get());
    if (pTaskRequestSetExecute)
    {
      if (!needCleanup && pTaskRequestSetExecute->getSessionOptions().cleanup == SessionOptions::CleanupFlag::cfDelete)
        needCleanup = true;

      if (needCleanup)
      {
        for (size_t i=0;i<inputJsonMessage.getSubtasksCount();++i)
        {
          DRCEInputJsonMessage inputMessage = inputJsonMessage.getSubtaskItem(i);
          DRCECommonTask::cleanup(inputMessage, nodeOptions, message, needCleanup);
        }
        DRCECommonTask::safeCleanup(FileStream::getDataFileName(nodeOptions.getTasksDataDir(), inputJsonMessage.getRequestId()));
        DRCECommonTask::safeCleanup(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), inputJsonMessage.getRequestId()));
        DRCECommonTask::safeCleanup(FileStream::getRequestFileName(nodeOptions.getTasksDataDir(), inputJsonMessage.getRequestId()));
        const DRCEFilesList& filesList = static_cast<const DRCEFilesList&>(*pTaskRequestSetExecute);
        for (size_t i=0;i<filesList.getFilesCount();++i)
        {
          DRCECommonTask::safeCleanup(filesList.getFileItem(i).name);
        }
      }
    }
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::safeCleanup(const std::string& path)
{
  try
  {
    Poco::File file(path);
    if (file.exists())
    {
      file.remove(true);

      std::stringstream outMsg;
      outMsg << "DRCECommonTask::safeCleanup  path: " << path;
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));
    }
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
size_t DRCECommonTask::getElapsedTimeMsec(const Poco::Timestamp& tsStart)
{
  Poco::Timestamp tsStop;
  Poco::Timestamp::TimeDiff diff = tsStop-tsStart;
  return static_cast<size_t>(diff/1000);
}
//-----------------------------------------------------------------------------
void DRCECommonTask::executeSubtasks(DRCEResultDataItem& resultDataItem)
{
  std::stringstream outMsg;
  outMsg << "DRCECommonTask::executeSubtasks  enter  taskId: " << resultDataItem.getRequestId();
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

  if (inputJsonMessage.getSubtasksCount())
  {
    DRCETaskRequestSetExecuteExecutor taskRequestSetExecuteExecutor(asyncTasksQueue, nodeOptions, message, resourceMonitor);
    for (size_t i=0;i<inputJsonMessage.getSubtasksCount();++i)
    {
      Poco::SharedPtr<DRCETaskRequest> pTaskRequest = inputJsonMessage.getSubtaskItem(i).getTaskRequest();
      if (!pTaskRequest.isNull() && !isCancelled())
      {
        DRCEInputJsonMessage inputMessage = inputJsonMessage.getSubtaskItem(i);
        pTaskRequest->setRequestType(inputMessage.getRequestType());
        pTaskRequest->setTaskId(inputMessage.getRequestId());
        pTaskRequest->setParentTaskId(inputJsonMessage.getRequestId());

        DRCEResultDataItem resultItem = taskRequestSetExecuteExecutor.execute(pTaskRequest.get(), inputMessage);
        resultDataItem.addSubtaskItem(std::forward<DRCEResultDataItem>(resultItem));
      }
    }
  }

  if (!isCancelled())
  {
    try
    {
      AsyncTaskLocker lock(asyncTasksQueue, resultDataItem.getRequestId());
      DataFileBuilder dataFileBuilder(message, nodeOptions.getTasksDataDir());
      DRCEResultData resultData;
      resultData.addDataItem(resultDataItem);
      dataFileBuilder.build(resultData);
    }
    catch(Poco::Exception& e)
    {
      std::stringstream outMsg;
      outMsg << "DRCECommonTask::executeSubtasks dataFileBuilder.build " << e.displayText();
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_ERROR));
      throw;
    }
    catch(std::exception& e)
    {
      std::stringstream outMsg;
      outMsg << "DRCECommonTask::executeSubtasks dataFileBuilder.build " << e.what();
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_ERROR));
      throw;
    }
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::updateTasks(DRCEResultDataItem& resultDataItem,
    DRCENodeOptions& nodeOptions,
    CustomMessage& message,
    DRCEAsyncTasksQueue& asyncTasksQueue,
    unsigned int& progressCount,
    DRCECommonTask* pCommonTask)
{
  const std::string dataFileName = FileStream::getDataFileName(nodeOptions.getTasksDataDir(), resultDataItem.getRequestId());
  Poco::File dataFile(dataFileName);
  if (dataFile.exists())
  {
    AsyncTaskLocker lock(asyncTasksQueue, resultDataItem.getRequestId());
    try
    {
      std::stringstream outMsg;
      outMsg << "DRCECommonTask::updateTasks  taskId: " << resultDataItem.getRequestId();
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

      try
      {
        DataFileExtractor dataFileExtractor(message);
        resultDataItem = dataFileExtractor.extract(dataFileName);
      }
      catch(Poco::Exception& e)
      {
        std::stringstream outMsg;
        outMsg << "DRCECommonTask::updateTasks  dataFileExtractor.extract: " << e.displayText();
        Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_DEBUG));
      }
      catch(std::exception& e)
      {
        std::stringstream outMsg;
        outMsg << "DRCECommonTask::updateTasks  dataFileExtractor.extract: " << e.what();
        Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_DEBUG));
      }

      if (resultDataItem.getState()==static_cast<unsigned int>(DRCETaskRequest::TaskState::SET_AS_NEW) ||
          resultDataItem.getState()==static_cast<unsigned int>(DRCETaskRequest::TaskState::IN_PROGRESS) ||
          resultDataItem.getState()==static_cast<unsigned int>(DRCETaskRequest::TaskState::QUEUED_TO_RUN))
          ++progressCount;

      // restart task if it continue hold in queue to run
      if (resultDataItem.getState()==static_cast<unsigned int>(DRCETaskRequest::TaskState::QUEUED_TO_RUN))
      {
        outMsg.str("");
        outMsg << "DRCECommonTask::updateTasks  pCommonTask->executeSubtasks  taskId: " << resultDataItem.getRequestId();
        Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

        if (pCommonTask)
          pCommonTask->executeSubtasks(resultDataItem);
      }

      for (size_t i=0;i<resultDataItem.getSubtasksCount();++i)
      {
        DRCEResultDataItem resultItem(resultDataItem.getSubtaskItem(i));
        unsigned int subtasksProgressCount = 0;
        updateTasks(resultItem, nodeOptions, message, asyncTasksQueue, subtasksProgressCount, pCommonTask);
        resultDataItem.setSubtaskItem(i, std::forward<DRCEResultDataItem>(resultItem));
      }

      DRCEResultData resultData;
      resultData.addDataItem(resultDataItem);
      DataFileBuilder dataFileBuilder(message, nodeOptions.getTasksDataDir());
      dataFileBuilder.build(resultData);

      StatusFileBuilder statusFileBuilder(message, nodeOptions.getTasksStatusDir());
      statusFileBuilder.build(resultData);
    }
    catch(Poco::Exception& e)
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::ERROR, e.displayText()), Poco::Message::Priority::PRIO_ERROR));
    }
    catch(std::exception& e)
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, message(message_const::ERROR, e.what()), Poco::Message::Priority::PRIO_ERROR));
    }
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::updateTasks(DRCEResultDataItem& resultDataItem, unsigned int& progressCount)
{
  updateTasks(resultDataItem, nodeOptions, message, asyncTasksQueue, progressCount, this);
}
//-----------------------------------------------------------------------------
void DRCECommonTask::getSubtasksList(DRCEInputJsonMessage& inputJsonMessage, SubtasksList& subtasks)
{
  if (inputJsonMessage.getRequestId())
  {
    DRCETaskRequestSetExecute* pTaskRequestSetExecute = dynamic_cast<DRCETaskRequestSetExecute*>(inputJsonMessage.getTaskRequest().get());
    if (pTaskRequestSetExecute)
    {
      subtasks.push_back(std::make_pair(inputJsonMessage.getRequestId(), pTaskRequestSetExecute->getSessionOptions().cleanup));

      std::stringstream outMsg;
      outMsg << "DRCECommonTask::getSubtasksList  taskId: " << inputJsonMessage.getRequestId()
          << " cleanup: " << std::boolalpha << (pTaskRequestSetExecute->getSessionOptions().cleanup==SessionOptions::CleanupFlag::cfDelete);
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

      for (size_t i=0;i<inputJsonMessage.getSubtasksCount();++i)
      {
        DRCEInputJsonMessage inputMessage(inputJsonMessage.getSubtaskItem(i));
        getSubtasksList(inputMessage, subtasks);
      }
    }
  }
}
//-----------------------------------------------------------------------------
DRCEInputJsonMessage DRCECommonTask::getRequestData(unsigned int taskId, DRCENodeOptions& nodeOptions, CustomMessage& message)
{
  DRCEInputJsonMessage inputJsonMessage;
  try
  {
    Poco::File requestFile(FileStream::getRequestFileName(nodeOptions.getTasksDataDir(), taskId));
    if (requestFile.exists())
    {
      RequestFileExtractor requestFileExtractor(message);
      inputJsonMessage = requestFileExtractor.extract(requestFile.path());
    }
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.what(), Poco::Message::Priority::PRIO_ERROR));
  }
  return inputJsonMessage;
}
//-----------------------------------------------------------------------------
void DRCECommonTask::waitUpdateAllTasks(unsigned int taskId,
    DRCENodeOptions& nodeOptions,
    CustomMessage& message,
    DRCEAsyncTasksQueue& asyncTasksQueue,
    Poco::Task* pTask)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCECommonTask::waitUpdateAllTasks enter", Poco::Message::Priority::PRIO_TRACE));

  const std::string dataFileName = FileStream::getDataFileName(nodeOptions.getTasksDataDir(), taskId);
  Poco::File dataFile(dataFileName);

  std::stringstream outMsg;
  outMsg << "DRCECommonTask::waitUpdateAllTasks  '" << dataFileName << "' isExist = " << std::boolalpha << dataFile.exists() << std::endl;
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

  if (dataFile.exists())
  {
    DataFileExtractor dataFileExtractor(message);
    try
    {
      DRCEResultDataItem resultDataItem = dataFileExtractor.extract(dataFileName);
      unsigned int progressCount = 0;
      DRCECommonTask::updateTasks(resultDataItem, nodeOptions, message, asyncTasksQueue, progressCount);
      while (progressCount)
      {
        progressCount = 0;
        Poco::Thread::sleep(1000);
        DRCECommonTask::updateTasks(resultDataItem, nodeOptions, message, asyncTasksQueue, progressCount);


        std::stringstream outMsg;
        outMsg << "DRCECommonTask::waitUpdateAllTasks  taskId: " << resultDataItem.getRequestId() << " progressCount: " << progressCount;
        Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

        if (pTask)
          if (pTask->isCancelled())
            break;
      }
    }
    catch(Poco::Exception& e)
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_DEBUG));
    }
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::waitEndAllTask(DRCEInputJsonMessage& inputJsonMessage, DRCEAsyncTasksQueue& asyncTasksQueue, Poco::Task* pTask)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCECommonTask::waitEndAllTask enter", Poco::Message::Priority::PRIO_TRACE));

  SubtasksList subtasks;
  getSubtasksList(inputJsonMessage, subtasks);

  for (SubtasksList::iterator iter=subtasks.begin();iter!=subtasks.end();++iter)
  {
    if ((*iter).first == inputJsonMessage.getRequestId())
    {
      subtasks.erase(iter);
      break;
    }
  }

  if (!subtasks.empty())
  {
    bool needWait = true;
    while (needWait)
    {
      for (size_t i=0;i<subtasks.size();++i)
      {
        if (asyncTasksQueue.isExistAsyncTask(subtasks[i].first))
        {
          std::stringstream outMsg;
          outMsg << "DRCECommonTask::waitEndAllTask (" << i << ") asyncTasksQueue.isExistAsyncTask(" <<subtasks[i].first <<")";
          Poco::Logger::root().log(Poco::Message(drce_const::moduleName, outMsg.str(), Poco::Message::Priority::PRIO_TRACE));

          break;
        }
        if ((i+1)==subtasks.size())
          return;
      }
      Poco::Thread::sleep(1000);
      if (pTask)
        if (pTask->isCancelled())
          break;
    }
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::saveRequestData(void)
{
  saveRequestData(inputJsonMessage, nodeOptions, message);
}
//-----------------------------------------------------------------------------
void DRCECommonTask::saveResultData(DRCEResultDataItem& resultDataItem)
{
  saveResultData(resultDataItem, nodeOptions, message, asyncTasksQueue);
}
//-----------------------------------------------------------------------------
void DRCECommonTask::saveRequestData(DRCEInputJsonMessage& inputJsonMessage, DRCENodeOptions& nodeOptions, CustomMessage& message)
{
  try
  {
    RequestFileBuilder requestFileBuilder(message, nodeOptions.getTasksDataDir());
    requestFileBuilder.build(inputJsonMessage);
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.what(), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::saveRequest(unsigned int taskId, DRCENodeOptions& nodeOptions, CustomMessage& message)
{
  try
  {
    Poco::File requestFile(FileStream::getRequestFileName(nodeOptions.getTasksDataDir(), taskId));
    if (requestFile.exists())
    {
      RequestFileExtractor requestFileExtractor(message);
      DRCEInputJsonMessage inputJsonMessage = requestFileExtractor.extract(requestFile.path());

      saveRequest(inputJsonMessage, nodeOptions, message);
    }
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.what(), Poco::Message::Priority::PRIO_ERROR));
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::saveRequest(DRCEInputJsonMessage& inputJsonMessage, DRCENodeOptions& nodeOptions, CustomMessage& message)
{
  try
  {
    RequestFileBuilder requestFileBuilder(message, nodeOptions.getTasksDataDir());
    requestFileBuilder.build(inputJsonMessage);
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.what(), Poco::Message::Priority::PRIO_ERROR));
  }

  for (size_t i=0;i<inputJsonMessage.getSubtasksCount();++i)
  {
    DRCEInputJsonMessage inputMessage(inputJsonMessage.getSubtaskItem(i));
    saveRequest(inputMessage, nodeOptions, message);
  }
}
//-----------------------------------------------------------------------------
void DRCECommonTask::saveResultData(DRCEResultDataItem& resultDataItem,  DRCENodeOptions& nodeOptions, CustomMessage& message, DRCEAsyncTasksQueue& asyncTasksQueue)
{
  Poco::Logger::root().log(Poco::Message(drce_const::moduleName, "DRCECommonTask::saveResultData enter", Poco::Message::Priority::PRIO_TRACE));

  try
  {
    AsyncTaskLocker lock(asyncTasksQueue, resultDataItem.getRequestId());

    DRCEResultData resultData;
    resultData.addDataItem(resultDataItem);
    DataFileBuilder dataFileBuilder(message, nodeOptions.getTasksDataDir());
    dataFileBuilder.build(resultData);

    StatusFileBuilder statusFileBuilder(message, nodeOptions.getTasksStatusDir());
    statusFileBuilder.build(resultData);
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
  catch(std::exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.what(), Poco::Message::Priority::PRIO_ERROR));
  }

  for (size_t i=0;i<resultDataItem.getSubtasksCount();++i)
  {
    DRCEResultDataItem resultItem(resultDataItem.getSubtaskItem(i));
    saveResultData(resultItem, nodeOptions, message, asyncTasksQueue);
  }
}
//-----------------------------------------------------------------------------
DRCEResultDataItem DRCECommonTask::makeResultDataItem(DRCEInputJsonMessage& inputJsonMessage, DRCENodeOptions& nodeOptions, CustomMessage& message,
    DRCETaskRequest::RequestType requestType, DRCETaskRequest::TaskState taskState, unsigned int errorCode, const std::string& errorMessage,
    unsigned int pid, size_t timeElapsed, unsigned int exitStatus)
{
  DRCEResultDataItem resultDataItem;

  for (size_t i=0;i<inputJsonMessage.getSubtasksCount();++i)
  {
    DRCEInputJsonMessage inputMessage(inputJsonMessage.getSubtaskItem(i));
    DRCEResultDataItem resultItem = makeResultDataItem(inputMessage, nodeOptions, message, requestType, taskState, errorCode, errorMessage, 0, timeElapsed, exitStatus);
    resultDataItem.addSubtaskItem(resultItem);
  }

  resultDataItem.setTime(timeElapsed);
  resultDataItem.setPid(pid);
  resultDataItem.setErrorCode(errorCode);
  resultDataItem.setErrorMessage(errorMessage);
  resultDataItem.setRequestId(inputJsonMessage.getRequestId());
  resultDataItem.setRequestType(requestType);
  resultDataItem.setState(static_cast<unsigned int>(taskState));
  resultDataItem.setExitStatus(exitStatus);

  Poco::File statusFile(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), inputJsonMessage.getRequestId()));
  if (statusFile.exists())
  {
    try
    {
      DataFileExtractor dataFileExtractor(message);
      DRCEResultDataItem resultItem = dataFileExtractor.extract(statusFile.path());

      resultDataItem.setPid(resultItem.getPid());
      resultDataItem.setTime(resultItem.getTime());
      resultDataItem.setRequestId(resultItem.getRequestId());
      resultDataItem.setRequestType(requestType);
      resultDataItem.setState(static_cast<unsigned int>(taskState));
    }
    catch(Poco::Exception& e)
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_TRACE));
    }
  }

  resultDataItem.setNodeName(nodeOptions.getNodeName());
  resultDataItem.setNodeHost(nodeOptions.getNodeHost());
  resultDataItem.setNodePort(nodeOptions.getNodePort());

  return resultDataItem;
}
//-----------------------------------------------------------------------------
unsigned int DRCECommonTask::getHostParentTaskId(unsigned int taskId, DRCEAsyncTasksQueue& asyncTasksQueue)
{
  unsigned int parentTaskId = taskId;

  try
  {
    if (asyncTasksQueue.isExistAsyncTask(taskId))
    {
      AsyncTasks asyncTask = asyncTasksQueue.getAsyncTask(taskId);
      if (asyncTask.parentTaskId)
        parentTaskId = getHostParentTaskId(asyncTask.parentTaskId, asyncTasksQueue);
      else
        parentTaskId = asyncTask.taskId;
    }
  }
  catch(Poco::Exception& e)
  {
    Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
  }
  return parentTaskId;
}
//-----------------------------------------------------------------------------
DRCEResultDataItem DRCECommonTask::extractResultDataItem(const std::string& fileName)
{
  DRCEResultDataItem resultDataItem;
  Poco::File dataFile(fileName);
  if (dataFile.exists())
  {
    try
    {
      MessagesCollection messagesCollection(message_const::messages);
      CustomMessage message(messagesCollection);
      DataFileExtractor dataFileExctractor(message);
      resultDataItem = dataFileExctractor.extract(dataFile.path());
    }
    catch(Poco::Exception& e)
    {
      Poco::Logger::root().log(Poco::Message(drce_const::moduleName, e.displayText(), Poco::Message::Priority::PRIO_ERROR));
    }
  }
  return resultDataItem;
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
DRCEResultDataItem DRCECommonTask::extractStatusResultDataItem(unsigned int taskId,
    DRCENodeOptions& nodeOptions,
    DRCEAsyncTasksQueue& asyncTasksQueue)
{
  AsyncTaskLocker lock(asyncTasksQueue, taskId);
  return extractResultDataItem(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), taskId));
}
//-----------------------------------------------------------------------------
// cppcheck-suppress unusedFunction
DRCEResultDataItem DRCECommonTask::extractDataResultDataItem(unsigned int taskId,
    DRCENodeOptions& nodeOptions,
    DRCEAsyncTasksQueue& asyncTasksQueue)
{
  AsyncTaskLocker lock(asyncTasksQueue, taskId);
  return extractResultDataItem(FileStream::getDataFileName(nodeOptions.getTasksDataDir(), taskId));
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
} // end namespace drce 
} // end namespace HCE