QPID-1040: Patch from Ted Ross: Asynchronous Connector

Code to allow non-blocking connection of new sockets


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

2 files changed, 64 insertions, 1 deletions

diff --git a/cpp/src/qpid/sys/posix/AsynchIO.cpp b/cpp/src/qpid/sys/posix/AsynchIO.cpp
index cedad5c011..9dcb841992 100644
--- a/cpp/src/qpid/sys/posix/AsynchIO.cpp
+++ b/cpp/src/qpid/sys/posix/AsynchIO.cpp
@@ -97,6 +97,57 @@ void AsynchAcceptor::readable(DispatchHandle& h) {
 }
 
 /*
+ * Asynch Connector
+ */
+
+AsynchConnector::AsynchConnector(const Socket& s,
+                                 Poller::shared_ptr poller,
+                                 std::string hostname,
+                                 uint16_t port,
+                                 ConnectedCallback connCb,
+                                 FailedCallback failCb) :
+    DispatchHandle(s,
+                   0,
+                   boost::bind(&AsynchConnector::connComplete, this, _1),
+                   boost::bind(&AsynchConnector::connComplete, this, _1)),
+    connCallback(connCb),
+    failCallback(failCb),
+    socket(s)
+{
+    socket.setNonblocking();
+    try {
+        socket.connect(hostname, port);
+        startWatch(poller);
+    } catch(std::exception& e) {
+        failure(-1, std::string(e.what()));
+    }
+}
+
+void AsynchConnector::connComplete(DispatchHandle& h)
+{
+    int errCode = socket.getError();
+
+    h.stopWatch();
+    if (errCode == 0) {
+        connCallback(socket);
+	DispatchHandle::doDelete();
+    } else {
+        failure(errCode, std::string(strerror(errCode)));
+    }
+}
+
+void AsynchConnector::failure(int errCode, std::string message)
+{
+    if (failCallback)
+        failCallback(errCode, message);
+
+    socket.close();
+    delete &socket;
+
+    DispatchHandle::doDelete();
+}
+
+/*
  * Asynch reader/writer
  */
 AsynchIO::AsynchIO(const Socket& s,
diff --git a/cpp/src/qpid/sys/posix/Socket.cpp b/cpp/src/qpid/sys/posix/Socket.cpp
index 5f10cd84c2..67f6b6db4c 100644
--- a/cpp/src/qpid/sys/posix/Socket.cpp
+++ b/cpp/src/qpid/sys/posix/Socket.cpp
@@ -148,7 +148,8 @@ void Socket::connect(const std::string& host, int port) const
     if (hp == 0)
         throw Exception(QPID_MSG("Cannot resolve " << host << ": " << h_errstr(h_errno)));
     ::memcpy(&name.sin_addr.s_addr, hp->h_addr_list[0], hp->h_length);
-    if (::connect(socket, (struct sockaddr*)(&name), sizeof(name)) < 0)
+    if ((::connect(socket, (struct sockaddr*)(&name), sizeof(name)) < 0) &&
+        (errno != EINPROGRESS))
         throw qpid::Exception(QPID_MSG(strError(errno) << ": " << host << ":" << port));
 }
 
@@ -257,6 +258,17 @@ uint16_t Socket::getRemotePort() const
     return atoi(getService(impl->fd, true).c_str());
 }
 
+int Socket::getError() const
+{
+    int       result;
+    socklen_t rSize = sizeof (result);
+
+    if (::getsockopt(impl->fd, SOL_SOCKET, SO_ERROR, &result, &rSize) < 0)
+        throw QPID_POSIX_ERROR(errno);
+
+    return result;
+}
+
 void Socket::configure(const Configuration& c)
 {
     c.configurePosixTcpSocket(impl->fd);