// Copyright 2015 The Weave Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "examples/provider/event_task_runner.h"
#include <signal.h>
namespace weave {
namespace examples {
namespace {
event_base* g_event_base = nullptr;
}
void EventTaskRunner::PostDelayedTask(
const tracked_objects::Location& from_here,
const base::Closure& task,
base::TimeDelta delay) {
base::Time new_time = base::Time::Now() + delay;
if (queue_.empty() || new_time < queue_.top().first.first) {
ReScheduleEvent(delay);
}
queue_.emplace(std::make_pair(new_time, ++counter_), task);
}
void EventTaskRunner::AddIoCompletionTask(
int fd,
int16_t what,
const EventTaskRunner::IoCompletionCallback& task) {
int16_t flags = EV_PERSIST | EV_ET;
flags |= (what & kReadable) ? EV_READ : 0;
flags |= (what & kWriteable) ? EV_WRITE : 0;
#if LIBEVENT_VERSION_NUMBER >= 0x02010400
flags |= (what & kClosed) ? EV_CLOSED : 0;
#endif
event* ioevent = event_new(base_.get(), fd, flags, FdEventHandler, this);
EventPtr<event> ioeventPtr{ioevent};
fd_task_map_.insert(
std::make_pair(fd, std::make_pair(std::move(ioeventPtr), task)));
event_add(ioevent, nullptr);
}
void EventTaskRunner::RemoveIoCompletionTask(int fd) {
fd_task_map_.erase(fd);
}
void EventTaskRunner::Run() {
g_event_base = base_.get();
struct sigaction sa = {};
sa.sa_handler = [](int signal) {
event_base_loopexit(g_event_base, nullptr);
};
sigfillset(&sa.sa_mask);
sigaction(SIGINT, &sa, nullptr);
do {
event_base_loop(g_event_base, EVLOOP_ONCE);
} while (!event_base_got_exit(g_event_base));
g_event_base = nullptr;
}
void EventTaskRunner::ReScheduleEvent(base::TimeDelta delay) {
timespec ts = delay.ToTimeSpec();
timeval tv = {ts.tv_sec, ts.tv_nsec / 1000};
event_add(task_event_.get(), &tv);
}
void EventTaskRunner::EventHandler(int /* fd */,
int16_t /* what */,
void* runner) {
static_cast<EventTaskRunner*>(runner)->Process();
}
void EventTaskRunner::FreeEvent(event* evnt) {
event_del(evnt);
event_free(evnt);
}
void EventTaskRunner::Process() {
while (!queue_.empty() && queue_.top().first.first <= base::Time::Now()) {
auto cb = queue_.top().second;
queue_.pop();
cb.Run();
}
if (!queue_.empty()) {
base::TimeDelta delta = std::max(
base::TimeDelta(), queue_.top().first.first - base::Time::Now());
ReScheduleEvent(delta);
}
}
void EventTaskRunner::FdEventHandler(int fd, int16_t what, void* runner) {
static_cast<EventTaskRunner*>(runner)->ProcessFd(fd, what);
}
void EventTaskRunner::ProcessFd(int fd, int16_t what) {
auto it = fd_task_map_.find(fd);
if (it != fd_task_map_.end()) {
const IoCompletionCallback& callback = it->second.second;
callback.Run(fd, what, this);
}
}
} // namespace examples
} // namespace weave