Merge pull request #583 from ianbarber/master

Small stream engine issue and test_monitor tidy up

src/stream_engine.cpp

@@ -660,6 +660,12 @@ void zmq::stream_engine_t::mechanism_ready ()
         msg_t identity;
         mechanism->peer_identity (&identity);
         const int rc = session->push_msg (&identity);
+        if (rc == -1 && errno == EAGAIN) {
+            // If the write is failing at this stage with
+            // an EAGAIN the pipe must be being shut down,
+            // so we can just bail out of the identity set.
+            return;
+        }
         errno_assert (rc == 0);
     }
 

tests/test_monitor.cpp

@@ -42,11 +42,13 @@ static bool read_msg(void* s, zmq_event_t& event, std::string& ep)
     rc = zmq_msg_recv (&msg1, s, 0);
     if (rc == -1 && zmq_errno() == ETERM)
         return true ;
+
     assert (rc != -1);
     assert (zmq_msg_more(&msg1) != 0);
     rc = zmq_msg_recv (&msg2, s, 0);
     if (rc == -1 && zmq_errno() == ETERM)
         return true;
+
     assert (rc != -1);
     assert (zmq_msg_more(&msg2) == 0);
     // copy binary data to event struct
@@ -204,9 +206,6 @@ int main (void)
     rc = pthread_create (&threads [0], NULL, rep_socket_monitor, ctx);
     assert (rc == 0);
     
-    rc = zmq_bind (rep, addr.c_str());
-    assert (rc == 0);
-
     // REQ socket
     req = zmq_socket (ctx, ZMQ_REQ);
     assert (req);
@@ -216,6 +215,11 @@ int main (void)
     assert (rc == 0);
     rc = pthread_create (&threads [1], NULL, req_socket_monitor, ctx);
     assert (rc == 0);
+    sleep(1);
+
+    // Bind REQ and REP
+    rc = zmq_bind (rep, addr.c_str());
+    assert (rc == 0);
 
     rc = zmq_connect (req, addr.c_str());
     assert (rc == 0);
@@ -271,4 +275,3 @@ int main (void)
 
     return 0 ;
 }
-