140 lines
3.4 KiB
C++
140 lines
3.4 KiB
C++
#include "stdafx.h"
|
|
#include "BatchProcessor.h"
|
|
#include "DBProxyServerEx.h"
|
|
|
|
namespace batch {
|
|
|
|
//==================================================================================================
|
|
BatchProcessor*
|
|
BatchProcessor::batch_processor_static_ = NULL;
|
|
|
|
|
|
void
|
|
BatchProcessor::CreateInstance() {
|
|
// thread unsafe singleton
|
|
if (batch_processor_static_ == NULL) {
|
|
batch_processor_static_ = new BatchProcessor;
|
|
batch_processor_static_->RegisterBatchUnits();
|
|
}
|
|
}
|
|
|
|
BatchProcessor::BatchProcessor()
|
|
: auto_mode_(false)
|
|
, started_(false)
|
|
, batch_unit_(NULL)
|
|
, batch_query_pool_(NULL)
|
|
{
|
|
auto_mode_ = !!GetPrivateProfileInt("DBPUNIT_DATA", "UNIT_AUTO_MODE", 0, "./DBPUnit.ini");
|
|
}
|
|
|
|
BatchProcessor::~BatchProcessor()
|
|
{
|
|
UnregisterBatchUnits();
|
|
}
|
|
|
|
void
|
|
BatchProcessor::ChangeRunState(BOOLEAN start)
|
|
{
|
|
if (batch_processor_static_ == NULL)
|
|
return;
|
|
|
|
BatchProcessor* const this_ = batch_processor_static_;
|
|
BOOLEAN prev_state = this_->started_;
|
|
BOOLEAN post_state = !!start;
|
|
this_->started_ = post_state;
|
|
|
|
size_t pendings = this_->batch_units_.size();
|
|
SUNLOG(eCRITICAL_LOG, "|ÐÅÏ¢=Run batch unit processor|Prev=%s|Next=%s|Pendings=%d|",
|
|
prev_state ? "Started" : "Stopped",
|
|
post_state ? "Started" : "Stopped",
|
|
pendings);
|
|
}
|
|
|
|
BOOLEAN
|
|
BatchProcessor::Update()
|
|
{
|
|
if (batch_processor_static_ == NULL)
|
|
return false;
|
|
|
|
BatchProcessor* const this_ = batch_processor_static_;
|
|
|
|
if (this_->started_ == false)
|
|
return false;
|
|
|
|
IBatchUnit* batch_unit = this_->batch_unit_;
|
|
if (batch_unit == NULL)
|
|
{
|
|
PendingBatchUnits& pendings = this_->batch_units_;
|
|
if (pendings.empty())
|
|
{
|
|
ChangeRunState(false);
|
|
#if DBPROXY_TYPE_SELECTOR == DBPROXY_TYPE_UNIT
|
|
if (this_->auto_mode_)
|
|
g_DBProxyServer.RequestShutdown();
|
|
#endif
|
|
|
|
return false;
|
|
}
|
|
this_->batch_unit_ = batch_unit = pendings.front();
|
|
pendings.pop_front();
|
|
}
|
|
|
|
const IBatchUnit::eSTEP step = batch_unit->CurrentStep();
|
|
switch (step)
|
|
{
|
|
case batch_unit->STEP_NONE:
|
|
batch_unit->BatchInit();
|
|
break;
|
|
case batch_unit->STEP_PROCESSING:
|
|
batch_unit->BatchProcess();
|
|
break;
|
|
case batch_unit->STEP_COMPLETED:
|
|
batch_unit->BatchRelease();
|
|
this_->batch_unit_ = NULL;
|
|
break;
|
|
case batch_unit->STEP_RELEASED:
|
|
//ASSERT(!"Unexpected batch unit step");
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
BOOLEAN
|
|
BatchProcessor::RegisterBatchUnit(IBatchUnit* batch_unit)
|
|
{
|
|
if (batch_unit == NULL ||
|
|
batch_unit->CurrentStep() != batch_unit->STEP_NONE)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
// no check duplicated batch unit
|
|
BatchProcessor* const this_ = batch_processor_static_;
|
|
PendingBatchUnits& pendings = this_->batch_units_;
|
|
pendings.push_back(batch_unit);
|
|
|
|
return true;
|
|
}
|
|
|
|
void
|
|
BatchProcessor::QueryResultCallBack(DWORD index, QueryResult* query_result)
|
|
{
|
|
if (batch_processor_static_ == NULL)
|
|
return;
|
|
BatchProcessor* const this_ = batch_processor_static_;
|
|
|
|
IBatchUnit* const batch_unit = \
|
|
reinterpret_cast<IBatchUnit*>(query_result->GetVoidObject());
|
|
|
|
if(batch_unit != this_->batch_unit_) {
|
|
SUNLOG(eCRITICAL_LOG, "Warning! skipped result occurred!");
|
|
return;
|
|
}
|
|
|
|
batch_unit->DBResult(index, query_result);
|
|
}
|
|
|
|
}; //end of namespace 'batch'
|
|
|