generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-11-09 23:57:50 +01:00
Code Issues Projects Releases Wiki Activity

Simplified the zmq_event_t structure for easier access and better usability towards language bindings.

Browse Source
This commit is contained in:
Guido Goldstein
2013-03-08 10:34:25 +01:00
parent 068909f444
commit 21eeb03b6c
4 changed files with 51 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
include/zmq.h
View File

@@ -294,51 +294,11 @@ ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int option, int optval);
ZMQ_EVENT_ACCEPT_FAILED | ZMQ_EVENT_CLOSED | \ ZMQ_EVENT_ACCEPT_FAILED | ZMQ_EVENT_CLOSED | \
ZMQ_EVENT_CLOSE_FAILED | ZMQ_EVENT_DISCONNECTED ) ZMQ_EVENT_CLOSE_FAILED | ZMQ_EVENT_DISCONNECTED )
/* Socket event data (union member per event) */ /* Socket event data */
typedef struct { typedef struct {
int event; unsigned int event; // id of the event as bitfield
union { char *addr; // endpoint affected as c string
struct { int value ; // value is either error code, fd or reconnect interval
char *addr;
int fd;
} connected;
struct {
char *addr;
int err;
} connect_delayed;
struct {
char *addr;
int interval;
} connect_retried;
struct {
char *addr;
int fd;
} listening;
struct {
char *addr;
int err;
} bind_failed;
struct {
char *addr;
int fd;
} accepted;
struct {
char *addr;
int err;
} accept_failed;
struct {
char *addr;
int fd;
} closed;
struct {
char *addr;
int err;
} close_failed;
struct {
char *addr;
int fd;
} disconnected;
} data;
} zmq_event_t; } zmq_event_t;
ZMQ_EXPORT void *zmq_socket (void *, int type); ZMQ_EXPORT void *zmq_socket (void *, int type);

60
src/socket_base.cpp
View File

@@ -1091,9 +1091,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_CONNECTED; event.event = ZMQ_EVENT_CONNECTED;
event.data.connected.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.connected.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.connected.fd = fd_; event.value = fd_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1103,9 +1103,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_CONNECT_DELAYED; event.event = ZMQ_EVENT_CONNECT_DELAYED;
event.data.connect_delayed.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.connect_delayed.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.connect_delayed.err = err_; event.value = err_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1115,9 +1115,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_CONNECT_RETRIED; event.event = ZMQ_EVENT_CONNECT_RETRIED;
event.data.connect_retried.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.connect_retried.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.connect_retried.interval = interval_; event.value = interval_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1127,9 +1127,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_LISTENING; event.event = ZMQ_EVENT_LISTENING;
event.data.listening.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.listening.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.listening.fd = fd_; event.value = fd_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1139,9 +1139,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_BIND_FAILED; event.event = ZMQ_EVENT_BIND_FAILED;
event.data.bind_failed.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.bind_failed.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.bind_failed.err = err_; event.value = err_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1151,9 +1151,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_ACCEPTED; event.event = ZMQ_EVENT_ACCEPTED;
event.data.accepted.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.accepted.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.accepted.fd = fd_; event.value = fd_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1163,9 +1163,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_ACCEPT_FAILED; event.event = ZMQ_EVENT_ACCEPT_FAILED;
event.data.accept_failed.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.accept_failed.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.accept_failed.err= err_; event.value= err_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1175,9 +1175,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_CLOSED; event.event = ZMQ_EVENT_CLOSED;
event.data.closed.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.closed.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.closed.fd = fd_; event.value = fd_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1187,9 +1187,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_CLOSE_FAILED; event.event = ZMQ_EVENT_CLOSE_FAILED;
event.data.close_failed.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.close_failed.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.close_failed.err = err_; event.value = err_;
monitor_event (event); monitor_event (event);
} }
} }
@@ -1199,9 +1199,9 @@ 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; zmq_event_t event;
event.event = ZMQ_EVENT_DISCONNECTED; event.event = ZMQ_EVENT_DISCONNECTED;
event.data.disconnected.addr = (char *) malloc (addr_.size () + 1); event.addr = (char *) malloc (addr_.size () + 1);
copy_monitor_address (event.data.disconnected.addr, addr_); copy_monitor_address (event.addr, addr_);
event.data.disconnected.fd = fd_; event.value = fd_;
monitor_event (event); monitor_event (event);
} }
} }

