From 92c7c18367f91c6341fc617026f5e25000466b05 Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Sun, 20 Mar 2011 11:50:51 +0100 Subject: Message atomicity problem solved in PUB socket When new peer connects to a PUB socket while it is in the middle of sending of multi-part messages, it gets just the remaining part of the message, i.e. message atomicity is broken. This patch drops the tail part of the message and starts sending to the peer only when new message is started. Signed-off-by: Martin Sustrik --- src/dist.hpp | 15 +++++++++++++++ 1 file changed, 15 insertions(+) (limited to 'src/dist.hpp') diff --git a/src/dist.hpp b/src/dist.hpp index 7eef4ad..ad9767a 100644 --- a/src/dist.hpp +++ b/src/dist.hpp @@ -21,6 +21,8 @@ #ifndef __ZMQ_DIST_HPP_INCLUDED__ #define __ZMQ_DIST_HPP_INCLUDED__ +#include + #include "array.hpp" #include "pipe.hpp" @@ -51,10 +53,23 @@ namespace zmq // fails. In such a case false is returned. bool write (class writer_t *pipe_, zmq_msg_t *msg_); + // Put the message to all active pipes. + void distribute (zmq_msg_t *msg_, int flags_); + + // Plug in all the delayed pipes. + void clear_new_pipes (); + // List of outbound pipes. typedef array_t pipes_t; pipes_t pipes; + // List of new pipes that were not yet inserted into 'pipes' list. + // These pipes are moves to 'pipes' list once the current multipart + // message is fully sent. This way we avoid sending incomplete messages + // to peers. + typedef std::vector new_pipes_t; + new_pipes_t new_pipes; + // Number of active pipes. All the active pipes are located at the // beginning of the pipes array. pipes_t::size_type active; -- cgit v1.2.3