Posix EventChannel implementation using epoll. Placeholder for kevents.

Dynamic thread pool EventChannelThreads to serve EventChannel.
Misc cleanup/enhancements.


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

3 files changed, 22 insertions, 17 deletions

diff --git a/cpp/src/qpid/apr/APRBase.cpp b/cpp/src/qpid/apr/APRBase.cpp
index c3dbda3df0..7a35f098ec 100644
--- a/cpp/src/qpid/apr/APRBase.cpp
+++ b/cpp/src/qpid/apr/APRBase.cpp
@@ -87,7 +87,8 @@ void qpid::sys::check(apr_status_t status, const std::string& file, const int li
         const int size = 50;
         char tmp[size];
         std::string msg(apr_strerror(status, tmp, size));
-        throw QpidError(APR_ERROR + ((int) status), msg, file, line);
+        throw QpidError(APR_ERROR + ((int) status), msg,
+                        qpid::SrcLine(file, line));
     }
 }
 
diff --git a/cpp/src/qpid/apr/LFProcessor.cpp b/cpp/src/qpid/apr/LFProcessor.cpp
index c8a583a34c..0187beab10 100644
--- a/cpp/src/qpid/apr/LFProcessor.cpp
+++ b/cpp/src/qpid/apr/LFProcessor.cpp
@@ -134,8 +134,8 @@ void LFProcessor::run(){
                 session->stopProcessing();
             }
         }
-    }catch(QpidError error){
-	std::cout << "Error [" << error.code << "] " << error.msg << " (" << error.file << ":" << error.line << ")" << std::endl;
+    }catch(std::exception e){
+	std::cout << e.what() << std::endl;
     }
 }
 
diff --git a/cpp/src/qpid/apr/Socket.cpp b/cpp/src/qpid/apr/Socket.cpp
index 9ef6baeb88..3cf510872f 100644
--- a/cpp/src/qpid/apr/Socket.cpp
+++ b/cpp/src/qpid/apr/Socket.cpp
@@ -27,21 +27,24 @@
 
 using namespace qpid::sys;
 
-Socket::Socket()
-{
+Socket Socket::createTcp() {
+    Socket s;
     CHECK_APR_SUCCESS(
         apr_socket_create(
-            &socket, APR_INET, SOCK_STREAM, APR_PROTO_TCP,
+            &s.socket, APR_INET, SOCK_STREAM, APR_PROTO_TCP,
             APRPool::get()));
+    return s;
 }
 
-void Socket::setTimeout(long msecs)
-{
-    apr_socket_timeout_set(socket, msecs*1000);
+Socket::Socket(apr_socket_t* s) {
+    socket = s;
 }
 
-void Socket::connect(const std::string& host, int port)
-{
+void Socket::setTimeout(Time interval) {
+    apr_socket_timeout_set(socket, interval/TIME_USEC);
+}
+
+void Socket::connect(const std::string& host, int port) {
     apr_sockaddr_t* address;
     CHECK_APR_SUCCESS(
         apr_sockaddr_info_get(
@@ -50,27 +53,28 @@ void Socket::connect(const std::string& host, int port)
     CHECK_APR_SUCCESS(apr_socket_connect(socket, address));
 }
 
-void Socket::close()
-{
+void Socket::close() {
     if (socket == 0) return;
     CHECK_APR_SUCCESS(apr_socket_close(socket));
     socket = 0;
 }
 
-ssize_t Socket::send(const char* data, size_t size)
+ssize_t Socket::send(const void* data, size_t size)
 {
     apr_size_t sent = size;
-    apr_status_t status = apr_socket_send(socket, data, &sent);
+    apr_status_t status =
+        apr_socket_send(socket, reinterpret_cast<const char*>(data), &sent);
     if (APR_STATUS_IS_TIMEUP(status)) return SOCKET_TIMEOUT;
     if (APR_STATUS_IS_EOF(status)) return SOCKET_EOF;
     CHECK_APR_SUCCESS(status);
     return sent;
 }
 
-ssize_t Socket::recv(char* data, size_t size)
+ssize_t Socket::recv(void* data, size_t size)
 {
     apr_size_t received = size;
-    apr_status_t status = apr_socket_recv(socket, data, &received);
+    apr_status_t status =
+        apr_socket_recv(socket, reinterpret_cast<char*>(data), &received);
     if (APR_STATUS_IS_TIMEUP(status)) return SOCKET_TIMEOUT;
     CHECK_APR_SUCCESS(status);
      return received;