QPID-2511: Altered shutdown sequence in connectors; after connect() has returned there will always now be a shutdown callback made and only at that point is it guaranteed that there will be no further io callbacks. ConnectionImpl consequently waits for this before allowing itself to be deleted.

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

1 files changed, 14 insertions, 19 deletions

diff --git a/cpp/src/qpid/client/TCPConnector.cpp b/cpp/src/qpid/client/TCPConnector.cpp
index 1a245fe2c8..3ead7e2ce0 100644
--- a/cpp/src/qpid/client/TCPConnector.cpp
+++ b/cpp/src/qpid/client/TCPConnector.cpp
@@ -105,7 +105,7 @@ void TCPConnector::connected(const Socket&) {
                        boost::bind(&TCPConnector::readbuff, this, _1, _2),
                        boost::bind(&TCPConnector::eof, this, _1),
                        boost::bind(&TCPConnector::eof, this, _1),
-                       0, // closed
+                       boost::bind(&TCPConnector::socketClosed, this, _1, _2),
                        0, // nobuffs
                        boost::bind(&TCPConnector::writebuff, this, _1));
     start(aio);
@@ -128,28 +128,28 @@ void TCPConnector::initAmqp() {
 }
 
 void TCPConnector::connectFailed(const std::string& msg) {
-    QPID_LOG(warning, "Connecting failed: " << msg);
+    QPID_LOG(warning, "Connect failed: " << msg);
     socket.close();
-    if (!closed && shutdownHandler) {
+    if (!closed)
         closed = true;
+    if (shutdownHandler)
         shutdownHandler->shutdown();
-    }
 }
 
-bool TCPConnector::closeInternal() {
+void TCPConnector::close() {
     Mutex::ScopedLock l(lock);
-    bool ret = !closed;
-    closed = true;
-    if (ret) {
+    if (!closed) {
+        closed = true;
         if (aio)
-            aio->queueForDeletion();
-        socket.close();
+            aio->queueWriteClose();
     }
-    return ret;
 }
 
-void TCPConnector::close() {
-    closeInternal();
+void TCPConnector::socketClosed(AsynchIO&, const Socket&) {
+    if (aio)
+        aio->queueForDeletion();
+    if (shutdownHandler)
+        shutdownHandler->shutdown();
 }
 
 void TCPConnector::abort() {
@@ -214,11 +214,6 @@ void TCPConnector::send(AMQFrame& frame) {
     }
 }
 
-void TCPConnector::handleClosed() {
-    if (closeInternal() && shutdownHandler)
-        shutdownHandler->shutdown();
-}
-
 void TCPConnector::writebuff(AsynchIO& /*aio*/) 
 {
     // It's possible to be disconnected and be writable
@@ -315,7 +310,7 @@ void TCPConnector::writeDataBlock(const AMQDataBlock& data) {
 }
 
 void TCPConnector::eof(AsynchIO&) {
-    handleClosed();
+    close();
 }
 
 void TCPConnector::activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer> sl)