From 89783c37d2b8a7b5519eab7922b460449aa0bf3f Mon Sep 17 00:00:00 2001
From: Martin Sustrik <sustrik@250bpm.com>
Date: Wed, 19 May 2010 06:31:57 +0200
Subject: incomplete messages can be stored in ypipe

---
 src/pipe.cpp | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

(limited to 'src/pipe.cpp')

diff --git a/src/pipe.cpp b/src/pipe.cpp
index 61ddf38..f592f8c 100644
--- a/src/pipe.cpp
+++ b/src/pipe.cpp
@@ -162,7 +162,7 @@ bool zmq::writer_t::write (zmq_msg_t *msg_)
         return false;
     }
 
-    pipe->write (*msg_);
+    pipe->write (*msg_, msg_->flags & ZMQ_MSG_MORE);
     if (!(msg_->flags & ZMQ_MSG_MORE))
         msgs_written++;
     return true;
@@ -172,11 +172,9 @@ void zmq::writer_t::rollback ()
 {
     zmq_msg_t msg;
 
+    //  Remove all incomplete messages from the pipe.
     while (pipe->unwrite (&msg)) {
-        if (!(msg.flags & ZMQ_MSG_MORE)) {
-            pipe->write (msg);
-            break;
-        }
+        zmq_assert (msg.flags & ZMQ_MSG_MORE);
         zmq_msg_close (&msg);
         msgs_written--;
     }
@@ -206,7 +204,7 @@ void zmq::writer_t::term ()
     const unsigned char *offset = 0;
     msg.content = (void*) (offset + ZMQ_DELIMITER);
     msg.flags = 0;
-    pipe->write (msg);
+    pipe->write (msg, false);
     pipe->flush ();
 }
 
-- 
cgit v1.2.3