Imported Upstream version 2.1.7upstream/2.1.7

1 files changed, 15 insertions, 3 deletions

diff --git a/src/xrep.cpp b/src/xrep.cpp
index 75dc30e..7317056 100644
--- a/src/xrep.cpp
+++ b/src/xrep.cpp
@@ -63,7 +63,7 @@ void zmq::xrep_t::xattach_pipes (reader_t *inpipe_, writer_t *outpipe_,
 
         if (terminating) {
             register_term_acks (1);
-            outpipe_->terminate ();        
+            outpipe_->terminate ();
         }
     }
 
@@ -102,11 +102,13 @@ void zmq::xrep_t::terminated (reader_t *pipe_)
     for (inpipes_t::iterator it = inpipes.begin (); it != inpipes.end ();
           ++it) {
         if (it->reader == pipe_) {
+            if ((inpipes_t::size_type) (it - inpipes.begin ()) < current_in)
+                current_in--;
             inpipes.erase (it);
-            if (terminating)
-                unregister_term_ack ();
             if (current_in >= inpipes.size ())
                 current_in = 0;
+            if (terminating)
+                unregister_term_ack ();
             return;
         }
     }
@@ -288,6 +290,16 @@ int zmq::xrep_t::xrecv (zmq_msg_t *msg_, int flags_)
     return -1;
 }
 
+int zmq::xrep_t::rollback (void)
+{
+    if (current_out) {
+        current_out->rollback ();
+        current_out = NULL;
+        more_out = false;
+    }
+    return 0;
+}
+
 bool zmq::xrep_t::xhas_in ()
 {
     //  There are subsequent parts of the partly-read message available.