Destruction of session is delayed till both in & out pipes are closed

src/pipe.cpp

@@ -130,11 +130,6 @@ void zmq::reader_t::terminate ()
     send_pipe_term (writer);
 }
 
-bool zmq::reader_t::is_terminating ()
-{
-    return terminating;
-}
-
 void zmq::reader_t::process_revive ()
 {
     //  Forward the event to the sink (either socket or session).

src/pipe.hpp

@@ -67,10 +67,6 @@ namespace zmq
         //  Ask pipe to terminate.
         void terminate ();
 
-        //  Returns true if the pipe is already terminating
-        //  (say if delimiter was already read).
-        bool is_terminating ();
-
     private:
 
         reader_t (class object_t *parent_, pipe_t *pipe_, uint64_t lwm_);

src/session.cpp

@@ -71,7 +71,8 @@ zmq::session_t::~session_t ()
 
 bool zmq::session_t::is_terminable ()
 {
-    return in_pipe->is_terminating ();
+    //  The session won't send term_ack until both in & out pipe are closed.
+    return !in_pipe && !out_pipe;
 }
 
 bool zmq::session_t::read (::zmq_msg_t *msg_)
@@ -81,8 +82,6 @@ bool zmq::session_t::read (::zmq_msg_t *msg_)
 
     if (!in_pipe->read (msg_)) {
         active = false;
-        if (in_pipe->is_terminating ())
-            finalise ();
         return false;
     }
 
@@ -179,11 +178,13 @@ void zmq::session_t::terminated (reader_t *pipe_)
 {
     active = false;
     in_pipe = NULL;
+    finalise ();
 }
 
 void zmq::session_t::terminated (writer_t *pipe_)
 {
     out_pipe = NULL;
+    finalise ();
 }
 
 void zmq::session_t::activated (reader_t *pipe_)