35
src/zmq.cpp
View File

@@ -1004,39 +1004,8 @@ int zmq_device (int /* type */, void *frontend_, void *backend_)
void zmq_free_event (void *event_data, void * /* hint */) void zmq_free_event (void *event_data, void * /* hint */)
{ {
zmq_event_t *event = (zmq_event_t *) event_data; const zmq_event_t *event = (zmq_event_t *) event_data;
switch (event->event) { free (event->addr);
case ZMQ_EVENT_CONNECTED:
free (event->data.connected.addr);
break;
case ZMQ_EVENT_CONNECT_DELAYED:
free (event->data.connect_delayed.addr);
break;
case ZMQ_EVENT_CONNECT_RETRIED:
free (event->data.connect_retried.addr);
break;
case ZMQ_EVENT_LISTENING:
free (event->data.listening.addr);
break;
case ZMQ_EVENT_BIND_FAILED:
free (event->data.bind_failed.addr);
break;
case ZMQ_EVENT_ACCEPTED:
free (event->data.accepted.addr);
break;
case ZMQ_EVENT_ACCEPT_FAILED:
free (event->data.accept_failed.addr);
break;
case ZMQ_EVENT_CLOSED:
free (event->data.closed.addr);
break;
case ZMQ_EVENT_CLOSE_FAILED:
free (event->data.close_failed.addr);
break;
case ZMQ_EVENT_DISCONNECTED:
free (event->data.disconnected.addr);
break;
}
free (event_data); free (event_data);
} }

39
tests/test_monitor.cpp
View File

