Merge pull request #121 from minrk/pollout

fix scope of itemsout poll

src/proxy.cpp

@@ -93,18 +93,18 @@ int zmq::proxy (
  if (unlikely (rc < 0))
  return -1;
 
- // Process a control command if any
- if (control_ && items [2].revents & ZMQ_POLLIN) {
- rc = control_->recv (&msg, 0);
- if (unlikely (rc < 0))
- return -1;
-
  // Get the pollout separately because when combining this with pollin it maxes the CPU
  // because pollout shall most of the time return directly
  rc = zmq_poll (&itemsout [0], 2, 0);
  if (unlikely (rc < 0))
  return -1;
 
+ // Process a control command if any
+ if (control_ && items [2].revents & ZMQ_POLLIN) {
+ rc = control_->recv (&msg, 0);
+ if (unlikely (rc < 0))
+ return -1;
+
  moresz = sizeof more;
  rc = control_->getsockopt (ZMQ_RCVMORE, &more, &moresz);
  if (unlikely (rc < 0) || more)

tests/test_proxy_terminate.cpp

@@ -86,22 +86,41 @@ int main (void)
  rc = zmq_connect (publisher, "tcp://127.0.0.1:15564");
  assert (rc == 0);
 
+ // Start a secondary puller which reads the data out the other end
+ char buf[255];
+ void *puller = zmq_socket (ctx, ZMQ_PULL);
+ assert (puller);
+ rc = zmq_connect (puller, "tcp://127.0.0.1:15563");
+ assert (rc == 0);
+
  msleep (50);
  rc = zmq_send (publisher, "This is a test", 14, 0);
  assert (rc == 14);
 
+ rc = zmq_recv (puller, buf, 255, 0);
+ assert (rc == 14);
+
  msleep (50);
  rc = zmq_send (publisher, "This is a test", 14, 0);
  assert (rc == 14);
 
+ rc = zmq_recv (puller, buf, 255, 0);
+ assert (rc == 14);
+
  msleep (50);
  rc = zmq_send (publisher, "This is a test", 14, 0);
  assert (rc == 14);
+
+ rc = zmq_recv (puller, buf, 255, 0);
+ assert (rc == 14);
+
  rc = zmq_send (control, "TERMINATE", 9, 0);
  assert (rc == 9);
 
  rc = zmq_close (publisher);
  assert (rc == 0);
+ rc = zmq_close (puller);
+ assert (rc == 0);
  rc = zmq_close (control);
  assert (rc == 0);