summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/dist.cpp44
-rw-r--r--src/dist.hpp15
2 files changed, 54 insertions, 5 deletions
diff --git a/src/dist.cpp b/src/dist.cpp
index dd2166a..e447bc1 100644
--- a/src/dist.cpp
+++ b/src/dist.cpp
@@ -41,6 +41,13 @@ zmq::dist_t::~dist_t ()
void zmq::dist_t::attach (writer_t *pipe_)
{
+ // If we are in the middle of sending a message, let's postpone plugging
+ // in the pipe.
+ if (!terminating && more) {
+ new_pipes.push_back (pipe_);
+ return;
+ }
+
pipe_->set_event_sink (this);
pipes.push_back (pipe_);
@@ -84,13 +91,30 @@ void zmq::dist_t::activated (writer_t *pipe_)
int zmq::dist_t::send (zmq_msg_t *msg_, int flags_)
{
+ // Is this end of a multipart message?
+ bool msg_more = msg_->flags & ZMQ_MSG_MORE;
+
+ // Push the message to active pipes.
+ distribute (msg_, flags_);
+
+ // If mutlipart message is fully sent, activate new pipes.
+ if (more && !msg_more)
+ clear_new_pipes ();
+
+ more = msg_more;
+
+ return 0;
+}
+
+void zmq::dist_t::distribute (zmq_msg_t *msg_, int flags_)
+{
// If there are no active pipes available, simply drop the message.
if (active == 0) {
int rc = zmq_msg_close (msg_);
zmq_assert (rc == 0);
rc = zmq_msg_init (msg_);
zmq_assert (rc == 0);
- return 0;
+ return;
}
msg_content_t *content = (msg_content_t*) msg_->content;
@@ -102,7 +126,7 @@ int zmq::dist_t::send (zmq_msg_t *msg_, int flags_)
i++;
int rc = zmq_msg_init (msg_);
zmq_assert (rc == 0);
- return 0;
+ return;
}
// Optimisation for the case when there's only a single pipe
@@ -115,7 +139,7 @@ int zmq::dist_t::send (zmq_msg_t *msg_, int flags_)
}
int rc = zmq_msg_init (msg_);
zmq_assert (rc == 0);
- return 0;
+ return;
}
// There are at least 2 destinations for the message. That means we have
@@ -139,8 +163,6 @@ int zmq::dist_t::send (zmq_msg_t *msg_, int flags_)
// Detach the original message from the data buffer.
int rc = zmq_msg_init (msg_);
zmq_assert (rc == 0);
-
- return 0;
}
bool zmq::dist_t::has_out ()
@@ -160,3 +182,15 @@ bool zmq::dist_t::write (class writer_t *pipe_, zmq_msg_t *msg_)
return true;
}
+void zmq::dist_t::clear_new_pipes ()
+{
+ for (new_pipes_t::iterator it = new_pipes.begin (); it != new_pipes.end ();
+ ++it) {
+ (*it)->set_event_sink (this);
+ pipes.push_back (*it);
+ pipes.swap (active, pipes.size () - 1);
+ active++;
+ }
+ new_pipes.clear ();
+}
+
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 <vector>
+
#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 <class writer_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 <class writer_t*> 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;