Skip to content

L
libzmq
Commit 2d6f5b0e authored by Ian Barber's avatar Ian Barber
Browse files
Options

Merge pull request #985 from hintjens/master 

Cleaned up socket monitor code, tests, and man page
parents 50afebe4 9753de85
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 56 additions and 119 deletions
doc/zmq.txt
View file @ 2d6f5b0e
...@@ -102,6 +102,7 @@ Access message content:: ...@@ -102,6 +102,7 @@ Access message content::
linkzmq:zmq_msg_more[3] linkzmq:zmq_msg_more[3]
Work with message properties:: Work with message properties::
linkzmq:zmq_msg_gets[3]
linkzmq:zmq_msg_get[3] linkzmq:zmq_msg_get[3]
linkzmq:zmq_msg_set[3] linkzmq:zmq_msg_set[3]
......
doc/zmq_socket_monitor.txt
View file @ 2d6f5b0e
This diff is collapsed.
include/zmq.h
View file @ 2d6f5b0e
...@@ -326,34 +326,20 @@ ZMQ_EXPORT char *zmq_msg_gets (zmq_msg_t *msg, char *property); ...@@ -326,34 +326,20 @@ ZMQ_EXPORT char *zmq_msg_gets (zmq_msg_t *msg, char *property);
/* 0MQ socket events and monitoring */ /* 0MQ socket events and monitoring */
/******************************************************************************/ /******************************************************************************/
/* Socket transport events (tcp and ipc only) */ /* Socket transport events (TCP and IPC only) */
#define ZMQ_EVENT_CONNECTED 1
#define ZMQ_EVENT_CONNECT_DELAYED 2 #define ZMQ_EVENT_CONNECTED 0x0001
#define ZMQ_EVENT_CONNECT_RETRIED 4 #define ZMQ_EVENT_CONNECT_DELAYED 0x0002
#define ZMQ_EVENT_CONNECT_RETRIED 0x0004
#define ZMQ_EVENT_LISTENING 8 #define ZMQ_EVENT_LISTENING 0x0008
#define ZMQ_EVENT_BIND_FAILED 16 #define ZMQ_EVENT_BIND_FAILED 0x0010
#define ZMQ_EVENT_ACCEPTED 0x0020
#define ZMQ_EVENT_ACCEPTED 32 #define ZMQ_EVENT_ACCEPT_FAILED 0x0040
#define ZMQ_EVENT_ACCEPT_FAILED 64 #define ZMQ_EVENT_CLOSED 0x0080
#define ZMQ_EVENT_CLOSE_FAILED 0x0100
#define ZMQ_EVENT_CLOSED 128 #define ZMQ_EVENT_DISCONNECTED 0x0200
#define ZMQ_EVENT_CLOSE_FAILED 256 #define ZMQ_EVENT_MONITOR_STOPPED 0x0400
#define ZMQ_EVENT_DISCONNECTED 512 #define ZMQ_EVENT_ALL 0xFFFF
#define ZMQ_EVENT_MONITOR_STOPPED 1024
#define ZMQ_EVENT_ALL ( ZMQ_EVENT_CONNECTED | ZMQ_EVENT_CONNECT_DELAYED | \
ZMQ_EVENT_CONNECT_RETRIED | ZMQ_EVENT_LISTENING | \
ZMQ_EVENT_BIND_FAILED | ZMQ_EVENT_ACCEPTED | \
ZMQ_EVENT_ACCEPT_FAILED | ZMQ_EVENT_CLOSED | \
ZMQ_EVENT_CLOSE_FAILED | ZMQ_EVENT_DISCONNECTED | \
ZMQ_EVENT_MONITOR_STOPPED)
/* Socket event data */
typedef struct {
uint16_t event; // id of the event as bitfield
int32_t value ; // value is either error code, fd or reconnect interval
} zmq_event_t;
ZMQ_EXPORT void *zmq_socket (void *, int type); ZMQ_EXPORT void *zmq_socket (void *, int type);
ZMQ_EXPORT int zmq_close (void *s); ZMQ_EXPORT int zmq_close (void *s);
......
src/socket_base.cpp
View file @ 2d6f5b0e
...@@ -1169,22 +1169,19 @@ void zmq::socket_base_t::extract_flags (msg_t *msg_) ...@@ -1169,22 +1169,19 @@ void zmq::socket_base_t::extract_flags (msg_t *msg_)
int zmq::socket_base_t::monitor (const char *addr_, int events_) int zmq::socket_base_t::monitor (const char *addr_, int events_)
{ {
int rc;
if (unlikely (ctx_terminated)) { if (unlikely (ctx_terminated)) {
errno = ETERM; errno = ETERM;
return -1; return -1;
} }
// Support deregistering monitoring endpoints as well
// Support deregistering monitoring endpoints as well
if (addr_ == NULL) { if (addr_ == NULL) {
stop_monitor (); stop_monitor ();
return 0; return 0;
} }
// Parse addr_ string. // Parse addr_ string.
std::string protocol; std::string protocol;
std::string address; std::string address;
rc = parse_uri (addr_, protocol, address); int rc = parse_uri (addr_, protocol, address);
if (rc != 0) if (rc != 0)
return -1; return -1;
...@@ -1192,25 +1189,24 @@ int zmq::socket_base_t::monitor (const char *addr_, int events_) ...@@ -1192,25 +1189,24 @@ int zmq::socket_base_t::monitor (const char *addr_, int events_)
if (rc != 0) if (rc != 0)
return -1; return -1;
// Event notification only supported over inproc:// // Event notification only supported over inproc://
if (protocol != "inproc") { if (protocol != "inproc") {
errno = EPROTONOSUPPORT; errno = EPROTONOSUPPORT;
return -1; return -1;
} }
// Register events to monitor
// Register events to monitor
monitor_events = events_; monitor_events = events_;
monitor_socket = zmq_socket (get_ctx (), ZMQ_PAIR); monitor_socket = zmq_socket (get_ctx (), ZMQ_PAIR);
if (monitor_socket == NULL) if (monitor_socket == NULL)
return -1; return -1;
// Never block context termination on pending event messages // Never block context termination on pending event messages
int linger = 0; int linger = 0;
rc = zmq_setsockopt (monitor_socket, ZMQ_LINGER, &linger, sizeof (linger)); rc = zmq_setsockopt (monitor_socket, ZMQ_LINGER, &linger, sizeof (linger));
if (rc == -1) if (rc == -1)
stop_monitor (); stop_monitor ();
// Spawn the monitor socket endpoint // Spawn the monitor socket endpoint
rc = zmq_bind (monitor_socket, addr_); rc = zmq_bind (monitor_socket, addr_);
if (rc == -1) if (rc == -1)
stop_monitor (); stop_monitor ();
...@@ -1229,134 +1225,88 @@ zmq::fd_t zmq::socket_base_t::fd() ...@@ -1229,134 +1225,88 @@ zmq::fd_t zmq::socket_base_t::fd()
void zmq::socket_base_t::event_connected (std::string &addr_, int fd_) void zmq::socket_base_t::event_connected (std::string &addr_, int fd_)
{ {
if (monitor_events & ZMQ_EVENT_CONNECTED) { if (monitor_events & ZMQ_EVENT_CONNECTED)
zmq_event_t event; monitor_event (ZMQ_EVENT_CONNECTED, fd_, addr_);
event.event = ZMQ_EVENT_CONNECTED;
event.value = fd_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_connect_delayed (std::string &addr_, int err_) void zmq::socket_base_t::event_connect_delayed (std::string &addr_, int err_)
{ {
if (monitor_events & ZMQ_EVENT_CONNECT_DELAYED) { if (monitor_events & ZMQ_EVENT_CONNECT_DELAYED)
zmq_event_t event; monitor_event (ZMQ_EVENT_CONNECT_DELAYED, err_, addr_);
event.event = ZMQ_EVENT_CONNECT_DELAYED;
event.value = err_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_connect_retried (std::string &addr_, int interval_) void zmq::socket_base_t::event_connect_retried (std::string &addr_, int interval_)
{ {
if (monitor_events & ZMQ_EVENT_CONNECT_RETRIED) { if (monitor_events & ZMQ_EVENT_CONNECT_RETRIED)
zmq_event_t event; monitor_event (ZMQ_EVENT_CONNECT_RETRIED, interval_, addr_);
event.event = ZMQ_EVENT_CONNECT_RETRIED;
event.value = interval_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_listening (std::string &addr_, int fd_) void zmq::socket_base_t::event_listening (std::string &addr_, int fd_)
{ {
if (monitor_events & ZMQ_EVENT_LISTENING) { if (monitor_events & ZMQ_EVENT_LISTENING)
zmq_event_t event; monitor_event (ZMQ_EVENT_LISTENING, fd_, addr_);
event.event = ZMQ_EVENT_LISTENING;
event.value = fd_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_bind_failed (std::string &addr_, int err_) void zmq::socket_base_t::event_bind_failed (std::string &addr_, int err_)
{ {
if (monitor_events & ZMQ_EVENT_BIND_FAILED) { if (monitor_events & ZMQ_EVENT_BIND_FAILED)
zmq_event_t event; monitor_event (ZMQ_EVENT_BIND_FAILED, err_, addr_);
event.event = ZMQ_EVENT_BIND_FAILED;
event.value = err_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_accepted (std::string &addr_, int fd_) void zmq::socket_base_t::event_accepted (std::string &addr_, int fd_)
{ {
if (monitor_events & ZMQ_EVENT_ACCEPTED) { if (monitor_events & ZMQ_EVENT_ACCEPTED)
zmq_event_t event; monitor_event (ZMQ_EVENT_ACCEPTED, fd_, addr_);
event.event = ZMQ_EVENT_ACCEPTED;
event.value = fd_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_accept_failed (std::string &addr_, int err_) void zmq::socket_base_t::event_accept_failed (std::string &addr_, int err_)
{ {
if (monitor_events & ZMQ_EVENT_ACCEPT_FAILED) { if (monitor_events & ZMQ_EVENT_ACCEPT_FAILED)
zmq_event_t event; monitor_event (ZMQ_EVENT_ACCEPT_FAILED, err_, addr_);
event.event = ZMQ_EVENT_ACCEPT_FAILED;
event.value= err_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_closed (std::string &addr_, int fd_) void zmq::socket_base_t::event_closed (std::string &addr_, int fd_)
{ {
if (monitor_events & ZMQ_EVENT_CLOSED) { if (monitor_events & ZMQ_EVENT_CLOSED)
zmq_event_t event; monitor_event (ZMQ_EVENT_CLOSED, fd_, addr_);
event.event = ZMQ_EVENT_CLOSED;
event.value = fd_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_close_failed (std::string &addr_, int err_) void zmq::socket_base_t::event_close_failed (std::string &addr_, int err_)
{ {
if (monitor_events & ZMQ_EVENT_CLOSE_FAILED) { if (monitor_events & ZMQ_EVENT_CLOSE_FAILED)
zmq_event_t event; monitor_event (ZMQ_EVENT_CLOSE_FAILED, err_, addr_);
event.event = ZMQ_EVENT_CLOSE_FAILED;
event.value = err_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::event_disconnected (std::string &addr_, int fd_) void zmq::socket_base_t::event_disconnected (std::string &addr_, int fd_)
{ {
if (monitor_events & ZMQ_EVENT_DISCONNECTED) { if (monitor_events & ZMQ_EVENT_DISCONNECTED)
zmq_event_t event; monitor_event (ZMQ_EVENT_DISCONNECTED, fd_, addr_);
event.event = ZMQ_EVENT_DISCONNECTED;
event.value = fd_;
monitor_event (event, addr_);
}
} }
void zmq::socket_base_t::monitor_event (zmq_event_t event_, const std::string& addr_) // Send a monitor event
void zmq::socket_base_t::monitor_event (int event_, int value_, const std::string &addr_)
{ {
if (monitor_socket) { if (monitor_socket) {
const uint16_t eid = (uint16_t)event_.event; // Send event in first frame
const uint32_t value = (uint32_t)event_.value;
// prepare and send first message frame
// containing event id and value
zmq_msg_t msg; zmq_msg_t msg;
zmq_msg_init_size (&msg, sizeof(eid) + sizeof(value)); zmq_msg_init_size (&msg, 6);
char* data1 = (char*)zmq_msg_data(&msg); uint8_t *data = (uint8_t *) zmq_msg_data (&msg);
memcpy (data1, &eid, sizeof(eid)); *(uint16_t *) (data + 0) = (uint16_t) event_;
memcpy (data1+sizeof(eid), &value, sizeof(value)); *(uint32_t *) (data + 2) = (uint32_t) value_;
zmq_sendmsg (monitor_socket, &msg, ZMQ_SNDMORE); zmq_sendmsg (monitor_socket, &msg, ZMQ_SNDMORE);
// prepare and send second message frame
// containing the address (endpoint) // Send address in second frame
zmq_msg_init_size (&msg, addr_.size()); zmq_msg_init_size (&msg, addr_.size());
memcpy(zmq_msg_data(&msg), addr_.c_str(), addr_.size()); memcpy (zmq_msg_data (&msg), addr_.c_str (), addr_.size ());
zmq_sendmsg (monitor_socket, &msg, 0); zmq_sendmsg (monitor_socket, &msg, 0);
} }
} }
void zmq::socket_base_t::stop_monitor() void zmq::socket_base_t::stop_monitor (void)
{ {
if (monitor_socket) { if (monitor_socket) {
if (monitor_events & ZMQ_EVENT_MONITOR_STOPPED) { if (monitor_events & ZMQ_EVENT_MONITOR_STOPPED)
zmq_event_t event; monitor_event (ZMQ_EVENT_MONITOR_STOPPED, 0, "");
event.event = ZMQ_EVENT_MONITOR_STOPPED;
event.value = 0;
monitor_event (event, "");
}
zmq_close (monitor_socket); zmq_close (monitor_socket);
monitor_socket = NULL; monitor_socket = NULL;
monitor_events = 0; monitor_events = 0;
......
src/socket_base.hpp
View file @ 2d6f5b0e
...@@ -160,7 +160,7 @@ namespace zmq ...@@ -160,7 +160,7 @@ namespace zmq
void process_destroy (); void process_destroy ();
// Socket event data dispath // Socket event data dispath
void monitor_event (zmq_event_t data_, const std::string& addr_); void monitor_event (int event_, int value_, const std::string& addr_);
// Monitor socket cleanup // Monitor socket cleanup
void stop_monitor (); void stop_monitor ();
......
tests/test_monitor.cpp
View file @ 2d6f5b0e
This diff is collapsed.
tests/testutil.hpp
View file @ 2d6f5b0e
...@@ -244,7 +244,7 @@ void s_recv_seq (void *socket, ...) ...@@ -244,7 +244,7 @@ void s_recv_seq (void *socket, ...)
} }
// Sets a zero linger period on a socket and closes it. // Sets a zero linger period on a socket and closes it.
void close_zero_linger (void *socket) void close_zero_linger (void *socket)
{ {
int linger = 0; int linger = 0;
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment