From 5d4f6b18cd57897cc0e77e474118e104a0d5cfc3 Mon Sep 17 00:00:00 2001
From: Martin Hurton <hurtonm@gmail.com>
Date: Mon, 1 Mar 2010 16:55:13 +0100
Subject: Implement flow control for ZMQ_P2P sockets

---
 src/p2p.cpp | 21 +++++++++++++++------
 1 file changed, 15 insertions(+), 6 deletions(-)

(limited to 'src/p2p.cpp')

diff --git a/src/p2p.cpp b/src/p2p.cpp
index 728854b..f81c6c4 100644
--- a/src/p2p.cpp
+++ b/src/p2p.cpp
@@ -47,6 +47,7 @@ void zmq::p2p_t::xattach_pipes (class reader_t *inpipe_,
     zmq_assert (!inpipe && !outpipe);
     inpipe = inpipe_;
     outpipe = outpipe_;
+    outpipe_alive = true;
 }
 
 void zmq::p2p_t::xdetach_inpipe (class reader_t *pipe_)
@@ -75,7 +76,8 @@ void zmq::p2p_t::xrevive (class reader_t *pipe_)
 
 void zmq::p2p_t::xrevive (class writer_t *pipe_)
 {
-    zmq_not_implemented ();
+    zmq_assert (!outpipe_alive);
+    outpipe_alive = true;
 }
 
 int zmq::p2p_t::xsetsockopt (int option_, const void *optval_,
@@ -87,13 +89,17 @@ int zmq::p2p_t::xsetsockopt (int option_, const void *optval_,
 
 int zmq::p2p_t::xsend (zmq_msg_t *msg_, int flags_)
 {
-    if (!outpipe) {
+    if (outpipe == NULL || !outpipe_alive) {
+        errno = EAGAIN;
+        return -1;
+    }
+
+    if (!outpipe->write (msg_)) {
+        outpipe_alive = false;
         errno = EAGAIN;
         return -1;
     }
 
-    bool written = outpipe->write (msg_);
-    zmq_assert (written);
     if (!(flags_ & ZMQ_NOFLUSH))
         outpipe->flush ();
 
@@ -132,7 +138,10 @@ bool zmq::p2p_t::xhas_in ()
 
 bool zmq::p2p_t::xhas_out ()
 {
-    //  TODO: Implement this once queue limits are in-place.
-    return true;
+    if (outpipe == NULL || !outpipe_alive)
+        return false;
+
+    outpipe_alive = outpipe->check_write ();
+    return outpipe_alive;
 }
 
-- 
cgit v1.2.3