QPID-4272: Large amounts of code are duplicated between the SSL and TCP transports

Refactor SslMux support simplifying it and remove need for separate
SslHandler and SslIo code.

Refactored SSL client code to use the same connect and broker SSL to use the same
accept sequences as the TCP code. This also solves QPID-3565: IPv6 support for SSL
transport on Unix C++ client/broker

Remove now unneeded ssl files.

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

17 files changed, 162 insertions, 1063 deletions

diff --git a/cpp/src/qpid/client/SslConnector.cpp b/cpp/src/qpid/client/SslConnector.cpp
index c49deaa279..3a146592e6 100644
--- a/cpp/src/qpid/client/SslConnector.cpp
+++ b/cpp/src/qpid/client/SslConnector.cpp
@@ -30,8 +30,9 @@
 #include "qpid/framing/AMQFrame.h"
 #include "qpid/framing/InitiationHandler.h"
 #include "qpid/sys/ssl/util.h"
-#include "qpid/sys/ssl/SslIo.h"
+#include "qpid/sys/AsynchIO.h"
 #include "qpid/sys/ssl/SslSocket.h"
+#include "qpid/sys/SocketAddress.h"
 #include "qpid/sys/Dispatcher.h"
 #include "qpid/sys/Poller.h"
 #include "qpid/sys/SecuritySettings.h"
@@ -72,7 +73,8 @@ class SslConnector : public Connector
 
     sys::ssl::SslSocket socket;
 
-    sys::ssl::SslIO* aio;
+    sys::AsynchConnector* connector;
+    sys::AsynchIO* aio;
     std::string identifier;
     Poller::shared_ptr poller;
     SecuritySettings securitySettings;
@@ -86,6 +88,8 @@ class SslConnector : public Connector
     void disconnected(AsynchIO&);
 
     void connect(const std::string& host, const std::string& port);
+    void connected(const sys::Socket&);
+    void connectFailed(const std::string& msg);
     void close();
     void send(framing::AMQFrame& frame);
     void abort() {} // TODO: Need to fix for heartbeat timeouts to work
@@ -164,24 +168,28 @@ SslConnector::~SslConnector() {
     close();
 }
 
