Fix: decoding error causes cluster broker to exit.

An error decoding a connection was causing the broker::Connection
object to be deleted in the connection IO thread rather than the
cluster deliver thread, causing an exit with "critical Modified
cluster state outside of cluster context"


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

1 files changed, 16 insertions, 4 deletions

diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index a375a65851..909ff68d92 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -159,6 +159,11 @@ bool Connection::doOutput() {
 
 // Received from a directly connected client.
 void Connection::received(framing::AMQFrame& f) {
+    if (!connection.get()) {
+        QPID_LOG(warning, cluster << " ignoring frame on closed connection "
+                 << *this << ": " << f);
+        return;
+    }
     QPID_LOG(trace, cluster << " RECV " << *this << ": " << f);
     if (isLocal()) {            // Local catch-up connection.
         currentChannel = f.getChannel();
@@ -231,7 +236,7 @@ void Connection::closed() {
         }
         else if (isUpdated()) {
             QPID_LOG(debug, cluster << " closed update connection " << *this);
-            connection->closed();
+            if (connection.get()) connection->closed();
         }
         else if (isLocal()) {
             QPID_LOG(debug, cluster << " local close of replicated connection " << *this);
@@ -250,13 +255,21 @@ void Connection::closed() {
 // Self-delivery of close message, close the connection.
 void Connection::deliverClose () {
     assert(!catchUp);
-    connection->closed();
+    if (connection.get()) {
+        connection->closed();
+        // Ensure we delete the broker::Connection in the deliver thread.
+        connection.reset();
+    }
     cluster.erase(self);
 }
 
 // The connection has been killed for misbehaving
 void Connection::abort() {
-    if (connection.get()) connection->abort();
+    if (connection.get()) {
+        connection->abort();
+        // Ensure we delete the broker::Connection in the deliver thread.
+        connection.reset();
+    }
     cluster.erase(self);
 }
 
@@ -324,7 +337,6 @@ void Connection::sessionState(
     const SequenceSet& unknownCompleted,
     const SequenceSet& receivedIncomplete)
 {
-    
     sessionState().setState(
         replayStart,
         sendCommandPoint,