Fix assertion due to doOutput control being sent after local connection closed.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@740872 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 9 insertions, 6 deletions

diff --git a/qpid/cpp/src/qpid/cluster/Connection.cpp b/qpid/cpp/src/qpid/cluster/Connection.cpp
index a71950ef1d..2a831ae435 100644
--- a/qpid/cpp/src/qpid/cluster/Connection.cpp
+++ b/qpid/cpp/src/qpid/cluster/Connection.cpp
@@ -112,7 +112,7 @@ void Connection::received(framing::AMQFrame& f) {
                 cluster.addShadowConnection(this);
             AMQFrame ok((ConnectionCloseOkBody()));
             connection.getOutput().send(ok);
-            output.setOutputHandler(discardHandler);
+            output.closeOutput(discardHandler);
             catchUp = false;
         }
         else
@@ -165,7 +165,7 @@ void Connection::closed() {
             // This was a local replicated connection. Multicast a deliver
             // closed and process any outstanding frames from the cluster
             // until self-delivery of deliver-close.
-            output.setOutputHandler(discardHandler);
+            output.closeOutput(discardHandler);
             cluster.getMulticast().mcastControl(ClusterConnectionDeliverCloseBody(), self);
         }
     }
diff --git a/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp b/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp
index dbed75beed..45a369eea9 100644
--- a/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp
+++ b/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp
@@ -34,7 +34,8 @@ using namespace framing;
 
 OutputInterceptor::OutputInterceptor(
     cluster::Connection& p, sys::ConnectionOutputHandler& h)
-    : parent(p), next(&h), sent(), writeEstimate(p.getCluster().getWriteEstimate()),
+    : parent(p), closing(false), next(&h), sent(),
+      writeEstimate(p.getCluster().getWriteEstimate()),
       moreOutput(), doingOutput()
 {}
 
@@ -54,7 +55,7 @@ void OutputInterceptor::activateOutput() {
         sys::Mutex::ScopedLock l(lock);
         next->activateOutput();
     }
-    else {
+    else if (!closing) {        // Don't send do ouput after output stopped.
         QPID_LOG(trace,  parent << " activateOutput - sending doOutput");
         moreOutput = true;
         sendDoOutput();
@@ -117,8 +118,9 @@ void OutputInterceptor::sendDoOutput() {
     QPID_LOG(trace, parent << "Send doOutput request for " << request);
 }
 
-void OutputInterceptor::setOutputHandler(sys::ConnectionOutputHandler& h) {
+void OutputInterceptor::closeOutput(sys::ConnectionOutputHandler& h) {
     sys::Mutex::ScopedLock l(lock);
+    closing = true;
     next = &h;
 }
 
diff --git a/qpid/cpp/src/qpid/cluster/OutputInterceptor.h b/qpid/cpp/src/qpid/cluster/OutputInterceptor.h
index 6cf381178d..c080a419e1 100644
--- a/qpid/cpp/src/qpid/cluster/OutputInterceptor.h
+++ b/qpid/cpp/src/qpid/cluster/OutputInterceptor.h
@@ -53,7 +53,7 @@ class OutputInterceptor : public sys::ConnectionOutputHandler, sys::LatencyMetri
     // Intercept doOutput requests on Connection.
     bool doOutput();
 
-    void setOutputHandler(sys::ConnectionOutputHandler& h);
+    void closeOutput(sys::ConnectionOutputHandler& h);
 
     cluster::Connection& parent;
     
@@ -63,6 +63,7 @@ class OutputInterceptor : public sys::ConnectionOutputHandler, sys::LatencyMetri
     void sendDoOutput();
 
     mutable sys::Mutex lock;
+    bool closing;
     sys::ConnectionOutputHandler* next;
     size_t sent;
     size_t lastDoOutput;