-void SslConnector::connect(const std::string& host, const std::string& port){
+void SslConnector::connect(const std::string& host, const std::string& port) {
     Mutex::ScopedLock l(lock);
     assert(closed);
-    try {
-        socket.connect(SocketAddress(host, port));
-    } catch (const std::exception& e) {
-        socket.close();
-        throw TransportFailure(e.what());
-    }
-
+    connector = AsynchConnector::create(
+        socket,
+        host, port,
+        boost::bind(&SslConnector::connected, this, _1),
+        boost::bind(&SslConnector::connectFailed, this, _3));
     closed = false;
-    aio = new SslIO(socket,
-                       boost::bind(&SslConnector::readbuff, this, _1, _2),
-                       boost::bind(&SslConnector::eof, this, _1),
-                       boost::bind(&SslConnector::disconnected, this, _1),
-                       boost::bind(&SslConnector::socketClosed, this, _1, _2),
-                       0, // nobuffs
-                       boost::bind(&SslConnector::writebuff, this, _1));
+
+    connector->start(poller);
+}
+
+void SslConnector::connected(const Socket&) {
+    connector = 0;
+    aio = AsynchIO::create(socket,
+                           boost::bind(&SslConnector::readbuff, this, _1, _2),
+                           boost::bind(&SslConnector::eof, this, _1),
+                           boost::bind(&SslConnector::disconnected, this, _1),
+                           boost::bind(&SslConnector::socketClosed, this, _1, _2),
+                           0, // nobuffs
+                           boost::bind(&SslConnector::writebuff, this, _1));
 
     aio->createBuffers(maxFrameSize);
     identifier = str(format("[%1%]") % socket.getFullAddress());
@@ -190,6 +198,16 @@ void SslConnector::connect(const std::string& host, const std::string& port){
     aio->start(poller);
 }
 
+void SslConnector::connectFailed(const std::string& msg) {
+    connector = 0;
+    QPID_LOG(warning, "Connect failed: " << msg);
+    socket.close();
+    if (!closed)
+        closed = true;
+    if (shutdownHandler)
+        shutdownHandler->shutdown();
+}
+
 void SslConnector::close() {
     Mutex::ScopedLock l(lock);
     if (!closed) {
@@ -265,7 +283,7 @@ void SslConnector::writebuff(AsynchIO& /*aio*/)
         return;
     }
 
-    SslIO::BufferBase* buffer = aio->getQueuedBuffer();
+    AsynchIOBufferBase* buffer = aio->getQueuedBuffer();
     if (buffer) {
 
         size_t encoded = encode(buffer->bytes, buffer->byteCount);
@@ -343,7 +361,7 @@ size_t SslConnector::decode(const char* buffer, size_t size)
 }
 
 void SslConnector::writeDataBlock(const AMQDataBlock& data) {
-    SslIO::BufferBase* buff = aio->getQueuedBuffer();
+    AsynchIOBufferBase* buff = aio->getQueuedBuffer();
     assert(buff);
     framing::Buffer out(buff->bytes, buff->byteCount);
     data.encode(out);
diff --git a/cpp/src/qpid/sys/Socket.h b/cpp/src/qpid/sys/Socket.h
index ab59850969..38183bd5fd 100644
--- a/cpp/src/qpid/sys/Socket.h
+++ b/cpp/src/qpid/sys/Socket.h
@@ -46,6 +46,7 @@ public:
     virtual void setTcpNoDelay() const = 0;
 
     virtual void connect(const SocketAddress&) const = 0;
+    virtual void finishConnect(const SocketAddress&) const = 0;
 
     virtual void close() const = 0;
 
diff --git a/cpp/src/qpid/sys/SslPlugin.cpp b/cpp/src/qpid/sys/SslPlugin.cpp
index 77cda40056..3b56f9788b 100644
--- a/cpp/src/qpid/sys/SslPlugin.cpp
+++ b/cpp/src/qpid/sys/SslPlugin.cpp
@@ -22,20 +22,17 @@
 #include "qpid/sys/ProtocolFactory.h"
 
 #include "qpid/Plugin.h"
-#include "qpid/sys/ssl/check.h"
-#include "qpid/sys/ssl/util.h"
-#include "qpid/sys/ssl/SslHandler.h"
+#include "qpid/broker/Broker.h"
+#include "qpid/log/Statement.h"
 #include "qpid/sys/AsynchIOHandler.h"
 #include "qpid/sys/AsynchIO.h"
-#include "qpid/sys/ssl/SslIo.h"
+#include "qpid/sys/ssl/util.h"
 #include "qpid/sys/ssl/SslSocket.h"
 #include "qpid/sys/SocketAddress.h"
-#include "qpid/broker/Broker.h"
-#include "qpid/log/Statement.h"
+#include "qpid/sys/Poller.h"
 
 #include <boost/bind.hpp>
-#include <memory>
-
+#include <boost/ptr_container/ptr_vector.hpp>
 
 namespace qpid {
 namespace sys {
@@ -65,38 +62,33 @@ struct SslServerOptions : ssl::SslOptions
     }
 };
 
-template <class T>
-class SslProtocolFactoryTmpl : public ProtocolFactory {
-  private:
-
+class SslProtocolFactory : public ProtocolFactory {
+    boost::ptr_vector<Socket> listeners;
+    boost::ptr_vector<AsynchAcceptor> acceptors;
     Timer& brokerTimer;
     uint32_t maxNegotiateTime;
+    uint16_t listeningPort;
     const bool tcpNoDelay;
-    T listener;
-    const uint16_t listeningPort;
-    std::auto_ptr<SslAcceptor> acceptor;
     bool nodict;
 
   public:
-    SslProtocolFactoryTmpl(const std::string& host, const std::string& port,
+    SslProtocolFactory(const std::string& host, const std::string& port,
                            const SslServerOptions&,
                            int backlog, bool nodelay,
                            Timer& timer, uint32_t maxTime);
     void accept(Poller::shared_ptr, ConnectionCodec::Factory*);
     void connect(Poller::shared_ptr, const std::string& host, const std::string& port,
                  ConnectionCodec::Factory*,
-                 boost::function2<void, int, std::string> failed);
+                 ConnectFailedCallback);
 
     uint16_t getPort() const;
 
   private:
     void established(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*,
                      bool isClient);
+    void connectFailed(const Socket&, int, const std::string&, ConnectFailedCallback);
 };
 
-typedef SslProtocolFactoryTmpl<SslSocket> SslProtocolFactory;
-typedef SslProtocolFactoryTmpl<SslMuxSocket> SslMuxProtocolFactory;
-
 
 // Static instance to initialise plugin
 static struct SslPlugin : public Plugin {
@@ -125,7 +117,7 @@ static struct SslPlugin : public Plugin {
             }
         }
     }
-    
+
     void initialize(Target& target) {
         QPID_LOG(trace, "Initialising SSL plugin");
         broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
@@ -140,12 +132,7 @@ static struct SslPlugin : public Plugin {
 
                     const broker::Broker::Options& opts = broker->getOptions();
 
-                    ProtocolFactory::shared_ptr protocol(options.multiplex ?
-                        static_cast<ProtocolFactory*>(new SslMuxProtocolFactory("", boost::lexical_cast<std::string>(options.port),
-                                                                                options,
-                                                                                opts.connectionBacklog,
-                                                                                opts.tcpNoDelay,
-                                                                                broker->getTimer(), opts.maxNegotiateTime)) :
+                    ProtocolFactory::shared_ptr protocol(
                         static_cast<ProtocolFactory*>(new SslProtocolFactory("", boost::lexical_cast<std::string>(options.port),
                                                                              options,
                                                                              opts.connectionBacklog,
@@ -153,7 +140,7 @@ static struct SslPlugin : public Plugin {
                                                                              broker->getTimer(), opts.maxNegotiateTime)));
                     QPID_LOG(notice, "Listening for " <<
                                      (options.multiplex ? "SSL or TCP" : "SSL") <<
-                                     " connections on TCP port " <<
+                                     " connections on TCP/TCP6 port " <<
                                      protocol->getPort());
                     broker->registerProtocolFactory("ssl", protocol);
                 } catch (const std::exception& e) {
@@ -164,23 +151,48 @@ static struct SslPlugin : public Plugin {
     }
 } sslPlugin;
 
-template <class T>
-SslProtocolFactoryTmpl<T>::SslProtocolFactoryTmpl(const std::string& host, const std::string& port,
+SslProtocolFactory::SslProtocolFactory(const std::string& host, const std::string& port,
                                                   const SslServerOptions& options,
                                                   int backlog, bool nodelay,
                                                   Timer& timer, uint32_t maxTime) :
     brokerTimer(timer),
     maxNegotiateTime(maxTime),
     tcpNoDelay(nodelay),
-    listener(options.certName, options.clientAuth),
-    listeningPort(listener.listen(SocketAddress(host, port), backlog)),
     nodict(options.nodict)
-{}
+{
+    SocketAddress sa(host, port);
+
+    // We must have at least one resolved address
+    QPID_LOG(info, "Listening to: " << sa.asString())
+    Socket* s = options.multiplex ?
+        new SslMuxSocket(options.certName, options.clientAuth) :
+        new SslSocket(options.certName, options.clientAuth);
+    uint16_t lport = s->listen(sa, backlog);
+    QPID_LOG(debug, "Listened to: " << lport);
+    listeners.push_back(s);
+
+    listeningPort = lport;
+
+    // Try any other resolved addresses
+    while (sa.nextAddress()) {
+        // Hack to ensure that all listening connections are on the same port
+        sa.setAddrInfoPort(listeningPort);
+        QPID_LOG(info, "Listening to: " << sa.asString())
+        Socket* s = options.multiplex ?
+            new SslMuxSocket(options.certName, options.clientAuth) :
+            new SslSocket(options.certName, options.clientAuth);
+        uint16_t lport = s->listen(sa, backlog);
+        QPID_LOG(debug, "Listened to: " << lport);
+        listeners.push_back(s);
+    }
+
+}
 
-void SslEstablished(Poller::shared_ptr poller, const qpid::sys::SslSocket& s,
-                    ConnectionCodec::Factory* f, bool isClient,
-                    Timer& timer, uint32_t maxTime, bool tcpNoDelay, bool nodict) {
-    qpid::sys::ssl::SslHandler* async = new qpid::sys::ssl::SslHandler(s.getFullAddress(), f, nodict);
+
+void SslProtocolFactory::established(Poller::shared_ptr poller, const Socket& s,
+                                     ConnectionCodec::Factory* f, bool isClient) {
+
+    AsynchIOHandler* async = new AsynchIOHandler(s.getFullAddress(), f, nodict);
 
     if (tcpNoDelay) {
         s.setTcpNoDelay();
@@ -191,76 +203,43 @@ void SslEstablished(Poller::shared_ptr poller, const qpid::sys::SslSocket& s,
         async->setClient();
     }
 
-    qpid::sys::ssl::SslIO* aio = new qpid::sys::ssl::SslIO(s,
-                                 boost::bind(&qpid::sys::ssl::SslHandler::readbuff, async, _1, _2),
-                                 boost::bind(&qpid::sys::ssl::SslHandler::eof, async, _1),
-                                 boost::bind(&qpid::sys::ssl::SslHandler::disconnect, async, _1),
-                                 boost::bind(&qpid::sys::ssl::SslHandler::closedSocket, async, _1, _2),
-                                 boost::bind(&qpid::sys::ssl::SslHandler::nobuffs, async, _1),
-                                 boost::bind(&qpid::sys::ssl::SslHandler::idle, async, _1));
+    AsynchIO* aio = AsynchIO::create(
+        s,
+        boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
+        boost::bind(&AsynchIOHandler::eof, async, _1),
+        boost::bind(&AsynchIOHandler::disconnect, async, _1),
+        boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
+        boost::bind(&AsynchIOHandler::nobuffs, async, _1),
+        boost::bind(&AsynchIOHandler::idle, async, _1));
 
-    async->init(aio,timer, maxTime);
+    async->init(aio, brokerTimer, maxNegotiateTime);
     aio->start(poller);
 }
 
-template <>
-void SslProtocolFactory::established(Poller::shared_ptr poller, const Socket& s,
-                                     ConnectionCodec::Factory* f, bool isClient) {
-    const SslSocket *sslSock = dynamic_cast<const SslSocket*>(&s);
-
-    SslEstablished(poller, *sslSock, f, isClient, brokerTimer, maxNegotiateTime, tcpNoDelay, nodict);
-}
-
-template <class T>
-uint16_t SslProtocolFactoryTmpl<T>::getPort() const {
+uint16_t SslProtocolFactory::getPort() const {
     return listeningPort; // Immutable no need for lock.
 }
 
-template <class T>
-void SslProtocolFactoryTmpl<T>::accept(Poller::shared_ptr poller,
-                                       ConnectionCodec::Factory* fact) {
-    acceptor.reset(
-        new SslAcceptor(listener,
-                        boost::bind(&SslProtocolFactoryTmpl<T>::established,
-                                    this, poller, _1, fact, false)));
-    acceptor->start(poller);
-}
-
-template <>
-void SslMuxProtocolFactory::established(Poller::shared_ptr poller, const Socket& s,
-                                        ConnectionCodec::Factory* f, bool isClient) {
-    const SslSocket *sslSock = dynamic_cast<const SslSocket*>(&s);
-
-    if (sslSock) {
-        SslEstablished(poller, *sslSock, f, isClient, brokerTimer, maxNegotiateTime, tcpNoDelay, nodict);
-        return;
+void SslProtocolFactory::accept(Poller::shared_ptr poller,
+                                ConnectionCodec::Factory* fact) {
+    for (unsigned i = 0; i<listeners.size(); ++i) {
+        acceptors.push_back(
+            AsynchAcceptor::create(listeners[i],
+                            boost::bind(&SslProtocolFactory::established, this, poller, _1, fact, false)));
+        acceptors[i].start(poller);
     }
+}
 
-    AsynchIOHandler* async = new AsynchIOHandler(s.getFullAddress(), f, false);
-
-    if (tcpNoDelay) {
-        s.setTcpNoDelay();
-        QPID_LOG(info, "Set TCP_NODELAY on connection to " << s.getPeerAddress());
-    }
-
-    if (isClient) {
-        async->setClient();
-    }
-    AsynchIO* aio = AsynchIO::create
-      (s,
-       boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
-       boost::bind(&AsynchIOHandler::eof, async, _1),
-       boost::bind(&AsynchIOHandler::disconnect, async, _1),
-       boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
-       boost::bind(&AsynchIOHandler::nobuffs, async, _1),
-       boost::bind(&AsynchIOHandler::idle, async, _1));
-
-    async->init(aio, brokerTimer, maxNegotiateTime);
-    aio->start(poller);
+void SslProtocolFactory::connectFailed(
+    const Socket& s, int ec, const std::string& emsg,
+    ConnectFailedCallback failedCb)
+{
+    failedCb(ec, emsg);
+    s.close();
+    delete &s;
 }
 
-template <class T>
-void SslProtocolFactoryTmpl<T>::connect(
+void SslProtocolFactory::connect(
     Poller::shared_ptr poller,
     const std::string& host, const std::string& port,
     ConnectionCodec::Factory* fact,
@@ -272,10 +251,23 @@ void SslProtocolFactoryTmpl<T>::connect(
     // shutdown.  The allocated SslConnector frees itself when it
     // is no longer needed.
 
-    qpid::sys::ssl::SslSocket* socket = new qpid::sys::ssl::SslSocket();
-    new SslConnector(*socket, poller, host, port,
-                     boost::bind(&SslProtocolFactoryTmpl<T>::established, this, poller, _1, fact, true),
-                     failed);
+    Socket* socket = new qpid::sys::ssl::SslSocket();
+    try {
+    AsynchConnector* c = AsynchConnector::create(
+        *socket,
+        host,
+        port,
+        boost::bind(&SslProtocolFactory::established,
+                    this, poller, _1, fact, true),
+        boost::bind(&SslProtocolFactory::connectFailed,
+                    this, _1, _2, _3, failed));
+    c->start(poller);
+    } catch (std::exception&) {
+        // TODO: Design question - should we do the error callback and also throw?
+        int errCode = socket->getError();
+        connectFailed(*socket, errCode, strError(errCode), failed);
+        throw;
+    }
 }
 
 }} // namespace qpid::sys
diff --git a/cpp/src/qpid/sys/TCPIOPlugin.cpp b/cpp/src/qpid/sys/TCPIOPlugin.cpp
index 56f0158d37..69f48228ad 100644
--- a/cpp/src/qpid/sys/TCPIOPlugin.cpp
+++ b/cpp/src/qpid/sys/TCPIOPlugin.cpp
@@ -20,15 +20,15 @@
  */
 
 #include "qpid/sys/ProtocolFactory.h"
-#include "qpid/sys/AsynchIOHandler.h"
-#include "qpid/sys/AsynchIO.h"
 
 #include "qpid/Plugin.h"
+#include "qpid/broker/Broker.h"
+#include "qpid/log/Statement.h"
+#include "qpid/sys/AsynchIOHandler.h"
+#include "qpid/sys/AsynchIO.h"
 #include "qpid/sys/Socket.h"
 #include "qpid/sys/SocketAddress.h"
 #include "qpid/sys/Poller.h"
-#include "qpid/broker/Broker.h"
-#include "qpid/log/Statement.h"
 
 #include <boost/bind.hpp>
 #include <boost/ptr_container/ptr_vector.hpp>
diff --git a/cpp/src/qpid/sys/posix/AsynchIO.cpp b/cpp/src/qpid/sys/posix/AsynchIO.cpp
index a5a6937475..7eb25fd861 100644
--- a/cpp/src/qpid/sys/posix/AsynchIO.cpp
+++ b/cpp/src/qpid/sys/posix/AsynchIO.cpp
@@ -196,6 +196,13 @@ void AsynchConnector::connComplete(DispatchHandle& h)
     int errCode = socket.getError();
     if (errCode == 0) {
         h.stopWatch();
+        try {
+            socket.finishConnect(sa);
+        } catch (const std::exception& e) {
+            failCallback(socket, 0, e.what());
+            DispatchHandle::doDelete();
+            return;
+        }
         connCallback(socket);
     } else {
         // Retry while we cause an immediate exception
diff --git a/cpp/src/qpid/sys/posix/BSDSocket.cpp b/cpp/src/qpid/sys/posix/BSDSocket.cpp
index 905696cb8b..4fe85b93fb 100644
--- a/cpp/src/qpid/sys/posix/BSDSocket.cpp
+++ b/cpp/src/qpid/sys/posix/BSDSocket.cpp
@@ -176,6 +176,10 @@ void BSDSocket::connect(const SocketAddress& addr) const
     }
 }
 
+void BSDSocket::finishConnect(const SocketAddress&) const
+{
+}
+
 void
 BSDSocket::close() const
 {
diff --git a/cpp/src/qpid/sys/posix/BSDSocket.h b/cpp/src/qpid/sys/posix/BSDSocket.h
index 2ddc69e2f8..862d36c1b9 100644
--- a/cpp/src/qpid/sys/posix/BSDSocket.h
+++ b/cpp/src/qpid/sys/posix/BSDSocket.h
@@ -55,6 +55,7 @@ public:
     QPID_COMMON_EXTERN virtual void setTcpNoDelay() const;
 
     QPID_COMMON_EXTERN virtual void connect(const SocketAddress&) const;
+    QPID_COMMON_EXTERN virtual void finishConnect(const SocketAddress&) const;
 
     QPID_COMMON_EXTERN virtual void close() const;
 
diff --git a/cpp/src/qpid/sys/ssl/SslHandler.cpp b/cpp/src/qpid/sys/ssl/SslHandler.cpp
deleted file mode 100644
index f6605da953..0000000000
--- a/cpp/src/qpid/sys/ssl/SslHandler.cpp
+++ /dev/null
@@ -1,220 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-#include "qpid/sys/ssl/SslHandler.h"
-#include "qpid/sys/ssl/SslIo.h"
-#include "qpid/sys/ssl/SslSocket.h"
-#include "qpid/sys/Timer.h"
-#include "qpid/framing/AMQP_HighestVersion.h"
-#include "qpid/framing/ProtocolInitiation.h"
-#include "qpid/log/Statement.h"
-
-#include <boost/bind.hpp>
-
-namespace qpid {
-namespace sys {
-namespace ssl {
-
-
-struct ProtocolTimeoutTask : public sys::TimerTask {
-    SslHandler& handler;
-    std::string id;
-
-    ProtocolTimeoutTask(const std::string& i, const Duration& timeout, SslHandler& h) :
-    TimerTask(timeout, "ProtocolTimeout"),
-        handler(h),
-        id(i)
-    {}
-
-    void fire() {
-        // If this fires it means that we didn't negotiate the connection in the timeout period
-        // Schedule closing the connection for the io thread
-        QPID_LOG(error, "Connection " << id << " No protocol received closing");
-        handler.abort();
-    }
-};
-
-SslHandler::SslHandler(std::string id, ConnectionCodec::Factory* f, bool _nodict) :
-    identifier(id),
-    aio(0),
-    factory(f),
-    codec(0),
-    readError(false),
-    isClient(false),
-    nodict(_nodict)
-{}
-
-SslHandler::~SslHandler() {
-    if (codec)
-        codec->closed();
-    if (timeoutTimerTask)
-        timeoutTimerTask->cancel();
-    delete codec;
-}
-
-void SslHandler::init(AsynchIO* a, Timer& timer, uint32_t maxTime) {
-    aio = a;
-
-    // Start timer for this connection
-    timeoutTimerTask = new ProtocolTimeoutTask(identifier, maxTime*TIME_MSEC, *this);
-    timer.add(timeoutTimerTask);
-
-    // Give connection some buffers to use
-    aio->createBuffers();
-}
-
-void SslHandler::write(const framing::ProtocolInitiation& data)
-{
-    QPID_LOG(debug, "SENT [" << identifier << "]: INIT(" << data << ")");
-    AsynchIOBufferBase* buff = aio->getQueuedBuffer();
-    assert(buff);
-    framing::Buffer out(buff->bytes, buff->byteCount);
-    data.encode(out);
-    buff->dataCount = data.encodedSize();
-    aio->queueWrite(buff);
-}
-
-void SslHandler::abort() {
-    // Don't disconnect if we're already disconnecting
-    if (!readError) {
-        aio->requestCallback(boost::bind(&SslHandler::eof, this, _1));
-    }
-}
-void SslHandler::activateOutput() {
-    aio->notifyPendingWrite();
-}
-
-void SslHandler::giveReadCredit(int32_t) {
-    // FIXME aconway 2008-12-05: not yet implemented.
-}
-
-// Input side
-void SslHandler::readbuff(AsynchIO& , AsynchIOBufferBase* buff) {
-    if (readError) {
-        return;
-    }
-    size_t decoded = 0;
-    if (codec) {                // Already initiated
-        try {
-            decoded = codec->decode(buff->bytes+buff->dataStart, buff->dataCount);
-        }catch(const std::exception& e){
-            QPID_LOG(error, e.what());
-            readError = true;
-            aio->queueWriteClose();
-        }
-    }else{
-        framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);
-        framing::ProtocolInitiation protocolInit;
-        if (protocolInit.decode(in)) {
-            // We've just got the protocol negotiation so we can cancel the timeout for that
-            timeoutTimerTask->cancel();
-
-            decoded = in.getPosition();
-            QPID_LOG(debug, "RECV [" << identifier << "]: INIT(" << protocolInit << ")");
-            try {
-                codec = factory->create(protocolInit.getVersion(), *this, identifier, getSecuritySettings(aio));
-                if (!codec) {
-                    //TODO: may still want to revise this...
-                    //send valid version header & close connection.
-                    write(framing::ProtocolInitiation(framing::highestProtocolVersion));
-                    readError = true;
-                    aio->queueWriteClose();
-                } else {
-                    //read any further data that may already have been sent
-                    decoded += codec->decode(buff->bytes+buff->dataStart+in.getPosition(), buff->dataCount-in.getPosition());
-                }
-            } catch (const std::exception& e) {
-                QPID_LOG(error, e.what());
-                readError = true;
-                aio->queueWriteClose();
-            }
-        }
-    }
-    // TODO: unreading needs to go away, and when we can cope
-    // with multiple sub-buffers in the general buffer scheme, it will
-    if (decoded != size_t(buff->dataCount)) {
-        // Adjust buffer for used bytes and then "unread them"
-        buff->dataStart += decoded;
-        buff->dataCount -= decoded;
-        aio->unread(buff);
-    } else {
-        // Give whole buffer back to aio subsystem
-        aio->queueReadBuffer(buff);
-    }
-}
-
-void SslHandler::eof(AsynchIO&) {
-    QPID_LOG(debug, "DISCONNECTED [" << identifier << "]");
-    if (codec) codec->closed();
-    aio->queueWriteClose();
-}
-
-void SslHandler::closedSocket(AsynchIO&, const Socket& s) {
-    // If we closed with data still to send log a warning
-    if (!aio->writeQueueEmpty()) {
-        QPID_LOG(warning, "CLOSING [" << identifier << "] unsent data (probably due to client disconnect)");
-    }
-    delete &s;
-    aio->queueForDeletion();
-    delete this;
-}
-
-void SslHandler::disconnect(AsynchIO& a) {
-    // treat the same as eof
-    eof(a);
-}
-
-// Notifications
-void SslHandler::nobuffs(AsynchIO&) {
-}
-
-void SslHandler::idle(AsynchIO&){
-    if (isClient && codec == 0) {
-        codec = factory->create(*this, identifier, getSecuritySettings(aio));
-        write(framing::ProtocolInitiation(codec->getVersion()));
-        // We've just sent the protocol negotiation so we can cancel the timeout for that
-        // This is not ideal, because we've not received anything yet, but heartbeats will
-        // be active soon
-        timeoutTimerTask->cancel();
-        return;
-    }
-    if (codec == 0) return;
-    if (!codec->canEncode()) {
-        return;
-    }
-    AsynchIOBufferBase* buff = aio->getQueuedBuffer();
-    if (buff) {
-        size_t encoded=codec->encode(buff->bytes, buff->byteCount);
-        buff->dataCount = encoded;
-        aio->queueWrite(buff);
-    }
-    if (codec->isClosed())
-        aio->queueWriteClose();
-}
-
-SecuritySettings SslHandler::getSecuritySettings(AsynchIO* aio)
-{
-    SecuritySettings settings = aio->getSecuritySettings();
-    settings.nodict = nodict;
-    return settings;
-}
-
-
-}}} // namespace qpid::sys::ssl
diff --git a/cpp/src/qpid/sys/ssl/SslHandler.h b/cpp/src/qpid/sys/ssl/SslHandler.h
deleted file mode 100644
index aedfea1888..0000000000
--- a/cpp/src/qpid/sys/ssl/SslHandler.h
+++ /dev/null
@@ -1,85 +0,0 @@
-#ifndef QPID_SYS_SSL_SSLHANDLER_H
-#define QPID_SYS_SSL_SSLHANDLER_H
-
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "qpid/sys/ConnectionCodec.h"
-#include "qpid/sys/OutputControl.h"
-#include "qpid/sys/SecuritySettings.h"
-
-#include <boost/intrusive_ptr.hpp>
-
-namespace qpid {
-
-namespace framing {
-    class ProtocolInitiation;
-}
-
-namespace sys {
-
-class AsynchIO;
-struct AsynchIOBufferBase;
-class Socket;
-class Timer;
-class TimerTask;
-
-namespace ssl {
-
-class SslHandler : public OutputControl {
-    std::string identifier;
-    AsynchIO* aio;
-    ConnectionCodec::Factory* factory;
-    ConnectionCodec* codec;
-    bool readError;
-    bool isClient;
-    bool nodict;
-    boost::intrusive_ptr<sys::TimerTask> timeoutTimerTask;
-
-    void write(const framing::ProtocolInitiation&);
-    qpid::sys::SecuritySettings getSecuritySettings(AsynchIO* aio);
-
-  public:
-    SslHandler(std::string id, ConnectionCodec::Factory* f, bool nodict);
-    ~SslHandler();
-    void init(AsynchIO* a, Timer& timer, uint32_t maxTime);
-
-    void setClient() { isClient = true; }
-
-    // Output side
-    void abort();
-    void activateOutput();
-    void giveReadCredit(int32_t);
-
-    // Input side
-    void readbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase* buff);
-    void eof(qpid::sys::AsynchIO&);
-    void disconnect(qpid::sys::AsynchIO& a);
-
-    // Notifications
-    void nobuffs(qpid::sys::AsynchIO&);
-    void idle(qpid::sys::AsynchIO&);
-    void closedSocket(qpid::sys::AsynchIO&, const qpid::sys::Socket& s);
-};
-
-}}} // namespace qpid::sys::ssl
-
-#endif  /*!QPID_SYS_SSL_SSLHANDLER_H*/
diff --git a/cpp/src/qpid/sys/ssl/SslIo.cpp b/cpp/src/qpid/sys/ssl/SslIo.cpp
deleted file mode 100644
index 92e51a2234..0000000000
--- a/cpp/src/qpid/sys/ssl/SslIo.cpp
+++ /dev/null
@@ -1,463 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "qpid/sys/ssl/SslIo.h"
-#include "qpid/sys/ssl/SslSocket.h"
-#include "qpid/sys/ssl/check.h"
-
-#include "qpid/sys/Time.h"
-#include "qpid/sys/posix/check.h"
-#include "qpid/log/Statement.h"
-
-// TODO The basic algorithm here is not really POSIX specific and with a bit more abstraction
-// could (should) be promoted to be platform portable
-#include <unistd.h>
-#include <sys/socket.h>
-#include <signal.h>
-#include <errno.h>
-#include <string.h>
-
-#include <boost/bind.hpp>
-
-namespace qpid {
-namespace sys {
-namespace ssl {
-
-namespace {
-
-/*
- * Make *process* not generate SIGPIPE when writing to closed
- * pipe/socket (necessary as default action is to terminate process)
- */
-void ignoreSigpipe() {
-    ::signal(SIGPIPE, SIG_IGN);
-}
-
-/*
- * We keep per thread state to avoid locking overhead. The assumption is that
- * on average all the connections are serviced by all the threads so the state
- * recorded in each thread is about the same. If this turns out not to be the
- * case we could rebalance the info occasionally.  
- */
-__thread int threadReadTotal = 0;
-__thread int threadReadCount = 0;
-__thread int threadWriteTotal = 0;
-__thread int threadWriteCount = 0;
-__thread int64_t threadMaxIoTimeNs = 2 * 1000000; // start at 2ms
-}
-
-/*
- * Asynch Acceptor
- */
-
-SslAcceptor::SslAcceptor(const Socket& s, Callback callback) :
-    acceptedCallback(callback),
-    handle(s, boost::bind(&SslAcceptor::readable, this, _1), 0, 0),
-    socket(s) {
-
-    s.setNonblocking();
-    ignoreSigpipe();
-}
-
-SslAcceptor::~SslAcceptor()
-{
-    handle.stopWatch();
-}
-
-void SslAcceptor::start(Poller::shared_ptr poller) {
-    handle.startWatch(poller);
-}
-
-/*
- * We keep on accepting as long as there is something to accept
- */
-void SslAcceptor::readable(DispatchHandle& h) {
-    Socket* s;
-    do {
-        errno = 0;
-        // TODO: Currently we ignore the peers address, perhaps we should
-        // log it or use it for connection acceptance.
-        try {
-            s = socket.accept();
-            if (s) {
-                acceptedCallback(*s);
-            } else {
-                break;
-            }
-        } catch (const std::exception& e) {
-            QPID_LOG(error, "Could not accept socket: " << e.what());
-        }
-    } while (true);
-
-    h.rewatch();
-}
-
-/*
- * Asynch Connector
- */
-
-SslConnector::SslConnector(const SslSocket& s,
-                                 Poller::shared_ptr poller,
-                                 std::string hostname,
-                                 std::string port,
-                                 ConnectedCallback connCb,
-                                 FailedCallback failCb) :
-    DispatchHandle(s,
-                   0,
-                   boost::bind(&SslConnector::connComplete, this, _1),
-                   boost::bind(&SslConnector::connComplete, this, _1)),
-    connCallback(connCb),
-    failCallback(failCb),
-    socket(s),
-    sa(hostname, port)
-{
-    //TODO: would be better for connect to be performed on a
-    //non-blocking socket, but that doesn't work at present so connect
-    //blocks until complete
-    try {
-        socket.connect(sa);
-        socket.setNonblocking();
-        startWatch(poller);
-    } catch(std::exception& e) {
-        failure(-1, std::string(e.what()));
-    }
-}
-
-void SslConnector::connComplete(DispatchHandle& h)
-{
-    int errCode = socket.getError();
-
-    h.stopWatch();
-    if (errCode == 0) {
-        connCallback(socket);
-        DispatchHandle::doDelete();
-    } else {
-        // TODO: This need to be fixed as strerror isn't thread safe
-        failure(errCode, std::string(::strerror(errCode)));
-    }
-}
-
-void SslConnector::failure(int errCode, std::string message)
-{
-    if (failCallback)
-        failCallback(errCode, message);
-
-    socket.close();
-    delete &socket;
-
-    DispatchHandle::doDelete();
-}
-
-/*
- * Asynch reader/writer
- */
-SslIO::SslIO(const SslSocket& s,
-                   ReadCallback rCb, EofCallback eofCb, DisconnectCallback disCb,
-                   ClosedCallback cCb, BuffersEmptyCallback eCb, IdleCallback iCb) :
-
-    DispatchHandle(s, 
-                   boost::bind(&SslIO::readable, this, _1),
-                   boost::bind(&SslIO::writeable, this, _1),
-                   boost::bind(&SslIO::disconnected, this, _1)),
-    readCallback(rCb),
-    eofCallback(eofCb),
-    disCallback(disCb),
-    closedCallback(cCb),
-    emptyCallback(eCb),
-    idleCallback(iCb),
-    socket(s),
-    queuedClose(false),
-    writePending(false) {
-
-    s.setNonblocking();
-}
-
-SslIO::~SslIO() {
-}
-
-void SslIO::queueForDeletion() {
-    DispatchHandle::doDelete();
-}
-
-void SslIO::start(Poller::shared_ptr poller) {
-    DispatchHandle::startWatch(poller);
-}
-
-void SslIO::createBuffers(uint32_t size) {
-    // Allocate all the buffer memory at once
-    bufferMemory.reset(new char[size*BufferCount]);
-
-    // Create the Buffer structs in a vector
-    // And push into the buffer queue
-    buffers.reserve(BufferCount);
-    for (uint32_t i = 0; i < BufferCount; i++) {
-        buffers.push_back(BufferBase(&bufferMemory[i*size], size));
-        queueReadBuffer(&buffers[i]);
-    }
-}
-
-void SslIO::queueReadBuffer(BufferBase* buff) {
-    assert(buff);
-    buff->dataStart = 0;
-    buff->dataCount = 0;
-    bufferQueue.push_back(buff);
-    DispatchHandle::rewatchRead();
-}
-
-void SslIO::unread(BufferBase* buff) {
-    assert(buff);
-    if (buff->dataStart != 0) {
-        memmove(buff->bytes, buff->bytes+buff->dataStart, buff->dataCount);
-        buff->dataStart = 0;
-    }
-    bufferQueue.push_front(buff);
-    DispatchHandle::rewatchRead();
-}
-
-void SslIO::queueWrite(BufferBase* buff) {
-    assert(buff);
-    // If we've already closed the socket then throw the write away
-    if (queuedClose) {
-        bufferQueue.push_front(buff);
-        return;
-    } else {
-        writeQueue.push_front(buff);
-    }
-    writePending = false;
-    DispatchHandle::rewatchWrite();
-}
-
-void SslIO::notifyPendingWrite() {
-    writePending = true;
-    DispatchHandle::rewatchWrite();
-}
-
-void SslIO::queueWriteClose() {
-    queuedClose = true;
-    DispatchHandle::rewatchWrite();
-}
-
-void SslIO::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(&SslIO::requestedCall, this, callback));
-}
-
-void SslIO::requestedCall(RequestCallback callback) {
-    assert(callback);
-    callback(*this);
-}
-
-/** Return a queued buffer if there are enough
- * to spare
- */
-SslIO::BufferBase* SslIO::getQueuedBuffer() {
-    // Always keep at least one buffer (it might have data that was "unread" in it)
-    if (bufferQueue.size()<=1)
-        return 0;
-    BufferBase* buff = bufferQueue.back();
-    assert(buff);
-    buff->dataStart = 0;
-    buff->dataCount = 0;
-    bufferQueue.pop_back();
-    return buff;
-}
-
-/*
- * We keep on reading as long as we have something to read and a buffer to put
- * it in
- */
-void SslIO::readable(DispatchHandle& h) {
-    AbsTime readStartTime = AbsTime::now();
-    do {
-        // (Try to) get a buffer
-        if (!bufferQueue.empty()) {
-            // Read into buffer
-            BufferBase* buff = bufferQueue.front();
-            assert(buff);
-            bufferQueue.pop_front();
-            errno = 0;
-            int readCount = buff->byteCount-buff->dataCount;
-            int rc = socket.read(buff->bytes + buff->dataCount, readCount);
-            if (rc > 0) {
-                buff->dataCount += rc;
-                threadReadTotal += rc;
-
-                readCallback(*this, buff);
-                if (rc != readCount) {
-                    // If we didn't fill the read buffer then time to stop reading
-                    break;
-                }
-
-                // Stop reading if we've overrun our timeslot
-                if (Duration(readStartTime, AbsTime::now()) > threadMaxIoTimeNs) {
-                    break;
-                }
-
-            } else {
-                // Put buffer back (at front so it doesn't interfere with unread buffers)
-                bufferQueue.push_front(buff);
-                assert(buff);
-
-                // Eof or other side has gone away
-                if (rc == 0 || errno == ECONNRESET) {
-                    eofCallback(*this);
-                    h.unwatchRead();
-                    break;
-                } else if (errno == EAGAIN) {
-                    // We have just put a buffer back so we know
-                    // we can carry on watching for reads
-                    break;
-                } else {
-                    // Report error then just treat as a socket disconnect
-                    QPID_LOG(error, "Error reading socket: " << getErrorString(PR_GetError()));
-                    eofCallback(*this);
-                    h.unwatchRead();
-                    break;
-                }
-            }
-        } else {
-            // Something to read but no buffer
-            if (emptyCallback) {
-                emptyCallback(*this);
-            }
-            // If we still have no buffers we can't do anything more
-            if (bufferQueue.empty()) {
-                h.unwatchRead();
-                break;
-            }
-
-        }
-    } while (true);
-
-    ++threadReadCount;
-    return;
-}
-
-/*
- * We carry on writing whilst we have data to write and we can write
- */
-void SslIO::writeable(DispatchHandle& h) {
-    AbsTime writeStartTime = AbsTime::now();
-    do {
-        // See if we've got something to write
-        if (!writeQueue.empty()) {
-            // Write buffer
-            BufferBase* buff = writeQueue.back();
-            writeQueue.pop_back();
-            errno = 0;
-            assert(buff->dataStart+buff->dataCount <= buff->byteCount);
-            int rc = socket.write(buff->bytes+buff->dataStart, buff->dataCount);
-            if (rc >= 0) {
-                threadWriteTotal += rc;
-
-                // If we didn't write full buffer put rest back
-                if (rc != buff->dataCount) {
-                    buff->dataStart += rc;
-                    buff->dataCount -= rc;
-                    writeQueue.push_back(buff);
-                    break;
-                }
-
-                // Recycle the buffer
-                queueReadBuffer(buff);
-
-                // Stop writing if we've overrun our timeslot
-                if (Duration(writeStartTime, AbsTime::now()) > threadMaxIoTimeNs) {
-                    break;
-                }
-            } else {
-                // Put buffer back
-                writeQueue.push_back(buff);
-                if (errno == ECONNRESET || errno == EPIPE) {
-                    // Just stop watching for write here - we'll get a
-                    // disconnect callback soon enough
-                    h.unwatchWrite();
-                    break;
-                } else if (errno == EAGAIN) {
-                    // We have just put a buffer back so we know
-                    // we can carry on watching for writes
-                    break;
-                } else {
-                    QPID_LOG(error, "Error writing to socket: " << getErrorString(PR_GetError()));
-                    h.unwatchWrite();
-                    break;
-                }
-            }
-        } else {
-            // If we're waiting to close the socket then can do it now as there is nothing to write
-            if (queuedClose) {
-                close(h);
-                break;
-            }
-            // Fd is writable, but nothing to write
-            if (idleCallback) {
-                writePending = false;
-                idleCallback(*this);
-            }
-            // If we still have no buffers to write we can't do anything more
-            if (writeQueue.empty() && !writePending && !queuedClose) {
-                h.unwatchWrite();
-                // The following handles the case where writePending is
-                // set to true after the test above; in this case its
-                // possible that the unwatchWrite overwrites the
-                // desired rewatchWrite so we correct that here
-                if (writePending)
-                    h.rewatchWrite();
-                break;
-            }
-        }
-    } while (true);
-
-    ++threadWriteCount;
-    return;
-}
-
-void SslIO::disconnected(DispatchHandle& h) {
-    // If we've already queued close do it instead of disconnected callback
-    if (queuedClose) {
-        close(h);
-    } else if (disCallback) {
-        disCallback(*this);
-        h.unwatch();
-    }
-}
-
-/*
- * Close the socket and callback to say we've done it
- */
-void SslIO::close(DispatchHandle& h) {
-    h.stopWatch();
-    socket.close();
-    if (closedCallback) {
-        closedCallback(*this, socket);
-    }
-}
-
-SecuritySettings SslIO::getSecuritySettings() {
-    SecuritySettings settings;
-    settings.ssf = socket.getKeyLen();
-    settings.authid = socket.getClientAuthId();
-    return settings;
-}
-
-}}}
diff --git a/cpp/src/qpid/sys/ssl/SslIo.h b/cpp/src/qpid/sys/ssl/SslIo.h
deleted file mode 100644
index a72cd7c76c..0000000000
--- a/cpp/src/qpid/sys/ssl/SslIo.h
+++ /dev/null
@@ -1,159 +0,0 @@
-#ifndef _sys_ssl_SslIO
-#define _sys_ssl_SslIO
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include <qpid/sys/AsynchIO.h>
-#include "qpid/sys/DispatchHandle.h"
-#include "qpid/sys/SecuritySettings.h"
-#include "qpid/sys/SocketAddress.h"
-
-#include <boost/function.hpp>
-#include <boost/shared_array.hpp>
-#include <deque>
-
-namespace qpid {
-namespace sys {
-
-class Socket;
-
-namespace ssl {
-
-class SslSocket;
-
-/*
- * Asynchronous ssl acceptor: accepts connections then does a callback
- * with the accepted fd
- */
-class SslAcceptor {
-public:
-    typedef boost::function1<void, const Socket&> Callback;
-
-private:
-    Callback acceptedCallback;
-    qpid::sys::DispatchHandle handle;
-    const Socket& socket;
-
-public:
-    SslAcceptor(const Socket& s, Callback callback);
-    ~SslAcceptor();
-    void start(qpid::sys::Poller::shared_ptr poller);
-
-private:
-    void readable(qpid::sys::DispatchHandle& handle);
-};
-
-/*
- * Asynchronous ssl connector: starts the process of initiating a
- * connection and invokes a callback when completed or failed.
- */
-class SslConnector : private qpid::sys::DispatchHandle {
-public:
-    typedef boost::function1<void, const SslSocket&> ConnectedCallback;
-    typedef boost::function2<void, int, std::string> FailedCallback;
-
-private:
-    ConnectedCallback connCallback;
-    FailedCallback failCallback;
-    const SslSocket& socket;
-    SocketAddress sa;
-
-public:
-    SslConnector(const SslSocket& socket,
-                    Poller::shared_ptr poller,
-                    std::string hostname,
-                    std::string port,
-                    ConnectedCallback connCb,
-                    FailedCallback failCb = 0);
-
-private:
-    void connComplete(DispatchHandle& handle);
-    void failure(int, std::string);
-};
-
-/*
- * Asychronous reader/writer: 
- * Reader accepts buffers to read into; reads into the provided buffers
- * and then does a callback with the buffer and amount read. Optionally it can callback
- * when there is something to read but no buffer to read it into.
- * 
- * Writer accepts a buffer and queues it for writing; can also be given
- * a callback for when writing is "idle" (ie fd is writable, but nothing to write)
- * 
- * The class is implemented in terms of DispatchHandle to allow it to be deleted by deleting
- * the contained DispatchHandle
- */
-class SslIO : public AsynchIO, private qpid::sys::DispatchHandle {
-public:
-    SslIO(const SslSocket& s,
-          ReadCallback rCb, EofCallback eofCb, DisconnectCallback disCb,
-          ClosedCallback cCb = 0, BuffersEmptyCallback eCb = 0, IdleCallback iCb = 0);
-private:
-    ReadCallback readCallback;
-    EofCallback eofCallback;
-    DisconnectCallback disCallback;
-    ClosedCallback closedCallback;
-    BuffersEmptyCallback emptyCallback;
-    IdleCallback idleCallback;
-    const SslSocket& socket;
-    std::deque<BufferBase*> bufferQueue;
-    std::deque<BufferBase*> writeQueue;
-    std::vector<BufferBase> buffers;
-    boost::shared_array<char> bufferMemory;
-    bool queuedClose;
-    /**
-     * This flag is used to detect and handle concurrency between
-     * calls to notifyPendingWrite() (which can be made from any thread) and
-     * the execution of the writeable() method (which is always on the
-     * thread processing this handle.
-     */
-    volatile bool writePending;
-
-public:
-    void queueForDeletion();
-
-    void start(qpid::sys::Poller::shared_ptr poller);
-    void createBuffers(uint32_t size = MaxBufferSize);
-    void queueReadBuffer(BufferBase* buff);
-    void unread(BufferBase* buff);
-    void queueWrite(BufferBase* buff);
-    void notifyPendingWrite();
-    void queueWriteClose();
-    bool writeQueueEmpty() { return writeQueue.empty(); }
-    void startReading() {};
-    void stopReading() {};
-    void requestCallback(RequestCallback);
-    BufferBase* getQueuedBuffer();
-
-    qpid::sys::SecuritySettings getSecuritySettings();
-
-private:
-    ~SslIO();
-    void readable(qpid::sys::DispatchHandle& handle);
-    void writeable(qpid::sys::DispatchHandle& handle);
-    void disconnected(qpid::sys::DispatchHandle& handle);
-    void requestedCall(RequestCallback);
-    void close(qpid::sys::DispatchHandle& handle);
-};
-
-}}}
-
-#endif // _sys_ssl_SslIO
diff --git a/cpp/src/qpid/sys/ssl/SslSocket.cpp b/cpp/src/qpid/sys/ssl/SslSocket.cpp
index 22f9f63fff..a328e49c13 100644
--- a/cpp/src/qpid/sys/ssl/SslSocket.cpp
+++ b/cpp/src/qpid/sys/ssl/SslSocket.cpp
@@ -87,6 +87,7 @@ SslSocket::SslSocket(const std::string& certName, bool clientAuth) :
 {
     //configure prototype socket:
     prototype = SSL_ImportFD(0, PR_NewTCPSocket());
+
     if (clientAuth) {
         NSS_CHECK(SSL_OptionSet(prototype, SSL_REQUEST_CERTIFICATE, PR_TRUE));
         NSS_CHECK(SSL_OptionSet(prototype, SSL_REQUIRE_CERTIFICATE, PR_TRUE));
@@ -131,7 +132,10 @@ void SslSocket::setTcpNoDelay() const
 void SslSocket::connect(const SocketAddress& addr) const
 {
     BSDSocket::connect(addr);
+}
 
+void SslSocket::finishConnect(const SocketAddress& addr) const
+{
     nssSocket = SSL_ImportFD(0, PR_ImportTCPSocket(fd));
 
     void* arg;
@@ -167,9 +171,9 @@ void SslSocket::close() const
 int SslSocket::listen(const SocketAddress& sa, int backlog) const
 {
     //get certificate and key (is this the correct way?)
-    std::string certName( (certname == "") ? "localhost.localdomain" : certname);
-    CERTCertificate *cert = PK11_FindCertFromNickname(const_cast<char*>(certName.c_str()), 0);
-    if (!cert) throw Exception(QPID_MSG("Failed to load certificate '" << certName << "'"));
+    std::string cName( (certname == "") ? "localhost.localdomain" : certname);
+    CERTCertificate *cert = PK11_FindCertFromNickname(const_cast<char*>(cName.c_str()), 0);
+    if (!cert) throw Exception(QPID_MSG("Failed to load certificate '" << cName << "'"));
     SECKEYPrivateKey *key = PK11_FindKeyByAnyCert(cert, 0);
     if (!key) throw Exception(QPID_MSG("Failed to retrieve private key from certificate"));
     NSS_CHECK(SSL_ConfigSecureServer(prototype, cert, key, NSS_FindCertKEAType(cert)));
diff --git a/cpp/src/qpid/sys/ssl/SslSocket.h b/cpp/src/qpid/sys/ssl/SslSocket.h
index 1efbbe4a88..fc97059cfd 100644
--- a/cpp/src/qpid/sys/ssl/SslSocket.h
+++ b/cpp/src/qpid/sys/ssl/SslSocket.h
@@ -57,6 +57,7 @@ public:
     void setCertName(const std::string& certName);
 
     void connect(const SocketAddress&) const;
+    void finishConnect(const SocketAddress&) const;
 
     void close() const;
 
diff --git a/cpp/src/qpid/sys/windows/WinSocket.cpp b/cpp/src/qpid/sys/windows/WinSocket.cpp
index cb321d1ffb..b2d2d79c63 100644
--- a/cpp/src/qpid/sys/windows/WinSocket.cpp
+++ b/cpp/src/qpid/sys/windows/WinSocket.cpp
@@ -165,6 +165,11 @@ WinSocket::connect(const SocketAddress& addr) const
 }
 
 void
+WinSocket::finishConnect(const SocketAddress&) const
+{
+}
+
+void
 WinSocket::close() const
 {
     SOCKET& socket = handle->fd;
diff --git a/cpp/src/qpid/sys/windows/WinSocket.h b/cpp/src/qpid/sys/windows/WinSocket.h
index 3aad6e7178..bee6a58e7a 100644
--- a/cpp/src/qpid/sys/windows/WinSocket.h
+++ b/cpp/src/qpid/sys/windows/WinSocket.h
@@ -61,6 +61,7 @@ public:
     QPID_COMMON_EXTERN virtual void setTcpNoDelay() const;
 
     QPID_COMMON_EXTERN virtual void connect(const SocketAddress&) const;
+    QPID_COMMON_EXTERN virtual void finishConnect(const SocketAddress&) const;
 
     QPID_COMMON_EXTERN virtual void close() const;
 
diff --git a/cpp/src/ssl.cmake b/cpp/src/ssl.cmake
index d66f59cfff..8e9e270c00 100644
--- a/cpp/src/ssl.cmake
+++ b/cpp/src/ssl.cmake
@@ -69,8 +69,6 @@ if (BUILD_SSL)
          qpid/sys/ssl/util.cpp
          qpid/sys/ssl/SslSocket.h
          qpid/sys/ssl/SslSocket.cpp
-         qpid/sys/ssl/SslIo.h
-         qpid/sys/ssl/SslIo.cpp
         )
 
     add_library (sslcommon SHARED ${sslcommon_SOURCES})
@@ -87,8 +85,6 @@ if (BUILD_SSL)
 
     set (ssl_SOURCES
          qpid/sys/SslPlugin.cpp
-         qpid/sys/ssl/SslHandler.h
-         qpid/sys/ssl/SslHandler.cpp
         )
     add_library (ssl MODULE ${ssl_SOURCES})
     target_link_libraries (ssl qpidbroker sslcommon ${Boost_PROGRAM_OPTIONS_LIBRARY})
diff --git a/cpp/src/ssl.mk b/cpp/src/ssl.mk
index 4dba9bb61c..89e7ed8049 100644
--- a/cpp/src/ssl.mk
+++ b/cpp/src/ssl.mk
@@ -25,9 +25,7 @@ libsslcommon_la_SOURCES = \
   qpid/sys/ssl/util.h \
   qpid/sys/ssl/util.cpp \
   qpid/sys/ssl/SslSocket.h \
-  qpid/sys/ssl/SslSocket.cpp \
-  qpid/sys/ssl/SslIo.h \
-  qpid/sys/ssl/SslIo.cpp
+  qpid/sys/ssl/SslSocket.cpp
 
 SSLCOMMON_VERSION_INFO  = 2:0:0
 libsslcommon_la_LDFLAGS = -version-info $(SSLCOMMON_VERSION_INFO)
@@ -37,9 +35,7 @@ libsslcommon_la_CXXFLAGS=$(AM_CXXFLAGS) $(SSL_CFLAGS)
 lib_LTLIBRARIES +=  libsslcommon.la
 
 ssl_la_SOURCES = \
-  qpid/sys/SslPlugin.cpp \
-  qpid/sys/ssl/SslHandler.h \
-  qpid/sys/ssl/SslHandler.cpp
+  qpid/sys/SslPlugin.cpp
 
 ssl_la_LIBADD= libqpidbroker.la libsslcommon.la