generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-03-04 07:27:26 +01:00
Code Issues Projects Releases Wiki Activity

Merge pull request #1534 from somdoron/master

Browse Source 
problem: new poller field in zmq_pollitem_t is changing the api
This commit is contained in:
Constantin Rack 2015-08-17 13:40:26 +02:00
commit a96f16e3fa
3 changed files with 35 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
include/zmq.h
View File

@ -383,8 +383,8 @@ ZMQ_EXPORT int zmq_send (void *s, const void *buf, size_t len, int flags);
ZMQ_EXPORT int zmq_send_const (void *s, const void *buf, size_t len, int flags); ZMQ_EXPORT int zmq_send_const (void *s, const void *buf, size_t len, int flags);
ZMQ_EXPORT int zmq_recv (void *s, void *buf, size_t len, int flags); ZMQ_EXPORT int zmq_recv (void *s, void *buf, size_t len, int flags);
ZMQ_EXPORT int zmq_socket_monitor (void *s, const char *addr, int events); ZMQ_EXPORT int zmq_socket_monitor (void *s, const char *addr, int events);
ZMQ_EXPORT int zmq_add_poller (void *s, void *p); ZMQ_EXPORT int zmq_add_pollfd (void *s, void *p);
ZMQ_EXPORT int zmq_remove_poller (void *s, void *p); ZMQ_EXPORT int zmq_remove_pollfd (void *s, void *p);
/******************************************************************************/ /******************************************************************************/
/* I/O multiplexing. */ /* I/O multiplexing. */
@ -405,14 +405,19 @@ typedef struct zmq_pollitem_t
#endif #endif
short events; short events;
short revents; short revents;
void *poller;
} zmq_pollitem_t; } zmq_pollitem_t;
#define ZMQ_POLLITEMS_DFLT 16 #define ZMQ_POLLITEMS_DFLT 16
ZMQ_EXPORT int zmq_poll (zmq_pollitem_t *items, int nitems, long timeout); ZMQ_EXPORT int zmq_poll (zmq_pollitem_t *items, int nitems, long timeout);
ZMQ_EXPORT void *zmq_poller_new (); ZMQ_EXPORT void *zmq_pollfd_new ();
ZMQ_EXPORT int zmq_poller_close (void *p); ZMQ_EXPORT int zmq_pollfd_close (void *p);
#if defined _WIN32
ZMQ_EXPORT SOCKET zmq_pollfd_fd (void *p);
#else
ZMQ_EXPORT int zmq_pollfd_fd (void *p);
#endif
/******************************************************************************/ /******************************************************************************/
/* Message proxying */ /* Message proxying */

29
src/zmq.cpp
View File

