Skip to content

Commit b5ce0ad

Browse files
committed
active_poller and poller_t support for file descriptors
1 parent c94c207 commit b5ce0ad

File tree

2 files changed

+58
-4
lines changed

2 files changed

+58
-4
lines changed

zmq.hpp

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2681,6 +2681,13 @@ template<typename T = no_user_data> class poller_t
26812681
}
26822682
}
26832683

2684+
void remove(fd_t fd)
2685+
{
2686+
if (0 != zmq_poller_remove_fd(poller_ptr.get(), fd)) {
2687+
throw error_t();
2688+
}
2689+
}
2690+
26842691
void modify(zmq::socket_ref socket, event_flags events)
26852692
{
26862693
if (0
@@ -2690,6 +2697,15 @@ template<typename T = no_user_data> class poller_t
26902697
}
26912698
}
26922699

2700+
void modify(fd_t fd, event_flags events)
2701+
{
2702+
if (0
2703+
!= zmq_poller_modify_fd(poller_ptr.get(), fd,
2704+
static_cast<short>(events))) {
2705+
throw error_t();
2706+
}
2707+
}
2708+
26932709
size_t wait_all(std::vector<event_type> &poller_events,
26942710
const std::chrono::milliseconds timeout)
26952711
{

zmq_addon.hpp

Lines changed: 42 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -683,10 +683,12 @@ class active_poller_t
683683

684684
void add(zmq::socket_ref socket, event_flags events, handler_type handler)
685685
{
686+
const Ref ref{socket};
687+
686688
if (!handler)
687-
throw std::invalid_argument("null handler in active_poller_t::add");
689+
throw std::invalid_argument("null handler in active_poller_t::add (socket)");
688690
auto ret = handlers.emplace(
689-
socket, std::make_shared<handler_type>(std::move(handler)));
691+
ref, std::make_shared<handler_type>(std::move(handler)));
690692
if (!ret.second)
691693
throw error_t(EINVAL); // already added
692694
try {
@@ -695,7 +697,28 @@ class active_poller_t
695697
}
696698
catch (...) {
697699
// rollback
698-
handlers.erase(socket);
700+
handlers.erase(ref);
701+
throw;
702+
}
703+
}
704+
705+
void add(fd_t fd, event_flags events, handler_type handler)
706+
{
707+
const Ref ref{fd};
708+
709+
if (!handler)
710+
throw std::invalid_argument("null handler in active_poller_t::add (fd)");
711+
auto ret = handlers.emplace(
712+
ref, std::make_shared<handler_type>(std::move(handler)));
713+
if (!ret.second)
714+
throw error_t(EINVAL); // already added
715+
try {
716+
base_poller.add(fd, events, ret.first->second.get());
717+
need_rebuild = true;
718+
}
719+
catch (...) {
720+
// rollback
721+
handlers.erase(ref);
699722
throw;
700723
}
701724
}
@@ -707,11 +730,23 @@ class active_poller_t
707730
need_rebuild = true;
708731
}
709732

733+
void remove(fd_t fd)
734+
{
735+
base_poller.remove(fd);
736+
handlers.erase(fd);
737+
need_rebuild = true;
738+
}
739+
710740
void modify(zmq::socket_ref socket, event_flags events)
711741
{
712742
base_poller.modify(socket, events);
713743
}
714744

745+
void modify(fd_t fd, event_flags events)
746+
{
747+
base_poller.modify(fd, events);
748+
}
749+
715750
size_t wait(std::chrono::milliseconds timeout)
716751
{
717752
if (need_rebuild) {
@@ -741,7 +776,10 @@ class active_poller_t
741776
bool need_rebuild{false};
742777

743778
poller_t<handler_type> base_poller{};
744-
std::unordered_map<socket_ref, std::shared_ptr<handler_type>> handlers{};
779+
780+
using Ref = std::variant<socket_ref, fd_t>;
781+
std::unordered_map<Ref, std::shared_ptr<handler_type>> handlers{};
782+
745783
std::vector<decltype(base_poller)::event_type> poller_events{};
746784
std::vector<std::shared_ptr<handler_type>> poller_handlers{};
747785
}; // class active_poller_t

0 commit comments

Comments
 (0)