Move the pipe termination code to socket_base_t

So far, the pipe termination code was spread among socket type
classes, fair queuer, load balancer, etc. This patch moves
all the associated logic to a single place.

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

1 files changed, 2 insertions, 23 deletions

diff --git a/src/lb.cpp b/src/lb.cpp
index 2ba902a..7aeef9e 100644
--- a/src/lb.cpp
+++ b/src/lb.cpp
@@ -21,16 +21,13 @@
 #include "lb.hpp"
 #include "pipe.hpp"
 #include "err.hpp"
-#include "own.hpp"
 #include "msg.hpp"
 
-zmq::lb_t::lb_t (own_t *sink_) :
+zmq::lb_t::lb_t () :
     active (0),
     current (0),
     more (false),
-    dropping (false),
-    sink (sink_),
-    terminating (false)
+    dropping (false)
 {
 }
 
@@ -44,21 +41,6 @@ void zmq::lb_t::attach (pipe_t *pipe_)
     pipes.push_back (pipe_);
     pipes.swap (active, pipes.size () - 1);
     active++;
-
-    if (terminating) {
-        sink->register_term_acks (1);
-        pipe_->terminate ();
-    }
-}
-
-void zmq::lb_t::terminate ()
-{
-    zmq_assert (!terminating);
-    terminating = true;
-
-    sink->register_term_acks ((int) pipes.size ());
-    for (pipes_t::size_type i = 0; i != pipes.size (); i++)
-        pipes [i]->terminate ();
 }
 
 void zmq::lb_t::terminated (pipe_t *pipe_)
@@ -78,9 +60,6 @@ void zmq::lb_t::terminated (pipe_t *pipe_)
             current = 0;
     }
     pipes.erase (pipe_);
-
-    if (terminating)
-        sink->unregister_term_ack ();
 }
 
 void zmq::lb_t::activated (pipe_t *pipe_)