QPID-4447: C++ Client can hang during connect if heartbeat disconnect fires

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

5 files changed, 31 insertions, 2 deletions

diff --git a/qpid/cpp/src/qpid/client/TCPConnector.cpp b/qpid/cpp/src/qpid/client/TCPConnector.cpp
index b92f342b74..783742764b 100644
--- a/qpid/cpp/src/qpid/client/TCPConnector.cpp
+++ b/qpid/cpp/src/qpid/client/TCPConnector.cpp
@@ -151,6 +151,11 @@ void TCPConnector::socketClosed(AsynchIO&, const Socket&) {
         shutdownHandler->shutdown();
 }
 
+void TCPConnector::connectAborted() {
+    connector->stop();
+    connectFailed("Connection timedout");
+}
+
 void TCPConnector::abort() {
     // Can't abort a closed connection
     if (!closed) {
@@ -159,8 +164,7 @@ void TCPConnector::abort() {
             aio->requestCallback(boost::bind(&TCPConnector::eof, this, _1));
         } else if (connector) {
             // We're still connecting
-            connector->stop();
-            connectFailed("Connection timedout");
+            connector->requestCallback(boost::bind(&TCPConnector::connectAborted, this));
         }
     }
 }
diff --git a/qpid/cpp/src/qpid/client/TCPConnector.h b/qpid/cpp/src/qpid/client/TCPConnector.h
index a90dffd3ef..63af3b878a 100644
--- a/qpid/cpp/src/qpid/client/TCPConnector.h
+++ b/qpid/cpp/src/qpid/client/TCPConnector.h
@@ -80,6 +80,7 @@ class TCPConnector : public Connector, public sys::Codec
     void close();
     void send(framing::AMQFrame& frame);
     void abort();
+    void connectAborted();
 
     void setInputHandler(framing::InputHandler* handler);
     void setShutdownHandler(sys::ShutdownHandler* handler);
diff --git a/qpid/cpp/src/qpid/sys/AsynchIO.h b/qpid/cpp/src/qpid/sys/AsynchIO.h
index 8eed72d40d..a531ee1dbb 100644
--- a/qpid/cpp/src/qpid/sys/AsynchIO.h
+++ b/qpid/cpp/src/qpid/sys/AsynchIO.h
@@ -58,6 +58,7 @@ class AsynchConnector {
 public:
     typedef boost::function1<void, const Socket&> ConnectedCallback;
     typedef boost::function3<void, const Socket&, int, const std::string&> FailedCallback;
+    typedef boost::function1<void, AsynchConnector&> RequestCallback;
 
     // Call create() to allocate a new AsynchConnector object with the
     // specified poller, addressing, and callbacks.
@@ -72,6 +73,7 @@ public:
                                    FailedCallback failCb);
     virtual void start(boost::shared_ptr<Poller> poller) = 0;
     virtual void stop() {};
+    virtual void requestCallback(RequestCallback) = 0;
 protected:
     AsynchConnector() {}
     virtual ~AsynchConnector() {}
diff --git a/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp b/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
index 7eb25fd861..2c17cc001c 100644
--- a/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
+++ b/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
@@ -143,6 +143,7 @@ class AsynchConnector : public qpid::sys::AsynchConnector,
 
 private:
     void connComplete(DispatchHandle& handle);
+    void requestedCall(RequestCallback rCb);
 
 private:
     ConnectedCallback connCallback;
@@ -158,6 +159,7 @@ public:
                     FailedCallback failCb);
     void start(Poller::shared_ptr poller);
     void stop();
+    void requestCallback(RequestCallback rCb);
 };
 
 AsynchConnector::AsynchConnector(const Socket& s,
@@ -191,6 +193,18 @@ void AsynchConnector::stop()
     stopWatch();
 }
 
+void AsynchConnector::requestCallback(RequestCallback callback) {
+    // TODO creating a function object every time isn't all that
+    // efficient - if this becomes heavily used do something better (what?)
+    assert(callback);
+    DispatchHandle::call(boost::bind(&AsynchConnector::requestedCall, this, callback));
+}
+
+void AsynchConnector::requestedCall(RequestCallback callback) {
+    assert(callback);
+    callback(*this);
+}
+
 void AsynchConnector::connComplete(DispatchHandle& h)
 {
     int errCode = socket.getError();
diff --git a/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp b/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp
index 7dcc5c5846..d8aa6efda7 100644
--- a/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp
+++ b/qpid/cpp/src/qpid/sys/windows/AsynchIO.cpp
@@ -198,6 +198,7 @@ public:
                     ConnectedCallback connCb,
                     FailedCallback failCb = 0);
     void start(Poller::shared_ptr poller);
+    void requestCallback(RequestCallback rCb);
 };
 
 AsynchConnector::AsynchConnector(const Socket& sock,
@@ -223,6 +224,13 @@ void AsynchConnector::start(Poller::shared_ptr)
     }
 }
 
+// This can never be called in the current windows code as connect
+// is blocking and requestCallback only makes sense if connect is
+// non-blocking with the results returned via a poller callback.
+void AsynchConnector::requestCallback(RequestCallback rCb)
+{
+}
+
 } // namespace windows
 
 AsynchAcceptor* AsynchAcceptor::create(const Socket& s,