1 files changed, 46 insertions, 16 deletions

diff --git a/src/fq.cpp b/src/fq.cpp
index 9028853..2c3a9d9 100644
--- a/src/fq.cpp
+++ b/src/fq.cpp
@@ -22,29 +22,42 @@
 #include "fq.hpp"
 #include "pipe.hpp"
 #include "err.hpp"
+#include "own.hpp"
 
-zmq::fq_t::fq_t () :
+zmq::fq_t::fq_t (own_t *sink_) :
     active (0),
     current (0),
-    more (false)
+    more (false),
+    sink (sink_),
+    terminating (false)
 {
 }
 
 zmq::fq_t::~fq_t ()
 {
-    for (pipes_t::size_type i = 0; i != pipes.size (); i++)
-        pipes [i]->term ();
+    zmq_assert (pipes.empty ());
 }
 
 void zmq::fq_t::attach (reader_t *pipe_)
 {
+    pipe_->set_event_sink (this);
+
     pipes.push_back (pipe_);
     pipes.swap (active, pipes.size () - 1);
     active++;
+
+    //  If we are already terminating, ask the pipe to terminate straight away.
+    if (terminating) {
+        sink->register_term_acks (1);
+        pipe_->terminate ();
+    }
 }
 
-void zmq::fq_t::detach (reader_t *pipe_)
+void zmq::fq_t::terminated (reader_t *pipe_)
 {
+    //  TODO: This is a problem with session-initiated termination. It breaks
+    //  message atomicity. However, for socket initiated termination it's
+    //  just fine.
     zmq_assert (!more || pipes [current] != pipe_);
 
     //  Remove the pipe from the list; adjust number of active pipes
@@ -55,18 +68,22 @@ void zmq::fq_t::detach (reader_t *pipe_)
             current = 0;
     }
     pipes.erase (pipe_);
+
+    if (terminating)
+        sink->unregister_term_ack ();
 }
 
-void zmq::fq_t::kill (reader_t *pipe_)
+void zmq::fq_t::terminate ()
 {
-    //  Move the pipe to the list of inactive pipes.
-    active--;
-    if (current == active)
-        current = 0;
-    pipes.swap (pipes.index (pipe_), active);
+    zmq_assert (!terminating);
+    terminating = true;
+
+    sink->register_term_acks (pipes.size ());
+    for (pipes_t::size_type i = 0; i != pipes.size (); i++)
+        pipes [i]->terminate ();
 }
 
-void zmq::fq_t::revive (reader_t *pipe_)
+void zmq::fq_t::activated (reader_t *pipe_)
 {
     //  Move the pipe to the list of active pipes.
     pipes.swap (pipes.index (pipe_), active);
@@ -84,10 +101,14 @@ int zmq::fq_t::recv (zmq_msg_t *msg_, int flags_)
         //  Try to fetch new message. If we've already read part of the message
         //  subsequent part should be immediately available.
         bool fetched = pipes [current]->read (msg_);
+
+        //  Check the atomicity of the message. If we've already received the
+        //  first part of the message we should get the remaining parts
+        //  without blocking.
         zmq_assert (!(more && !fetched));
 
-        //  Note that when message is not fetched, current pipe is killed and
-        //  replaced by another active pipe. Thus we don't have to increase
+        //  Note that when message is not fetched, current pipe is deactivated
+        //  and replaced by another active pipe. Thus we don't have to increase
         //  the 'current' pointer.
         if (fetched) {
             more = msg_->flags & ZMQ_MSG_MORE;
@@ -98,6 +119,12 @@ int zmq::fq_t::recv (zmq_msg_t *msg_, int flags_)
             }
             return 0;
         }
+        else {
+            active--;
+            pipes.swap (current, active);
+            if (current == active)
+                current = 0;
+        }
     }
 
     //  No message is available. Initialise the output parameter
@@ -120,8 +147,11 @@ bool zmq::fq_t::has_in ()
     for (int count = active; count != 0; count--) {
         if (pipes [current]->check_read ())
             return true;
-        current++;
-        if (current >= active)
+
+        //  Deactivate the pipe.
+        active--;
+        pipes.swap (current, active);
+        if (current == active)
             current = 0;
     }