@ -562,9 +562,9 @@ int zmq_recviov (void *s_, iovec *a_, size_t *count_, int flags_)
return nread; return nread;
} }
// Add/remove poller from a socket // Add/remove pollfd from a socket
int zmq_add_poller (void *s_, void *p_) int zmq_add_pollfd (void *s_, void *p_)
{ {
if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
errno = ENOTSOCK; errno = ENOTSOCK;
@ -576,7 +576,7 @@ int zmq_add_poller (void *s_, void *p_)
return s->add_signaler(p); return s->add_signaler(p);
} }
int zmq_remove_poller (void *s_, void *p_) int zmq_remove_pollfd (void *s_, void *p_)
{ {
if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
errno = ENOTSOCK; errno = ENOTSOCK;
@ -709,16 +709,16 @@ const char *zmq_msg_gets (zmq_msg_t *msg_, const char *property_)
} }
} }
// Create poller // Create pollfd
void *zmq_poller_new () void *zmq_pollfd_new ()
{ {
return new zmq::signaler_t (); return new zmq::signaler_t ();
} }
// Close poller // Close pollfd
int zmq_poller_close (void* p) int zmq_pollfd_close (void* p)
{ {
zmq::signaler_t *s = (zmq::signaler_t*)p; zmq::signaler_t *s = (zmq::signaler_t*)p;
LIBZMQ_DELETE(s); LIBZMQ_DELETE(s);
@ -727,8 +727,11 @@ int zmq_poller_close (void* p)
} }
// Get poller fd // Get poller fd
#if defined _WIN32
zmq::fd_t zmq_poller_get_fd (void *p) SOCKET zmq_pollfd_fd (void *p)
#else
int zmq_pollfd_fd (void *p)
#endif
{ {
zmq::signaler_t *s = (zmq::signaler_t*)p; zmq::signaler_t *s = (zmq::signaler_t*)p;
return s->get_fd (); return s->get_fd ();
@ -790,14 +793,14 @@ int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
} }
if (thread_safe) { if (thread_safe) {
if (!items_ [i].poller) { if (!items_ [i].fd) {
if (pollfds != spollfds) if (pollfds != spollfds)
free (pollfds); free (pollfds);
errno = EINVAL; errno = EINVAL;
return -1; return -1;
} }
pollfds [i].fd = zmq_poller_get_fd (items_ [i].poller); pollfds [i].fd = items_ [i].fd;
} }
else { else {
size_t zmq_fd_size = sizeof (zmq::fd_t); size_t zmq_fd_size = sizeof (zmq::fd_t);
@ -974,12 +977,12 @@ int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
zmq::fd_t notify_fd; zmq::fd_t notify_fd;
if (thread_safe) { if (thread_safe) {
if (!items_ [i].poller) { if (!items_ [i].fd) {
errno = EINVAL; errno = EINVAL;
return -1; return -1;
} }
notify_fd = zmq_poller_get_fd (items_ [i].poller); notify_fd = items_ [i].fd;
} }
else { else {
size_t zmq_fd_size = sizeof (zmq::fd_t); size_t zmq_fd_size = sizeof (zmq::fd_t);

24
tests/test_thread_safe_polling.cpp
View File

@ -39,25 +39,19 @@ int main (void)
void *server = zmq_socket (ctx, ZMQ_SERVER); void *server = zmq_socket (ctx, ZMQ_SERVER);
void *server2 = zmq_socket (ctx, ZMQ_SERVER); void *server2 = zmq_socket (ctx, ZMQ_SERVER);
void *poller = zmq_poller_new (); void *pollfd = zmq_pollfd_new ();
int rc; int rc;
rc = zmq_add_poller (server, poller); rc = zmq_add_pollfd (server, pollfd);
assert (rc == 0); assert (rc == 0);
rc = zmq_add_poller (server2, poller); rc = zmq_add_pollfd (server2, pollfd);
assert (rc == 0); assert (rc == 0);
zmq_pollitem_t items[2]; zmq_pollitem_t items[] = {
{server, zmq_pollfd_fd(pollfd), ZMQ_POLLIN, 0},
items[0].socket = server; {server2, zmq_pollfd_fd(pollfd), ZMQ_POLLIN, 0}};
items[0].poller = poller;
items[0].events = ZMQ_POLLIN;
items[1].socket = server2;
items[1].poller = poller;
items[1].events = ZMQ_POLLIN;
rc = zmq_bind (server, "tcp://127.0.0.1:5560"); rc = zmq_bind (server, "tcp://127.0.0.1:5560");
assert (rc == 0); assert (rc == 0);
@ -100,13 +94,13 @@ int main (void)
rc = zmq_msg_close(&msg); rc = zmq_msg_close(&msg);
assert (rc == 0); assert (rc == 0);
rc = zmq_remove_poller (server, poller); rc = zmq_remove_pollfd (server, pollfd);
assert (rc == 0); assert (rc == 0);
rc = zmq_remove_poller (server2, poller); rc = zmq_remove_pollfd (server2, pollfd);
assert (rc == 0); assert (rc == 0);
rc = zmq_poller_close (poller); rc = zmq_pollfd_close (pollfd);
assert (rc == 0); assert (rc == 0);
rc = zmq_close (server); rc = zmq_close (server);