Implement flow control for ZMQ_REP sockets

src/rep.cpp

@@ -146,7 +146,6 @@ void zmq::rep_t::xrevive (class reader_t *pipe_)
 
 void zmq::rep_t::xrevive (class writer_t *pipe_)
 {
-    zmq_not_implemented ();
 }
 
 int zmq::rep_t::xsetsockopt (int option_, const void *optval_,
@@ -163,16 +162,17 @@ int zmq::rep_t::xsend (zmq_msg_t *msg_, int flags_)
         return -1;
     }
 
-    //  TODO: Implement this once queue limits are in-place. If the reply
-    //  overloads the buffer, connection should be torn down.
-    zmq_assert (reply_pipe->check_write ());
-
     //  Push message to the selected pipe. If requester have disconnected
     //  in the meantime, drop the reply.
     if (reply_pipe) {
         bool written = reply_pipe->write (msg_);
-        zmq_assert (written);
-        reply_pipe->flush ();
+        if (written)
+            reply_pipe->flush ();
+        else
+            //  The pipe is full; just drop the reference to
+            //  the message content.
+            //  TODO: Tear down the underlying connection.
+            zmq_msg_close (msg_);
     }
     else {
         zmq_msg_close (msg_);