@@ -53,31 +53,27 @@ static void *req_socket_monitor (void *ctx)
assert (rc != -1); assert (rc != -1);
memcpy (&event, zmq_msg_data (&msg), sizeof (event)); memcpy (&event, zmq_msg_data (&msg), sizeof (event));
assert (!strcmp (event.addr, addr));
switch (event.event) { switch (event.event) {
case ZMQ_EVENT_CONNECTED: case ZMQ_EVENT_CONNECTED:
assert (event.data.connected.fd > 0); assert (event.value > 0);
assert (!strcmp (event.data.connected.addr, addr));
req_socket_events |= ZMQ_EVENT_CONNECTED; req_socket_events |= ZMQ_EVENT_CONNECTED;
req2_socket_events |= ZMQ_EVENT_CONNECTED; req2_socket_events |= ZMQ_EVENT_CONNECTED;
break; break;
case ZMQ_EVENT_CONNECT_DELAYED: case ZMQ_EVENT_CONNECT_DELAYED:
assert (event.data.connect_delayed.err != 0); assert (event.value != 0);
assert (!strcmp (event.data.connect_delayed.addr, addr));
req_socket_events |= ZMQ_EVENT_CONNECT_DELAYED; req_socket_events |= ZMQ_EVENT_CONNECT_DELAYED;
break; break;
case ZMQ_EVENT_CLOSE_FAILED: case ZMQ_EVENT_CLOSE_FAILED:
assert (event.data.close_failed.err != 0); assert (event.value != 0);
assert (!strcmp (event.data.close_failed.addr, addr));
req_socket_events |= ZMQ_EVENT_CLOSE_FAILED; req_socket_events |= ZMQ_EVENT_CLOSE_FAILED;
break; break;
case ZMQ_EVENT_CLOSED: case ZMQ_EVENT_CLOSED:
assert (event.data.closed.fd != 0); assert (event.value != 0);
assert (!strcmp (event.data.closed.addr, addr));
req_socket_events |= ZMQ_EVENT_CLOSED; req_socket_events |= ZMQ_EVENT_CLOSED;
break; break;
case ZMQ_EVENT_DISCONNECTED: case ZMQ_EVENT_DISCONNECTED:
assert (event.data.disconnected.fd != 0); assert (event.value != 0);
assert (!strcmp (event.data.disconnected.addr, addr));
req_socket_events |= ZMQ_EVENT_DISCONNECTED; req_socket_events |= ZMQ_EVENT_DISCONNECTED;
break; break;
} }
@@ -106,15 +102,14 @@ static void *req2_socket_monitor (void *ctx)
assert (rc != -1); assert (rc != -1);
memcpy (&event, zmq_msg_data (&msg), sizeof (event)); memcpy (&event, zmq_msg_data (&msg), sizeof (event));
assert (!strcmp (event.addr, addr));
switch (event.event) { switch (event.event) {
case ZMQ_EVENT_CONNECTED: case ZMQ_EVENT_CONNECTED:
assert (event.data.connected.fd > 0); assert (event.value > 0);
assert (!strcmp (event.data.connected.addr, addr));
req2_socket_events |= ZMQ_EVENT_CONNECTED; req2_socket_events |= ZMQ_EVENT_CONNECTED;
break; break;
case ZMQ_EVENT_CLOSED: case ZMQ_EVENT_CLOSED:
assert (event.data.closed.fd != 0); assert (event.value != 0);
assert (!strcmp (event.data.closed.addr, addr));
req2_socket_events |= ZMQ_EVENT_CLOSED; req2_socket_events |= ZMQ_EVENT_CLOSED;
break; break;
} }
@@ -143,30 +138,26 @@ static void *rep_socket_monitor (void *ctx)
assert (rc != -1); assert (rc != -1);
memcpy (&event, zmq_msg_data (&msg), sizeof (event)); memcpy (&event, zmq_msg_data (&msg), sizeof (event));
assert (!strcmp (event.addr, addr));
switch (event.event) { switch (event.event) {
case ZMQ_EVENT_LISTENING: case ZMQ_EVENT_LISTENING:
assert (event.data.listening.fd > 0); assert (event.value > 0);
assert (!strcmp (event.data.listening.addr, addr));
rep_socket_events |= ZMQ_EVENT_LISTENING; rep_socket_events |= ZMQ_EVENT_LISTENING;
break; break;
case ZMQ_EVENT_ACCEPTED: case ZMQ_EVENT_ACCEPTED:
assert (event.data.accepted.fd > 0); assert (event.value > 0);
assert (!strcmp (event.data.accepted.addr, addr));
rep_socket_events |= ZMQ_EVENT_ACCEPTED; rep_socket_events |= ZMQ_EVENT_ACCEPTED;
break; break;
case ZMQ_EVENT_CLOSE_FAILED: case ZMQ_EVENT_CLOSE_FAILED:
assert (event.data.close_failed.err != 0); assert (event.value != 0);
assert (!strcmp (event.data.close_failed.addr, addr));
rep_socket_events |= ZMQ_EVENT_CLOSE_FAILED; rep_socket_events |= ZMQ_EVENT_CLOSE_FAILED;
break; break;
case ZMQ_EVENT_CLOSED: case ZMQ_EVENT_CLOSED:
assert (event.data.closed.fd != 0); assert (event.value != 0);
assert (!strcmp (event.data.closed.addr, addr));
rep_socket_events |= ZMQ_EVENT_CLOSED; rep_socket_events |= ZMQ_EVENT_CLOSED;
break; break;
case ZMQ_EVENT_DISCONNECTED: case ZMQ_EVENT_DISCONNECTED:
assert (event.data.disconnected.fd != 0); assert (event.value != 0);
assert (!strcmp (event.data.disconnected.addr, addr));
rep_socket_events |= ZMQ_EVENT_DISCONNECTED; rep_socket_events |= ZMQ_EVENT_DISCONNECTED;
break; break;
} }