checking for available messages added to ypipe/pipe

src/pipe.cpp

@@ -36,6 +36,17 @@ zmq::reader_t::~reader_t ()
 {
 }
 
+bool zmq::reader_t::check_read ()
+{
+    //  Check if there's an item in the pipe.
+    if (pipe->check_read ())
+        return true;
+
+    //  If not, deactivate the pipe.
+    endpoint->kill (this);
+    return false;
+}
+
 bool zmq::reader_t::read (zmq_msg_t *msg_)
 {
     if (!pipe->read (msg_)) {

src/pipe.hpp

@@ -42,6 +42,9 @@ namespace zmq
 
         void set_endpoint (i_endpoint *endpoint_);
 
+        //  Returns true if there is at least one message to read in the pipe.
+        bool check_read ();
+
         //  Reads a message to the underlying pipe.
         bool read (zmq_msg_t *msg_);
 

src/ypipe.hpp

@@ -106,16 +106,12 @@ namespace zmq
             return true;
         }
 
-        //  Reads an item from the pipe. Returns false if there is no value.
-        //  available.
-        inline bool read (T *value_)
+        //  Check whether item is available for reading.
+        inline bool check_read ()
         {
-            //  Was the value prefetched already? If so, return it.
-            if (&queue.front () != r) {
-                 *value_ = queue.front ();
-                 queue.pop ();
+            //  Was the value prefetched already? If so, return.
+            if (&queue.front () != r)
                  return true;
-            }
 
             //  There's no prefetched value, so let us prefetch more values.
             //  (Note that D is a template parameter. Becaue of that one of
@@ -165,6 +161,18 @@ namespace zmq
                     return false;
             }
 
+            //  There was at least one value prefetched.
+            return true;
+        }
+
+        //  Reads an item from the pipe. Returns false if there is no value.
+        //  available.
+        inline bool read (T *value_)
+        {
+            //  Try to prefetch a value.
+            if (!check_read ())
+                return false;
+
             //  There was at least one value prefetched.
             //  Return it to the caller.
             *value_ = queue.front ();