Page MenuHomePhorge
Differential D309 Diff 974 src/job/cprjobhandler.cpp

Changeset View

Standalone View

View Options

src/job/cprjobhandler.cpp

Show First 20 Lines • Show All 123 Lines • ▼ Show 20 Lines struct CprJobHandler::Private
using JobMap = std::unordered_map<std::string, JobQueue>; using JobMap = std::unordered_map<std::string, JobQueue>;
JobMap jobQueues; JobMap jobQueues;
void submitImpl(BaseJob job, std::function<void(Response)> userCallback); void submitImpl(BaseJob job, std::function<void(Response)> userCallback);
void addToQueue(BaseJob job, Callback callback) { void addToQueue(BaseJob job, Callback callback) {
boost::asio::post( boost::asio::post(
executor, executor,
[=] { [=, this] {
// precondition: job has a queueId // precondition: job has a queueId
auto queueId = job.queueId().value(); auto queueId = job.queueId().value();
jobQueues[queueId].push_back(JobDesc{job, callback, Waiting}); jobQueues[queueId].push_back(JobDesc{job, callback, Waiting});
}); });
} }
void clearQueue(std::string queueId) { void clearQueue(std::string queueId) {
boost::asio::post( boost::asio::post(
executor, executor,
[=] { clearQueueImpl(queueId); }); [=, this] { clearQueueImpl(queueId); });
} }
void clearQueueImpl(std::string queueId) { void clearQueueImpl(std::string queueId) {
Response fakeResponse; Response fakeResponse;
fakeResponse.statusCode = headFailureCancelledStatusCode; fakeResponse.statusCode = headFailureCancelledStatusCode;
fakeResponse.body = JsonBody( fakeResponse.body = JsonBody(
json{ {"errcode", headFailureCancelledErrorCode}, json{ {"errcode", headFailureCancelledErrorCode},
{"error", headFailureCancelledErrorMsg} } {"error", headFailureCancelledErrorMsg} }
); );
kzo.job.dbg() << "clearQueueImpl called with " << queueId << std::endl; kzo.job.dbg() << "clearQueueImpl called with " << queueId << std::endl;
for (auto [job, callback, status] : jobQueues[queueId]) { for (auto [job, callback, status] : jobQueues[queueId]) {
kzo.job.dbg() << "this job is " << (status == Waiting ? "Waiting" : "Running") << std::endl; kzo.job.dbg() << "this job is " << (status == Waiting ? "Waiting" : "Running") << std::endl;
if (status == Waiting) { if (status == Waiting) {
// Run callback in a different thread, just as in submitImpl(). // Run callback in a different thread, just as in submitImpl().
q->async([=] { callback(job.genResponse(fakeResponse)); } ); q->async([=, this] { callback(job.genResponse(fakeResponse)); } );
} }
// if status is Running, the callback is already called // if status is Running, the callback is already called
} }
jobQueues[queueId].clear(); jobQueues[queueId].clear();
} }
void popJob(std::string queueId) { void popJob(std::string queueId) {
boost::asio::post( boost::asio::post(
executor, executor,
[=] { popJobImpl(queueId); }); [=, this] { popJobImpl(queueId); });
} }
void popJobImpl(std::string queueId) { void popJobImpl(std::string queueId) {
if (! jobQueues[queueId].empty()) { if (! jobQueues[queueId].empty()) {
jobQueues[queueId].pop_front(); jobQueues[queueId].pop_front();
} }
} }
void monitorQueues() { void monitorQueues() {
boost::asio::post( boost::asio::post(
executor, executor,
[=] { [=, this] {
// precondition: job has a queueId // precondition: job has a queueId
for (auto &[queueId, queue] : jobQueues) { // need to change queue for (auto &[queueId, queue] : jobQueues) { // need to change queue
if (! queue.empty()) { if (! queue.empty()) {
auto [job, callback, status] = queue.front(); auto [job, callback, status] = queue.front();
if (status == Waiting) { if (status == Waiting) {
queue.front().status = Running; queue.front().status = Running;
submitImpl( submitImpl(
job, job,
[=](Response r) { // in new thread [=, this](Response r) { // in new thread
callback(r); callback(r);
if (! r.success() // should be enough for now if (! r.success() // should be enough for now
&& job.queuePolicy() == CancelFutureIfFailed) { && job.queuePolicy() == CancelFutureIfFailed) {
clearQueue(queueId); // in executor thread clearQueue(queueId); // in executor thread
} else { } else {
popJob(queueId); // in executor thread popJob(queueId); // in executor thread
} }
}); });
} }
} }
} }
}); });
} }
void addTimerToMap(TimerSP timer, std::optional<std::string> timerId) { void addTimerToMap(TimerSP timer, std::optional<std::string> timerId) {
boost::asio::post( boost::asio::post(
executor, executor,
[=] { [=, this] {
timerMap[timerId].push_back(timer); timerMap[timerId].push_back(timer);
}); });
} }
void clearTimer(TimerSP timer, std::optional<std::string> timerId) { void clearTimer(TimerSP timer, std::optional<std::string> timerId) {
boost::asio::post( boost::asio::post(
executor, executor,
[=] { [=, this] {
std::remove(timerMap[timerId].begin(), timerMap[timerId].end(), timer); auto &timers = timerMap[timerId];
timers.erase(std::remove(timers.begin(), timers.end(), timer), timers.end());
}); });
} }
void cancelAllTimers(std::optional<std::string> timerId) { void cancelAllTimers(std::optional<std::string> timerId) {
boost::asio::post( boost::asio::post(
executor, executor,
[=] { [=, this] {
cancelAllTimersImpl(timerId); cancelAllTimersImpl(timerId);
}); });
} }
void cancelAllTimersImpl(std::optional<std::string> timerId) { void cancelAllTimersImpl(std::optional<std::string> timerId) {
auto timers = timerMap[timerId]; auto timers = timerMap[timerId];
timerMap.erase(timerId); timerMap.erase(timerId);
for (auto timer : timers) { for (auto timer : timers) {
timer->cancel(); timer->cancel();
} }
} }
void intervalTimerCallback(TimerSP timer, void intervalTimerCallback(TimerSP timer,
std::function<void()> func, std::function<void()> func,
int ms, int ms,
const boost::system::error_code &error) { const boost::system::error_code &error) {
if (! error) { if (! error) {
func(); func();
auto dur = boost::asio::chrono::milliseconds(ms); auto dur = boost::asio::chrono::milliseconds(ms);
timer->expires_at(timer->expiry() + dur); timer->expires_at(timer->expiry() + dur);
timer->async_wait( timer->async_wait(
[=](const boost::system::error_code &error) { [=, this](const boost::system::error_code &error) {
intervalTimerCallback(timer, func, ms, error); intervalTimerCallback(timer, func, ms, error);
}); });
} }
} }
// This makes the callbacks work with both versions of cpr. // This makes the callbacks work with both versions of cpr.
struct ReadCallback : public std::function<bool(char *, size_t &)> struct ReadCallback : public std::function<bool(char *, size_t &)>
{ {
Show All 26 Lines struct CprJobHandler::Private
} }
}; };
}; };
CprJobHandler::CprJobHandler(boost::asio::io_context::executor_type executor) CprJobHandler::CprJobHandler(boost::asio::io_context::executor_type executor)
: m_d(new Private{this, std::move(executor), Private::TimerMap{}, Private::JobMap{}}) : m_d(new Private{this, std::move(executor), Private::TimerMap{}, Private::JobMap{}})
{ {
setInterval( setInterval(
[=] { [=, this] {
m_d->monitorQueues(); m_d->monitorQueues();
}, },
50, // ms 50, // ms
"-queue-monitor"); "-queue-monitor");
} }
CprJobHandler::~CprJobHandler() = default; CprJobHandler::~CprJobHandler() = default;
void CprJobHandler::async(std::function<void()> func) void CprJobHandler::async(std::function<void()> func)
{ {
std::thread([func=std::move(func), guard=boost::asio::make_work_guard(m_d->executor)]() { std::thread([func=std::move(func), guard=boost::asio::make_work_guard(m_d->executor)]() {
func(); func();
}).detach(); }).detach();
} }
void CprJobHandler::setTimeout(std::function<void()> func, int ms, std::optional<std::string> timerId) void CprJobHandler::setTimeout(std::function<void()> func, int ms, std::optional<std::string> timerId)
{ {
auto timer=std::make_shared<boost::asio::steady_timer>( auto timer=std::make_shared<boost::asio::steady_timer>(
m_d->executor, boost::asio::chrono::milliseconds(ms)); m_d->executor, boost::asio::chrono::milliseconds(ms));
m_d->addTimerToMap(timer, timerId); m_d->addTimerToMap(timer, timerId);
timer->async_wait( timer->async_wait(
[=, timer=timer](const boost::system::error_code &error){ [=, this, timer=timer](const boost::system::error_code &error){
if (! error) { if (! error) {
func(); func();
this->m_d->clearTimer(timer, timerId); this->m_d->clearTimer(timer, timerId);
} }
}); });
} }
void CprJobHandler::setInterval(std::function<void()> func, int ms, std::optional<std::string> timerId) void CprJobHandler::setInterval(std::function<void()> func, int ms, std::optional<std::string> timerId)
{ {
auto dur = boost::asio::chrono::milliseconds(ms); auto dur = boost::asio::chrono::milliseconds(ms);
auto timer = std::make_shared<boost::asio::steady_timer>(m_d->executor, dur); auto timer = std::make_shared<boost::asio::steady_timer>(m_d->executor, dur);
m_d->addTimerToMap(timer, timerId); m_d->addTimerToMap(timer, timerId);
timer->async_wait( timer->async_wait(
[=](const boost::system::error_code &error) { [=, this](const boost::system::error_code &error) {
m_d->intervalTimerCallback(timer, func, ms, error); m_d->intervalTimerCallback(timer, func, ms, error);
}); });
} }
void CprJobHandler::cancel(std::string timerId) void CprJobHandler::cancel(std::string timerId)
{ {
m_d->cancelAllTimers(timerId); m_d->cancelAllTimers(timerId);
} }
▲ Show 20 Lines • Show All 115 Lines • ▼ Show 20 Lines void CprJobHandler::Private::submitImpl(BaseJob job, std::function<void(Response)> userCallback)
body, body,
BaseJob::Header(r.header.begin(), r.header.end()), BaseJob::Header(r.header.begin(), r.header.end()),
{} // extraData, will be added in genResponse {} // extraData, will be added in genResponse
}; };
}; };
std::shared_future<Response> res = std::visit(lager::visitor{ std::shared_future<Response> res = std::visit(lager::visitor{
[=](BaseJob::Get) { [=, this](BaseJob::Get) {
if (readCallback) { if (readCallback) {
return cpr::GetCallback(callback, url, cpr::ReadCallback(readCallback), header, params); return cpr::GetCallback(callback, url, cpr::ReadCallback(readCallback), header, params);
} else if (writeCallback) { } else if (writeCallback) {
return cpr::GetCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params); return cpr::GetCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params);
} else { } else {
return cpr::GetCallback(callback, url, header, body, params); return cpr::GetCallback(callback, url, header, body, params);
} }
}, },
[=](BaseJob::Post) { [=, this](BaseJob::Post) {
if (readCallback) { if (readCallback) {
return cpr::PostCallback(callback, url, cpr::ReadCallback(readCallback), header, params); return cpr::PostCallback(callback, url, cpr::ReadCallback(readCallback), header, params);
} else if (writeCallback) { } else if (writeCallback) {
return cpr::PostCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params); return cpr::PostCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params);
} else { } else {
return cpr::PostCallback(callback, url, header, body, params); return cpr::PostCallback(callback, url, header, body, params);
} }
}, },
[=](BaseJob::Put) { [=, this](BaseJob::Put) {
if (readCallback) { if (readCallback) {
return cpr::PutCallback(callback, url, cpr::ReadCallback(readCallback), header, params); return cpr::PutCallback(callback, url, cpr::ReadCallback(readCallback), header, params);
} else if (writeCallback) { } else if (writeCallback) {
return cpr::PutCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params); return cpr::PutCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params);
} else { } else {
return cpr::PutCallback(callback, url, header, body, params); return cpr::PutCallback(callback, url, header, body, params);
} }
}, },
[=](BaseJob::Delete) { [=, this](BaseJob::Delete) {
if (readCallback) { if (readCallback) {
return cpr::DeleteCallback(callback, url, cpr::ReadCallback(readCallback), header, params); return cpr::DeleteCallback(callback, url, cpr::ReadCallback(readCallback), header, params);
} else if (writeCallback) { } else if (writeCallback) {
return cpr::DeleteCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params); return cpr::DeleteCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params);
} else { } else {
return cpr::DeleteCallback(callback, url, header, body, params); return cpr::DeleteCallback(callback, url, header, body, params);
} }
} }
}, method).share(); }, method).share();
q->async([=]() { q->async([=, this]() {
userCallback(job.genResponse(res.get())); userCallback(job.genResponse(res.get()));
}); });
} }
void CprJobHandler::stop() void CprJobHandler::stop()
{ {
boost::asio::post( boost::asio::post(
m_d->executor, m_d->executor,
[=] { [=, this] {
auto ids = zug::into_vector( auto ids = zug::into_vector(
zug::map([](auto i) { return i.first; }), zug::map([](auto i) { return i.first; }),
m_d->timerMap); m_d->timerMap);
for (auto id : ids) { for (auto id : ids) {
m_d->cancelAllTimersImpl(id); m_d->cancelAllTimersImpl(id);
} }
m_d->jobQueues.clear(); m_d->jobQueues.clear();
}); });
} }
} }