Queue::Queue(const Properties& props)
: queue_(props.getInteger("Queue.Size"))
, worker_(queue_, [ this, &props ]() -> ThreadPoolT::Config
{ //lambda returns config begin
ThreadPoolT::Config c = { //config begin
props.getInteger("Queue.Workers"),
[ this, &props ](){ //thread task lambda
boost::this_thread::interruption_point();
LogCategory* log = Log::instance()->createCategory("Server.Queue");
try
{
DB db;
db.init("DBWriter", props, log);
boost::posix_time::milliseconds delay(props.getInteger("Queue.WriteDelay"));
while (true)
{
boost::this_thread::interruption_point();
Message* msg;
while (queue_.dequeue(msg))
{
assert(msg && "Queue data seems like inconsistent");
{
if (msg->type() == MessageID::E)
{
Parser::EMessage& er = *(Parser::EMessage*)msg;
db.callPreparedStoredProc("CALL p_LoadERepo(?,?,NOW(), ?,?,?,?,?,?,?,?)",
er , [ this ] (sql::PreparedStatement& stmt, const Parser::EMessage& er) {
stmt.setUInt (1, er.getExecID());
...
stmt.setUInt (11, t.getSellerIsPrinted());
});
}
}
}
LDebug(*log) << "queue is empty, wait for delay";
boost::this_thread::sleep(delay);
boost::this_thread::interruption_point();
}
}
catch (const std::exception& e)
{
LError(*log) << std::string("Queue initialization error: ").append(e.what());
}
},
[] (boost::thread& t) { //lambda that stopping thread
t.interrupt();
t.join();
}
}; //config ends here
return c; }() //call embracing lambda (//lambda returns config end)
)
{}
========== Build: 2 succeeded, 0 failed, 4 up-to-date, 0 skipped ==========
Два вопроса: «как?» И на*... «почему?»
