From 092e18244249340d4dff28e76854a8c642039282 Mon Sep 17 00:00:00 2001
From: Marco De Marco <demarco@oats.inaf.it>
Date: Mon, 17 Nov 2014 15:48:20 +0100
Subject: [PATCH] Ignore files, pre and post process added
---
src/Configuration.h | 14 +++---
src/PreProcessor.cpp | 22 +++-------
src/PreProcessor.h | 4 --
src/PreProcessor.xmi | 5 ---
src/PreProcessorClass.cpp | 14 ------
src/ScriptManager.cpp | 56 ++++++++++++------------
src/WorkerThread.cpp | 92 ++++++++++++++++++++++-----------------
7 files changed, 90 insertions(+), 117 deletions(-)
diff --git a/src/Configuration.h b/src/Configuration.h
index 52f9cd1..a2f1756 100644
--- a/src/Configuration.h
+++ b/src/Configuration.h
@@ -23,12 +23,12 @@ private:
//------------------------------------------------------------------------------
Configuration(std::string watchPath, std::string regularPath,
std::string warningPath, std::string errorPath, std::string scriptPath,
- int workerNumber, int sleepTime, int waitTime, uint32_t iNotifyMask,
- bool ignoreTimeout) : m_watchPath(watchPath), m_regularPath(regularPath),
+ int workerNumber, int sleepTime, int waitTime, uint32_t iNotifyMask) :
+ m_watchPath(watchPath), m_regularPath(regularPath),
m_warningPath(warningPath), m_errorPath(errorPath),
m_scriptPath(scriptPath), m_workerNumber(workerNumber),
m_sleepTime(sleepTime), m_waitTime(waitTime),
- m_iNotifyMask(iNotifyMask), m_ignoreTimeout(ignoreTimeout) { }
+ m_iNotifyMask(iNotifyMask) { }
virtual ~Configuration() {}
@@ -47,11 +47,11 @@ public:
static Configuration::SP create(std::string watchPath,
std::string regularPath, std::string warningPath, std::string errorPath,
std::string scriptPath, int workerNumber, int sleepTime, int waitTime,
- uint32_t iNotifyMask, bool ignoreTimeout)
+ uint32_t iNotifyMask)
{
Configuration::SP c_sp(new Configuration(watchPath, regularPath,
warningPath, errorPath, scriptPath, workerNumber, sleepTime,
- waitTime, iNotifyMask, ignoreTimeout), Configuration::Deleter());
+ waitTime, iNotifyMask), Configuration::Deleter());
return c_sp;
}
@@ -65,7 +65,6 @@ public:
unsigned int getSleepTime() const { return m_sleepTime; }
unsigned int getWaitTime() const { return m_waitTime; }
uint32_t getINotifyMask() const { return m_iNotifyMask; }
- bool getIgnoreTimeout() const { return m_ignoreTimeout; }
private:
//------------------------------------------------------------------------------
@@ -97,9 +96,6 @@ private:
//INotify mask
const uint32_t m_iNotifyMask;
-
- //Ignore timeout
- const bool m_ignoreTimeout;
};
} //End of namespace
diff --git a/src/PreProcessor.cpp b/src/PreProcessor.cpp
index 07fc4ea..d4c887a 100644
--- a/src/PreProcessor.cpp
+++ b/src/PreProcessor.cpp
@@ -229,7 +229,6 @@ void PreProcessor::get_device_property()
dev_prop.push_back(Tango::DbDatum("SleepTime"));
dev_prop.push_back(Tango::DbDatum("WaitTime"));
dev_prop.push_back(Tango::DbDatum("WorkerNumber"));
- dev_prop.push_back(Tango::DbDatum("IgnoreTimeout"));
dev_prop.push_back(Tango::DbDatum("AutoStart"));
@@ -345,17 +344,6 @@ void PreProcessor::get_device_property()
// And try to extract WorkerNumber value from database
if (dev_prop[i].is_empty()==false) dev_prop[i] >> workerNumber;
- // Try to initialize IgnoreTimeout from class property
- cl_prop = ds_class->get_class_property(dev_prop[++i].name);
- if (cl_prop.is_empty()==false) cl_prop >> ignoreTimeout;
- else {
- // Try to initialize IgnoreTimeout from default device value
- def_prop = ds_class->get_default_device_property(dev_prop[i].name);
- if (def_prop.is_empty()==false) def_prop >> ignoreTimeout;
- }
- // And try to extract IgnoreTimeout value from database
- if (dev_prop[i].is_empty()==false) dev_prop[i] >> ignoreTimeout;
-
// Try to initialize AutoStart from class property
cl_prop = ds_class->get_class_property(dev_prop[++i].name);
if (cl_prop.is_empty()==false) cl_prop >> autoStart;
@@ -408,17 +396,17 @@ void PreProcessor::get_device_property()
const uint32_t inotifyMask = create_inotify_mask(eventList);
- if(sleepTime<MIN_SLEEP_TIME || sleepTime>MAX_SLEEP_TIME)
+ if(sleepTime < MIN_SLEEP_TIME || sleepTime > MAX_SLEEP_TIME)
throw(invalid_argument("SleepTime property is out of range or not defined"));
- if(waitTime>MAX_WAIT_TIME)
+ if(waitTime > MAX_WAIT_TIME)
throw(invalid_argument("WaitTime property is out of range or not defined"));
- if(workerNumber<1 || workerNumber>MAX_WORKER_NUMBER)
+ if(workerNumber < 1 || workerNumber > MAX_WORKER_NUMBER)
throw(invalid_argument("WorkerNumber property out of range or not defined"));
m_configuration_sp = Configuration::create(watchPath, regularPath, warningPath,
- errorPath, scriptPath, workerNumber, sleepTime, waitTime, inotifyMask, ignoreTimeout);
+ errorPath, scriptPath, workerNumber, sleepTime, waitTime, inotifyMask);
}
catch(invalid_argument& ex)
{
@@ -439,7 +427,7 @@ void PreProcessor::get_device_property()
//--------------------------------------------------------
void PreProcessor::always_executed_hook()
{
- INFO_STREAM << "PreProcessor::always_executed_hook() " << device_name << endl;
+ DEBUG_STREAM << "PreProcessor::always_executed_hook() " << device_name << endl;
/*----- PROTECTED REGION ID(PreProcessor::always_executed_hook) ENABLED START -----*/
if(get_state() != Tango::FAULT)
diff --git a/src/PreProcessor.h b/src/PreProcessor.h
index 3cc9f13..791a26b 100644
--- a/src/PreProcessor.h
+++ b/src/PreProcessor.h
@@ -120,10 +120,6 @@ public:
Tango::DevUShort waitTime;
// WorkerNumber:
Tango::DevUShort workerNumber;
- // IgnoreTimeout: Behaviour for missing end of file after timeout case:
- // true = process those files like regulars files
- // false = process those files like erroneous files
- Tango::DevBoolean ignoreTimeout;
// AutoStart: Exec On command after init if state is not fault
Tango::DevBoolean autoStart;
diff --git a/src/PreProcessor.xmi b/src/PreProcessor.xmi
index 223886e..29d9504 100644
--- a/src/PreProcessor.xmi
+++ b/src/PreProcessor.xmi
@@ -41,11 +41,6 @@
<type xsi:type="pogoDsl:UShortType"/>
<status abstract="false" inherited="false" concrete="true" concreteHere="true"/>
</deviceProperties>
- <deviceProperties name="IgnoreTimeout" description="Behaviour for missing end of file after timeout case: 
true = process those files like regulars files
false = process those files like erroneous files">
- <type xsi:type="pogoDsl:BooleanType"/>
- <status abstract="false" inherited="false" concrete="true" concreteHere="true"/>
- <DefaultPropValue>false</DefaultPropValue>
- </deviceProperties>
<deviceProperties name="AutoStart" description="Exec On command after init if state is not fault">
<type xsi:type="pogoDsl:BooleanType"/>
<status abstract="false" inherited="false" concrete="true" concreteHere="true"/>
diff --git a/src/PreProcessorClass.cpp b/src/PreProcessorClass.cpp
index 8b9e200..3d6fadd 100644
--- a/src/PreProcessorClass.cpp
+++ b/src/PreProcessorClass.cpp
@@ -395,20 +395,6 @@ void PreProcessorClass::set_default_property()
dev_def_prop.push_back(data);
add_wiz_dev_prop(prop_name, prop_desc, prop_def);
}
- else
- add_wiz_dev_prop(prop_name, prop_desc);
- prop_name = "IgnoreTimeout";
- prop_desc = "Behaviour for missing end of file after timeout case: \ntrue = process those files like regulars files\nfalse = process those files like erroneous files";
- prop_def = "false";
- vect_data.clear();
- vect_data.push_back("false");
- if (prop_def.length()>0)
- {
- Tango::DbDatum data(prop_name);
- data << vect_data ;
- dev_def_prop.push_back(data);
- add_wiz_dev_prop(prop_name, prop_desc, prop_def);
- }
else
add_wiz_dev_prop(prop_name, prop_desc);
prop_name = "AutoStart";
diff --git a/src/ScriptManager.cpp b/src/ScriptManager.cpp
index 1a09932..5c785f2 100644
--- a/src/ScriptManager.cpp
+++ b/src/ScriptManager.cpp
@@ -1,6 +1,6 @@
#include <ScriptManager.h>
-#include <cstdio>
+#include <stdio.h>
namespace PreProcessor_ns
{
@@ -56,8 +56,12 @@ void ScriptManager::checkScriptCompliance()
std::string result = exec(command.str());
- if(result.find("CHECK OK") == std::string::npos)
- throw std::runtime_error("Invalid script");
+ if(result.find("CHECK OK") == std::string::npos)
+ {
+ std::stringstream errorStream;
+ errorStream << "Check script error: " << result;
+ throw std::runtime_error(errorStream.str());
+ }
}
//==============================================================================
@@ -70,7 +74,7 @@ bool ScriptManager::isFileNameValid(boost::filesystem::path& filePath)
std::stringstream command;
command << m_configuration_sp->getScriptPath()
- << " VALID " << filePath.string();
+ << " VALID " << filePath.string();
std::string result = exec(command.str());
@@ -78,7 +82,7 @@ bool ScriptManager::isFileNameValid(boost::filesystem::path& filePath)
{
return true;
}
- else if(result.find("VALID FATAL") != std::string::npos)
+ else if(result.find("VALID IGNORE") != std::string::npos)
{
return false;
}
@@ -137,28 +141,25 @@ void ScriptManager::preProcessFile(boost::filesystem::path& filePath,
<< " PREPROCESS " << filePath.string();
if(verified == OK)
+ {
command << " OK";
+ }
else if(verified == WAIT)
+ {
command << " WAIT";
+ }
else if(verified == FATAL)
+ {
command << " FATAL";
+ }
std::string result = exec(command.str());
if(result.find("PREPROCESS OK") == std::string::npos)
{
- if(result.find("PREPROCESS FATAL") != std::string::npos)
- {
- std::stringstream errorStream;
- errorStream << "Pre process error: " << result;
- throw std::runtime_error(errorStream.str());
- }
- else
- {
- std::stringstream errorStream;
- errorStream << "Unknown pre process error: " << result;
- throw std::runtime_error(errorStream.str());
- }
+ std::stringstream errorStream;
+ errorStream << "Pre process error: " << result;
+ throw std::runtime_error(errorStream.str());
}
}
@@ -175,28 +176,25 @@ void ScriptManager::postProcessFile(boost::filesystem::path& filePath,
<< " POSTPROCESS " << filePath.string();
if(verified == OK)
+ {
command << " OK";
+ }
else if(verified == WAIT)
+ {
command << " WAIT";
+ }
else if(verified == FATAL)
+ {
command << " FATAL";
+ }
std::string result = exec(command.str());
if(result.find("POSTPROCESS OK") == std::string::npos)
{
- if(result.find("POSTPROCESS FATAL") != std::string::npos)
- {
- std::stringstream errorStream;
- errorStream << "Post process error: " << result;
- throw std::runtime_error(errorStream.str());
- }
- else
- {
- std::stringstream errorStream;
- errorStream << "Unknown post process error: " << result;
- throw std::runtime_error(errorStream.str());
- }
+ std::stringstream errorStream;
+ errorStream << "Post process error: " << result;
+ throw std::runtime_error(errorStream.str());
}
}
diff --git a/src/WorkerThread.cpp b/src/WorkerThread.cpp
index bde7326..72ae6f4 100644
--- a/src/WorkerThread.cpp
+++ b/src/WorkerThread.cpp
@@ -42,9 +42,6 @@ void WorkerThread::workerLoop()
boost::chrono::steady_clock::duration waitTime =
boost::chrono::seconds(m_configuration_sp->getWaitTime());
- //Process end of file after timeout like regualr files
- bool ignoreTimeout = m_configuration_sp->getIgnoreTimeout();
-
while(true)
{
try
@@ -58,7 +55,7 @@ void WorkerThread::workerLoop()
boost::chrono::steady_clock::time_point start =
boost::chrono::steady_clock::now();
- INFO_STREAM << "WorkerThread::workerLoop() processing \""
+ DEBUG_STREAM << "WorkerThread::workerLoop() processing \""
<< fileName << "\"" << endl;
try
{
@@ -67,70 +64,80 @@ void WorkerThread::workerLoop()
{
ScriptManager::Verified verified;
+ //Waiting for the file to be written until timeout
do
{
+ //Check file state
verified = m_fileManager_sp->isFileVerified(origPath);
if(verified == ScriptManager::OK ||
verified == ScriptManager::FATAL)
break;
+ DEBUG_STREAM << "WorkerThread::workerLoop() waiting \""
+ << fileName << "\" to be written" << endl;
+
boost::this_thread::sleep_for(sleepTime);
}
while(boost::chrono::steady_clock::now()-start <= waitTime);
+ //Exec pre process script
m_fileManager_sp->preProcessFile(origPath, verified);
+ //Copy file to destination
copyToDestination(origPath, verified);
+ //Exec post process script
m_fileManager_sp->postProcessFile(origPath, verified);
if(verified == ScriptManager::Verified::OK)
{
INFO_STREAM << "WorkerThread::workerLoop() \"" << fileName
- << "\" ingested regularly" << endl;
+ << "\" is a regular file" << endl;
m_preProcessor_p->incrementRegularCounter();
}
else if(verified == ScriptManager::Verified::WAIT)
{
WARN_STREAM << "WorkerThread::workerLoop() \"" << fileName
- << "\" archived after timeout" << endl;
+ << "\" has reached a wait eof timeout" << endl;
m_preProcessor_p->incrementWarningCounter();
}
else if(verified == ScriptManager::Verified::FATAL)
{
WARN_STREAM << "WorkerThread::workerLoop() \"" << fileName
- << "\" not archived due to fits fatal error" << endl;
+ << "\" has fatal error" << endl;
m_preProcessor_p->incrementErrorCounter();
}
}
else
{
INFO_STREAM << "WorkerThread::workerLoop() \"" << fileName
- << "\" ignored" << endl;
+ << "\" is an ignore file" << endl;
m_preProcessor_p->incrementIgnoredCounter();
}
}
catch(std::exception& ex)
{
ERROR_STREAM << "WorkerThread::workerLoop() \"" << fileName
- << "\" not archived due to " << ex.what() << endl;
+ << "\" has an error due to " << ex.what() << endl;
m_preProcessor_p->incrementErrorCounter();
}
- catch(...)
- {
- ERROR_STREAM << "WorkerThread::workerLoop() \"" << fileName
- << "\" not archived due to unknown exception" << endl;
- m_preProcessor_p->incrementErrorCounter();
- }
m_eventBuffer_sp->markAsProcessed(origPath);
}
catch(boost::thread_interrupted& ex)
{
- DEBUG_STREAM << "WorkerThread::workerLoop() interrupt" << endl;
+ DEBUG_STREAM << "WorkerThread::workerLoop() interrupt request" << endl;
break;
}
+ catch(std::exception& ex)
+ {
+ ERROR_STREAM << "WorkerThread::workerLoop() " << ex.what() << endl;
+ }
+ catch(...)
+ {
+ ERROR_STREAM << "WorkerThread::workerLoop() unknown exception" << endl;
+ }
} //thread loop
}
@@ -140,50 +147,57 @@ void WorkerThread::workerLoop()
void WorkerThread::copyToDestination(boost::filesystem::path& origPath,
ScriptManager::Verified verified) throw (std::runtime_error)
{
- DEBUG_STREAM << "WorkerThread::moveFile()" << endl;
+ DEBUG_STREAM << "WorkerThread::copyToDestination()" << endl;
- //Process end of file after timeout like regualr files
- bool ignoreTimeout = m_configuration_sp->getIgnoreTimeout();
+ std::string fileName = origPath.stem().string();
boost::filesystem::path destPath;
+ //Retrieve destination path based verification result
if(verified == ScriptManager::OK)
{
destPath /= m_configuration_sp->getRegularPath();
}
else if(verified == ScriptManager::WAIT)
{
- if(ignoreTimeout)
- destPath /= m_configuration_sp->getRegularPath();
- else
- destPath /= m_configuration_sp->getWarningPath();
+ destPath /= m_configuration_sp->getWarningPath();
}
- else
+ else if(verified == ScriptManager::FATAL)
{
destPath /= m_configuration_sp->getErrorPath();
}
- //TODO: come fare per le destinazioni vuote?
+ //If destination is defined then copy file
+ if(!destPath.empty())
+ {
+ if(!boost::filesystem::is_regular_file(origPath))
+ throw std::runtime_error( "Origin path \"" +
+ origPath.string() + "\" is not a regular file");
- if(!boost::filesystem::is_regular_file(origPath))
- throw std::runtime_error( "Origin path \"" +
- origPath.string() + "\" is not a regular file");
+ if(!boost::filesystem::exists(destPath))
+ throw std::runtime_error( "Destination path \"" +
+ destPath.string() + "\" not exists");
- if(!boost::filesystem::exists(destPath))
- throw std::runtime_error( "Destination path \"" +
- destPath.string() + "\" not exists");
+ if(!boost::filesystem::is_directory(destPath))
+ throw std::runtime_error( "Destination path \"" +
+ destPath.string() + "\" is not a directory");
- if(!boost::filesystem::is_directory(destPath))
- throw std::runtime_error( "Destination path \"" +
- destPath.string() + "\" is not a directory");
+ destPath /= origPath.filename();
- destPath /= origPath.filename();
+ if(boost::filesystem::exists(destPath))
+ throw std::runtime_error( "Destination path \"" +
+ destPath.string() + "\" already exists");
- if(boost::filesystem::exists(destPath))
- throw std::runtime_error( "Destination path \"" +
- destPath.string() + "\" already exists");
+ boost::filesystem::copy(origPath, destPath);
- boost::filesystem::copy(origPath, destPath);
+ DEBUG_STREAM << "WorkerThread::copyToDestination() \"" << fileName
+ << "\" copied to \"" << destPath.parent_path() << "\"" << endl;
+ }
+ else
+ {
+ DEBUG_STREAM << "WorkerThread::copyToDestination() \"" << fileName
+ << "\" has not a defined destination" << endl;
+ }
}
} //namespace
--
GitLab