Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat/monitor poll #414

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Feat/monitor poll #414

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a4570b1
monitor: add get_event() method
lp35 May 14, 2020
78dab5d
monitor: add compatibility with zmq::poll()
lp35 May 14, 2020
3aa80d0
monitor: add compatibility with zmq::poller_t
lp35 May 14, 2020
c2298e8
monitor: expose close() method to public
lp35 May 14, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 46 additions & 1 deletion tests/monitor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ TEST_CASE("monitor move assign", "[monitor]")
}
}

TEST_CASE("monitor init event count", "[monitor]")
TEST_CASE("monitor init check event count", "[monitor]")
{
common_server_client_setup s{false};
mock_monitor_t monitor;
Expand All @@ -92,6 +92,51 @@ TEST_CASE("monitor init event count", "[monitor]")
CHECK(monitor.total == expected_event_count);
}

TEST_CASE("monitor init get event count", "[monitor]")
{
common_server_client_setup s{ false };
zmq::monitor_t monitor;

const int expected_event_count = 2;
monitor.init(s.client, "inproc://foo");

int total{ 0 };
int connect_delayed{ 0 };
int connected{ 0 };

auto lbd_count_event = [&](const zmq_event_t& event) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What does lbd stand for?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"lbd" is short for "lambda". I like to differentiate "callback" methods over classical variable in my code.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm. I am not sure if it's necessary to use a prefix at all (if it were a function pointer, would it make a difference?), but if you feel it's good to have, please use lambda_.

switch (event.event)
{
case ZMQ_EVENT_CONNECT_DELAYED:
connect_delayed++;
total++;
break;

case ZMQ_EVENT_CONNECTED:
connected++;
total++;
break;
}
};

zmq_event_t eventMsg;
std::string address;
CHECK_FALSE(monitor.get_event(eventMsg, address, zmq::recv_flags::dontwait));
s.init();

while (total < expected_event_count)
{
if (!monitor.get_event(eventMsg, address))
continue;

lbd_count_event(eventMsg);
}

CHECK(connect_delayed == 1);
CHECK(connected == 1);
CHECK(total == expected_event_count);
}

TEST_CASE("monitor init abort", "[monitor]")
{
class mock_monitor : public mock_monitor_t
Expand Down
47 changes: 47 additions & 0 deletions zmq.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2224,6 +2224,53 @@ class monitor_t
on_monitor_started();
}

#if ZMQ_VERSION_MAJOR >= 4
bool get_event(zmq_event_t& eventMsg, std::string& address, zmq::recv_flags flags = zmq::recv_flags::none)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure about the signature of this method. What about

std::optional<std::pair<zmq_event_t, std::string>> get_event(zmq::recv_flags flags = zmq::recv_flags::none)

(This requires C++17 this way, but we can also do this with the fallback to detail::trivial_optional as done for other types)

This removes the ambiguity of whether the original value of eventMsg and address are used and in what cases they are modified.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I'm not very familiar with std::optional and will try to implement this way

{
assert(_monitor_socket);

eventMsg.event = 0;
eventMsg.value = 0;
address = std::string();

{
message_t msg;
int rc = zmq_msg_recv(msg.handle(), _monitor_socket.handle(),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can't this use _monitor_socket.recv?

Copy link
Author

@lp35 lp35 May 22, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Of course, but then it is not possible to use the same behaviour as the old check_event. Otherwise I will need to use a try catch statement

static_cast<int>(flags));

if (rc == -1)
{
if (zmq_errno() == ETERM || zmq_errno() == EAGAIN)
return false;
Comment on lines +2253 to +2254
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why does this also returns false on ETERM? There's no way to distinguish EAGAIN and ETERM this way, and this seems inconsistent with the behavior of zmq::socket_t::recv.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Check https://github.com/zeromq/cppzmq/blob/master/zmq.hpp#L2274.

I just kept same behaviour implemented in check_event(). If this is not mandatory, I can just throw an exception and not use a std::optionnal as described above, but just std::pair<>

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, yes, that's right. It's inconsistent one way or the other, but I think the behaviour of zmq::socket_t::recv is the appropriate one.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what we should implement in this case? Maybe we shall throw an exception?

else
throw error_t();
}

const char *data = msg.data<char>();
memcpy(&eventMsg.event, data, sizeof(uint16_t));
data += sizeof(uint16_t);
memcpy(&eventMsg.value, data, sizeof(int32_t));
}

message_t addrMsg;
int rc = zmq_msg_recv(addrMsg.handle(), _monitor_socket.handle(),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Like above, can't this use _monitor_socket.recv?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See prev comment

Thanks for the extensions, I think these will be really useful.

Please see my comments on individual issues.

Also, please remember to reformat all commits with clang-format.

Could you add this to the contribution guidelines of the project?

Most of the travis builds are failing right now, could you give them a look and fix them, please?

I don't like to put effort on something that will be not merged. I was waiting for a green thumb before putting more work on this, as I did not asked anyone about the design of this PR ;)

Thanks for reviewing, will try to solve all of those next week. BR

static_cast<int>(flags));

if (rc == -1)
{
if (zmq_errno() == ETERM)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Again, why does this return false rather than throwing?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See my comment above

return false;
else
throw error_t();
}

const char *str = addrMsg.data<char>();
address = std::string(str, str + addrMsg.size());

return true;
}
#endif

bool check_event(int timeout = 0)
{
assert(_monitor_socket);
Expand Down