Rationalised AsynchConnector/Acceptor/IO to all use the same code structure

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

3 files changed, 38 insertions, 66 deletions

diff --git a/cpp/src/qpid/sys/AsynchIO.h b/cpp/src/qpid/sys/AsynchIO.h
index 419770568a..2a41f0a7d1 100644
--- a/cpp/src/qpid/sys/AsynchIO.h
+++ b/cpp/src/qpid/sys/AsynchIO.h
@@ -36,18 +36,13 @@ class Poller;
  * Asynchronous acceptor: accepts connections then does a callback with the
  * accepted fd
  */
-class AsynchAcceptorPrivate;
 class AsynchAcceptor {
 public:
     typedef boost::function1<void, const Socket&> Callback;
 
-private:
-    AsynchAcceptorPrivate* impl;
-
-public:
-    QPID_COMMON_EXTERN AsynchAcceptor(const Socket& s, Callback callback);
-    QPID_COMMON_EXTERN ~AsynchAcceptor();
-    QPID_COMMON_EXTERN void start(boost::shared_ptr<Poller> poller);
+    QPID_COMMON_EXTERN static AsynchAcceptor* create(const Socket& s, Callback callback);
+    virtual ~AsynchAcceptor() {};
+    virtual void start(boost::shared_ptr<Poller> poller) = 0;
 };
 
 /*
diff --git a/cpp/src/qpid/sys/TCPIOPlugin.cpp b/cpp/src/qpid/sys/TCPIOPlugin.cpp
index d51893374a..39ae12c189 100644
--- a/cpp/src/qpid/sys/TCPIOPlugin.cpp
+++ b/cpp/src/qpid/sys/TCPIOPlugin.cpp
@@ -114,7 +114,7 @@ std::string AsynchIOProtocolFactory::getHost() const {
 void AsynchIOProtocolFactory::accept(Poller::shared_ptr poller,
                                      ConnectionCodec::Factory* fact) {
     acceptor.reset(
-        new AsynchAcceptor(listener,
+        AsynchAcceptor::create(listener,
                            boost::bind(&AsynchIOProtocolFactory::established, this, poller, _1, fact, false)));
     acceptor->start(poller);
 }
diff --git a/cpp/src/qpid/sys/posix/AsynchIO.cpp b/cpp/src/qpid/sys/posix/AsynchIO.cpp
index c042dcef01..67b5cf0534 100644
--- a/cpp/src/qpid/sys/posix/AsynchIO.cpp
+++ b/cpp/src/qpid/sys/posix/AsynchIO.cpp
@@ -73,11 +73,12 @@ __thread int64_t threadMaxReadTimeNs = 2 * 1000000; // start at 2ms
  */
 namespace qpid {
 namespace sys {
+namespace posix {
 
-class AsynchAcceptorPrivate {
+class AsynchAcceptor : public qpid::sys::AsynchAcceptor {
 public:
-    AsynchAcceptorPrivate(const Socket& s, AsynchAcceptor::Callback callback);
-    ~AsynchAcceptorPrivate();
+    AsynchAcceptor(const Socket& s, AsynchAcceptor::Callback callback);
+    ~AsynchAcceptor();
     void start(Poller::shared_ptr poller);
 
 private:
@@ -90,40 +91,27 @@ private:
 
 };
 
-}} // namespace qpid::sys
-
-AsynchAcceptor::AsynchAcceptor(const Socket& s, Callback callback) :
-  impl(new AsynchAcceptorPrivate(s, callback))
-{}
-
-AsynchAcceptor::~AsynchAcceptor()
-{ delete impl;}
-
-void AsynchAcceptor::start(Poller::shared_ptr poller) {
-    impl->start(poller);
-}
-
-AsynchAcceptorPrivate::AsynchAcceptorPrivate(const Socket& s,
-                                             AsynchAcceptor::Callback callback) :
+AsynchAcceptor::AsynchAcceptor(const Socket& s,
+                               AsynchAcceptor::Callback callback) :
     acceptedCallback(callback),
-    handle(s, boost::bind(&AsynchAcceptorPrivate::readable, this, _1), 0, 0),
+    handle(s, boost::bind(&AsynchAcceptor::readable, this, _1), 0, 0),
     socket(s) {
 
     s.setNonblocking();
 }
 
-AsynchAcceptorPrivate::~AsynchAcceptorPrivate() {
+AsynchAcceptor::~AsynchAcceptor() {
     handle.stopWatch();
 }
 
-void AsynchAcceptorPrivate::start(Poller::shared_ptr poller) {
+void AsynchAcceptor::start(Poller::shared_ptr poller) {
     handle.startWatch(poller);
 }
 
 /*
  * We keep on accepting as long as there is something to accept
  */
-void AsynchAcceptorPrivate::readable(DispatchHandle& h) {
+void AsynchAcceptor::readable(DispatchHandle& h) {
     Socket* s;
     do {
         errno = 0;
@@ -146,13 +134,6 @@ void AsynchAcceptorPrivate::readable(DispatchHandle& h) {
 }
 
 /*
- * Asynch Connector
- */
-namespace qpid {
-namespace sys {
-namespace posix {
-
-/*
  * POSIX version of AsynchIO TCP socket connector.
  *
  * The class is implemented in terms of DispatchHandle to allow it to be
@@ -228,32 +209,12 @@ void AsynchConnector::failure(int errCode, const std::string& message)
     DispatchHandle::doDelete();
 }
 
-} // namespace posix
-
-
-AsynchConnector* qpid::sys::AsynchConnector::create(const Socket& s,
-                                                    Poller::shared_ptr poller,
-                                                    std::string hostname,
-                                                    uint16_t port,
-                                                    ConnectedCallback connCb,
-                                                    FailedCallback failCb)
-{
-    return new qpid::sys::posix::AsynchConnector(s,
-                                                 poller,
-                                                 hostname,
-                                                 port,
-                                                 connCb,
-                                                 failCb);
-}
-
 /*
  * POSIX version of AsynchIO reader/writer
  *
  * The class is implemented in terms of DispatchHandle to allow it to be
  * deleted by deleting the contained DispatchHandle.
  */
-namespace posix {
-
 class AsynchIO : public qpid::sys::AsynchIO, private DispatchHandle {
 
 public:
@@ -629,15 +590,31 @@ void AsynchIO::close(DispatchHandle& h) {
 
 } // namespace posix
 
-AsynchIO* qpid::sys::AsynchIO::create(const Socket& s,
-                                      AsynchIO::ReadCallback rCb,
-                                      AsynchIO::EofCallback eofCb,
-                                      AsynchIO::DisconnectCallback disCb,
-                                      AsynchIO::ClosedCallback cCb,
-                                      AsynchIO::BuffersEmptyCallback eCb,
-                                      AsynchIO::IdleCallback iCb)
+AsynchAcceptor* AsynchAcceptor::create(const Socket& s, 
+                                       Callback callback)
+{
+    return new posix::AsynchAcceptor(s, callback);
+}
+
+AsynchConnector* AsynchConnector::create(const Socket& s,
+                                         Poller::shared_ptr poller,
+                                         std::string hostname,
+                                         uint16_t port,
+                                         ConnectedCallback connCb,
+                                         FailedCallback failCb)
+{
+    return new posix::AsynchConnector(s, poller, hostname, port, connCb, failCb);
+}
+
+AsynchIO* AsynchIO::create(const Socket& s,
+                           AsynchIO::ReadCallback rCb,
+                           AsynchIO::EofCallback eofCb,
+                           AsynchIO::DisconnectCallback disCb,
+                           AsynchIO::ClosedCallback cCb,
+                           AsynchIO::BuffersEmptyCallback eCb,
+                           AsynchIO::IdleCallback iCb)
 {
-    return new qpid::sys::posix::AsynchIO(s, rCb, eofCb, disCb, cCb, eCb, iCb);
+    return new posix::AsynchIO(s, rCb, eofCb, disCb, cCb, eCb, iCb);
 }
 
 }} // namespace qpid::sys