Cleaned up some of the socket code, cleaned up the pipe engines, added support for binary sockets, and cleaned up the asynch connect/accept code

16 files changed, 728 insertions, 715 deletions

diff --git a/include/extern.h b/include/extern.h
index 0b90c6e14..1772c965c 100644
--- a/include/extern.h
+++ b/include/extern.h
@@ -106,7 +106,16 @@ E bool restarting;
 E Anope::string quitmsg;
 E time_t start_time;
 
-E ConnectionSocket *UplinkSock;
+class UplinkSocket : public ConnectionSocket, public BufferedSocket
+{
+ public:
+	UplinkSocket();
+	~UplinkSocket();
+	bool Read(const Anope::string &);
+	void OnConnect();
+	void OnError(const Anope::string &);
+};
+E UplinkSocket *UplinkSock;
 E int CurrentUplink;
 
 E void save_databases();
diff --git a/include/sockets.h b/include/sockets.h
index eb86130b5..b8d6033ee 100644
--- a/include/sockets.h
+++ b/include/sockets.h
@@ -106,22 +106,17 @@ class SocketException : public CoreException
 	virtual ~SocketException() throw() { }
 };
 
-enum SocketType
-{
-	SOCKTYPE_BASE,
-	SOCKTYPE_BUFFERED,
-	SOCKTYPE_CONNECTION,
-	SOCKTYPE_CLIENT,
-	SOCKTYPE_LISTEN
-};
-
 enum SocketFlag
 {
 	SF_DEAD,
-	SF_WRITABLE
+	SF_WRITABLE,
+	SF_CONNECTING,
+	SF_CONNECTED,
+	SF_ACCEPTING,
+	SF_ACCEPTED
 };
 
-static const Anope::string SocketFlagStrings[] = { "SF_DEAD", "SF_WRITABLE", "" };
+static const Anope::string SocketFlagStrings[] = { "SF_DEAD", "SF_WRITABLE", "SF_CONNECTING", "SF_CONNECTED", "SF_ACCEPTING", "SF_ACCEPTED", "" };
 
 class Socket;
 class ClientSocket;
@@ -140,11 +135,12 @@ class CoreExport SocketIO
 	virtual int Recv(Socket *s, char *buf, size_t sz);
 
 	/** Write something to the socket
- 	 * @param s The socket
-	 * @param buf What to write
-	 * @return Number of bytes written
+	 * @param s The socket
+	 * @param buf The data to write
+	 * @param size The length of the data
 	 */
-	virtual int Send(Socket *s, const Anope::string &buf);
+	virtual int Send(Socket *s, const char *buf, size_t sz);
+	int Send(Socket *s, const Anope::string &buf);
 
 	/** Accept a connection from a socket
 	 * @param s The socket
@@ -152,11 +148,11 @@ class CoreExport SocketIO
 	 */
 	virtual ClientSocket *Accept(ListenSocket *s);
 
-	/** Check if a connection has been accepted
-	 * @param s The client socket
-	 * @return -1 on error, 0 to wait, 1 on success
+	/** Finished accepting a connection from a socket
+	 * @param s The socket
+	 * @return SF_ACCEPTED if accepted, SF_ACCEPTING if still in process, SF_DEAD on error
 	 */
-	virtual int Accepted(ClientSocket *cs);
+	virtual SocketFlag FinishAccept(ClientSocket *cs);
 
 	/** Bind a socket
 	 * @param s The socket
@@ -172,18 +168,18 @@ class CoreExport SocketIO
 	 */
 	virtual void Connect(ConnectionSocket *s, const Anope::string &target, int port);
 
-	/** Check if this socket is connected
+	/** Called to potentially finish a pending connection
 	 * @param s The socket
-	 * @return -1 for error, 0 for wait, 1 for connected
+	 * @return SF_CONNECTED on success, SF_CONNECTING if still pending, and SF_DEAD on error.
 	 */
-	virtual int Connected(ConnectionSocket *s);
+	virtual SocketFlag FinishConnect(ConnectionSocket *s);
 
 	/** Called when the socket is destructing
 	 */
 	virtual void Destroy() { }
 };
 
-class CoreExport Socket : public Flags<SocketFlag, 2>
+class CoreExport Socket : public Flags<SocketFlag>
 {
  protected:
 	/* Socket FD */
@@ -198,19 +194,16 @@ class CoreExport Socket : public Flags<SocketFlag, 2>
 	/* I/O functions used for this socket */
  	SocketIO *IO;
 
-	/* Type this socket is */
-	SocketType Type;
-
-	/** Empty constructor, used for things such as the pipe socket
+	/** Empty constructor, should not be called.
 	 */
 	Socket();
 
 	/** Default constructor
-	 * @param sock The socket to use, 0 if we need to create our own
+	 * @param sock The socket to use, -1 if we need to create our own
 	 * @param ipv6 true if using ipv6
 	 * @param type The socket type, defaults to SOCK_STREAM
 	 */
- 	Socket(int sock, bool ipv6, int type = SOCK_STREAM);
+ 	Socket(int sock, bool ipv6 = false, int type = SOCK_STREAM);
 
 	/** Default destructor
 	 */
@@ -242,6 +235,11 @@ class CoreExport Socket : public Flags<SocketFlag, 2>
 	 */
 	void Bind(const Anope::string &ip, int port = 0);
 
+	/** Called when there either is a read or write event.
+	 * @return true to continue to call ProcessRead/ProcessWrite, false to not continue
+	 */
+	virtual bool Process();
+
 	/** Called when there is something to be received for this socket
 	 * @return true on success, false to drop this socket
 	 */
@@ -258,7 +256,7 @@ class CoreExport Socket : public Flags<SocketFlag, 2>
 	virtual void ProcessError();
 };
 
-class CoreExport BufferedSocket : public Socket
+class CoreExport BufferedSocket : public virtual Socket
 {
  protected:
 	/* Things to be written to the socket */
@@ -269,16 +267,9 @@ class CoreExport BufferedSocket : public Socket
 	int RecvLen;
 
  public:
-	/** Blank constructor
-	 */
-	BufferedSocket();
-
 	/** Constructor
-	 * @param fd FD to use
-	 * @param ipv6 true for ipv6
-	 * @param type socket type, defaults to SOCK_STREAM
 	 */
-	BufferedSocket(int fd, bool ipv6, int type = SOCK_STREAM);
+	BufferedSocket();
 
 	/** Default destructor
 	 */
@@ -317,6 +308,52 @@ class CoreExport BufferedSocket : public Socket
 	int WriteBufferLen() const;
 };
 
+class CoreExport BinarySocket : public virtual Socket
+{
+	struct DataBlock
+	{
+		char *buf;
+		size_t len;
+
+		DataBlock(const char *b, size_t l);
+		~DataBlock();
+	};
+
+	std::deque<DataBlock *> WriteBuffer;
+
+ public:
+	/** Constructor
+	 */
+	BinarySocket();
+
+	/** Default destructor
+	 */
+	virtual ~BinarySocket();
+
+	/** Called when there is something to be received for this socket
+	 * @return true on success, false to drop this socket
+	 */
+	bool ProcessRead();
+
+	/** Called when the socket is ready to be written to
+	 * @return true on success, false to drop this socket
+	 */
+	bool ProcessWrite();
+
+	/** Write data to the socket
+	 * @param buffer The data to write
+	 * @param l The length of the data
+	 */
+	void Write(const char *buffer, size_t l);
+
+	/** Called with data from the socket
+	 * @param buffer The data
+	 * @param l The length of buffer
+	 * @return true to continue reading, false to drop the socket
+	 */
+	virtual bool Read(const char *buffer, size_t l);
+};
+
 class CoreExport ListenSocket : public Socket
 {
  public:
@@ -341,22 +378,18 @@ class CoreExport ListenSocket : public Socket
  	 * @param addr The sockaddr for where the connection came from
 	 * @return The new socket
 	 */
-	virtual ClientSocket *OnAccept(int fd, const sockaddrs &addr);
+	virtual ClientSocket *OnAccept(int fd, const sockaddrs &addr) = 0;
 };
 
-class CoreExport ConnectionSocket : public BufferedSocket
+class CoreExport ConnectionSocket : public virtual Socket
 {
  public:
 	/* Sockaddrs for connection ip/port */
 	sockaddrs conaddr;
-	/* True if connected */
-	bool connected;
 
 	/** Constructor
-	 * @param ipv6 true to use IPv6
-	 * @param type The socket type, defaults to SOCK_STREAM
 	 */
-	ConnectionSocket(bool ipv6 = false, int type = SOCK_STREAM);
+	ConnectionSocket();
 
 	/** Connect the socket
 	 * @param TargetHost The target host to connect to
@@ -364,15 +397,11 @@ class CoreExport ConnectionSocket : public BufferedSocket
 	 */
 	void Connect(const Anope::string &TargetHost, int Port);
 
-	/** Called when there is something to be received for this socket
-	 * @return true on success, false to drop this socket
-	 */
-	bool ProcessRead();
-
-	/** Called when the socket is ready to be written to
-	 * @return true on success, false to drop this socket
+	/** Called when there either is a read or write event.
+	 * Used to determine whether or not this socket is connected yet.
+	 * @return true to continue to call ProcessRead/ProcessWrite, false to not continue
 	 */
-	bool ProcessWrite();
+	bool Process();
 
 	/** Called when there is an error for this socket
 	 * @return true on success, false to drop this socket
@@ -389,7 +418,7 @@ class CoreExport ConnectionSocket : public BufferedSocket
 	virtual void OnError(const Anope::string &error);
 };
 
-class CoreExport ClientSocket : public BufferedSocket
+class CoreExport ClientSocket : public virtual Socket
 {
  public:
 	/* Listen socket this connection came from */
@@ -399,23 +428,31 @@ class CoreExport ClientSocket : public BufferedSocket
 
 	/** Constructor
 	 * @param ls Listen socket this connection is from
-	 * @param fd New FD for this socket
 	 * @param addr Address the connection came from
 	 */
-	ClientSocket(ListenSocket *ls, int fd, const sockaddrs &addr);
+	ClientSocket(ListenSocket *ls, const sockaddrs &addr);
 
-	/** Called when there is something to be received for this socket
-	 * @return true on success, false to drop this socket
+	/** Called when there either is a read or write event.
+	 * Used to determine whether or not this socket is connected yet.
+	 * @return true to continue to call ProcessRead/ProcessWrite, false to not continue
 	 */
-	bool ProcessRead();
+	bool Process();
 
-	/** Called when the socket is ready to be written to
+	/** Called when there is an error for this socket
 	 * @return true on success, false to drop this socket
 	 */
-	bool ProcessWrite();
+	void ProcessError();
+
+	/** Called when a client has been accepted() successfully.
+	 */
+	virtual void OnAccept();
+
+	/** Called when there was an error accepting the client
+	 */
+	virtual void OnError(const Anope::string &error);
 };
 
-class CoreExport Pipe : public BufferedSocket
+class CoreExport Pipe : public Socket
 {
  public:
  	/** The FD of the write pipe (if this isn't evenfd)
@@ -427,13 +464,13 @@ class CoreExport Pipe : public BufferedSocket
 	 */
 	Pipe();
 
-	/** Called when data is to be read
+	/** Destructor
 	 */
-	bool ProcessRead();
+	~Pipe();
 
-	/** Function that calls OnNotify
+	/** Called when data is to be read
 	 */
-	bool Read(const Anope::string &);
+	bool ProcessRead();
 
 	/** Called when this pipe needs to be woken up
 	 */
diff --git a/modules/extra/m_dnsbl.cpp b/modules/extra/m_dnsbl.cpp
index f62908b39..7e5342d3c 100644
--- a/modules/extra/m_dnsbl.cpp
+++ b/modules/extra/m_dnsbl.cpp
@@ -56,20 +56,19 @@ class DNSBLResolver : public DNSRequest
 		reason = reason.replace_all_cs("%r", record_reason);
 		reason = reason.replace_all_cs("%N", Config->NetworkName);
 
-		XLine *x = NULL;
 		BotInfo *operserv = findbot(Config->OperServ);
-		if (this->add_to_akill && akills && (x = akills->Add(Anope::string("*@") + user->host, Config->OperServ, Anope::CurTime + this->blacklist.bantime, reason)))
+		Log(operserv) << "DNSBL: " << user->GetMask() << " appears in " << this->blacklist.name;
+		if (this->add_to_akill && akills)
 		{
-			Log(operserv) << "DNSBL: " << user->GetMask() << " appears in " << this->blacklist.name;
+			XLine *x = akills->Add("*@" + user->host, Config->OperServ, Anope::CurTime + this->blacklist.bantime, reason);
 			/* If AkillOnAdd is disabled send it anyway, noone wants bots around... */
 			if (!Config->AkillOnAdd)
-				ircdproto->SendAkill(user, x);
+				akills->Send(NULL, x);
 		}
 		else
 		{
-			Log(operserv) << "DNSBL: " << user->GetMask() << " appears in " << this->blacklist.name << "(" << reason << ")";
-			XLine xline(Anope::string("*@") + user->host, Config->OperServ, Anope::CurTime + this->blacklist.bantime, reason);
-			ircdproto->SendAkill(user, &xline);
+			XLine xline("*@" + user->host, Config->OperServ, Anope::CurTime + this->blacklist.bantime, reason);
+			ircdproto->SendAkill(NULL, &xline);
 		}
 	}
 };
diff --git a/modules/extra/m_ssl.cpp b/modules/extra/m_ssl.cpp
index 41bc49993..5143d964a 100644
--- a/modules/extra/m_ssl.cpp
+++ b/modules/extra/m_ssl.cpp
@@ -31,8 +31,6 @@ class SSLSocketIO : public SocketIO
  public:
 	/* The SSL socket for this socket */
 	SSL *sslsock;
-	/* -1 if not, 0 if waiting, 1 if true */
-	int connected, accepted;
 
 	/** Constructor
 	 */
@@ -46,12 +44,12 @@ class SSLSocketIO : public SocketIO
 	 */
 	int Recv(Socket *s, char *buf, size_t sz);
 
-	/** Really write something to the socket
- 	 * @param s The socket
-	 * @param buf What to write
-	 * @return Number of bytes written
+	/** Write something to the socket
+	 * @param s The socket
+	 * @param buf The data to write
+	 * @param size The length of the data
 	 */
-	int Send(Socket *s, const Anope::string &buf);
+	int Send(Socket *s, const char *buf, size_t sz);
 
 	/** Accept a connection from a socket
 	 * @param s The socket
@@ -59,11 +57,11 @@ class SSLSocketIO : public SocketIO
 	 */
 	ClientSocket *Accept(ListenSocket *s);
 
-	/** Check if a connection has been accepted
-	 * @param s The client socket
-	 * @return -1 on error, 0 to wait, 1 on success
+	/** Finished accepting a connection from a socket
+	 * @param s The socket
+	 * @return SF_ACCEPTED if accepted, SF_ACCEPTING if still in process, SF_DEAD on error
 	 */
-	int Accepted(ClientSocket *cs);
+	SocketFlag FinishAccept(ClientSocket *cs);
 
 	/** Connect the socket
 	 * @param s THe socket
@@ -72,11 +70,11 @@ class SSLSocketIO : public SocketIO
 	 */
 	void Connect(ConnectionSocket *s, const Anope::string &target, int port);
 
-	/** Check if this socket is connected
+	/** Called to potentially finish a pending connection
 	 * @param s The socket
- 	 * @return -1 for error, 0 for wait, 1 for connected
+	 * @return SF_CONNECTED on success, SF_CONNECTING if still pending, and SF_DEAD on error.
 	 */
-	int Connected(ConnectionSocket *s);
+	SocketFlag FinishConnect(ConnectionSocket *s);
 
 	/** Called when the socket is destructing
 	 */
@@ -194,7 +192,7 @@ void MySSLService::Init(Socket *s)
 	s->IO = new SSLSocketIO();
 }
 
-SSLSocketIO::SSLSocketIO() : connected(-1), accepted(-1)
+SSLSocketIO::SSLSocketIO()
 {
 	this->sslsock = NULL;
 }
@@ -206,9 +204,9 @@ int SSLSocketIO::Recv(Socket *s, char *buf, size_t sz)
 	return i;
 }
 
-int SSLSocketIO::Send(Socket *s, const Anope::string &buf)
+int SSLSocketIO::Send(Socket *s, const char *buf, size_t sz)
 {
-	int i = SSL_write(this->sslsock, buf.c_str(), buf.length());
+	int i = SSL_write(this->sslsock, buf, sz);
 	TotalWritten += i;
 	return i;
 }
@@ -217,8 +215,20 @@ ClientSocket *SSLSocketIO::Accept(ListenSocket *s)
 {
 	if (s->IO == &normalSocketIO)
 		throw SocketException("Attempting to accept on uninitialized socket with SSL");
-	
-	ClientSocket *newsocket = normalSocketIO.Accept(s);
+
+	sockaddrs conaddr;
+
+	socklen_t size = sizeof(conaddr);
+	int newsock = accept(s->GetFD(), &conaddr.sa, &size);
+
+#ifndef INVALID_SOCKET
+	const int INVALID_SOCKET = -1;
+#endif
+
+	if (newsock < 0 || newsock == INVALID_SOCKET)
+		throw SocketException("Unable to accept connection: " + Anope::LastError());
+
+	ClientSocket *newsocket = s->OnAccept(newsock, conaddr);
 	me->service.Init(newsocket);
 	SSLSocketIO *IO = debug_cast<SSLSocketIO *>(newsocket->IO);
 
@@ -231,47 +241,47 @@ ClientSocket *SSLSocketIO::Accept(ListenSocket *s)
 	if (!SSL_set_fd(IO->sslsock, newsocket->GetFD()))
 		throw SocketException("Unable to set SSL fd");
 
-	int ret = SSL_accept(IO->sslsock);
-	if (ret <= 0)
-	{
-		IO->accepted = 0;
-		int error = SSL_get_error(IO->sslsock, ret);
-		if (ret == -1 && (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE))
-		{
-			SocketEngine::MarkWritable(newsocket);
-			return newsocket;
-		}
-		
-		throw SocketException("Unable to accept new SSL connection: " + Anope::string(ERR_error_string(ERR_get_error(), NULL)));
-	}
+	newsocket->SetFlag(SF_ACCEPTING);
+	this->FinishAccept(newsocket);
 	
-	IO->accepted = 1;
 	return newsocket;
 }
 
-int SSLSocketIO::Accepted(ClientSocket *cs)
+SocketFlag SSLSocketIO::FinishAccept(ClientSocket *cs)
 {
-	SSLSocketIO *IO = debug_cast<SSLSocketIO *>(cs->IO);
+	if (cs->IO == &normalSocketIO)
+		throw SocketException("Attempting to finish connect uninitialized socket with SSL");
+	else if (cs->HasFlag(SF_ACCEPTED))
+		return SF_ACCEPTED;
+	else if (!cs->HasFlag(SF_ACCEPTING))
+		throw SocketException("SSLSocketIO::FinishAccept called for a socket not accepted nor accepting?");
 
-	if (IO->accepted == 0)
+	SSLSocketIO *IO = debug_cast<SSLSocketIO *>(cs->IO);
+	
+	int ret = SSL_accept(IO->sslsock);
+	if (ret <= 0)
 	{
-		int ret = SSL_accept(IO->sslsock);
-		if (ret <= 0)
+		int error = SSL_get_error(IO->sslsock, ret);
+		if (ret == -1 && (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE))
 		{
-			int error = SSL_get_error(IO->sslsock, ret);
-			if (ret == -1 && (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE))
-			{
-				SocketEngine::MarkWritable(cs);
-				return 0;
-			}
-
-			return -1;
+			SocketEngine::MarkWritable(cs);
+			return SF_ACCEPTING;
+		}
+		else
+		{
+			cs->OnError(ERR_error_string(ERR_get_error(), NULL));
+			cs->SetFlag(SF_DEAD);
+			cs->UnsetFlag(SF_ACCEPTING);
+			return SF_DEAD;
 		}
-		IO->accepted = 1;
-		return 0;
 	}
-
-	return IO->accepted;
+	else
+	{
+		cs->SetFlag(SF_ACCEPTED);
+		cs->UnsetFlag(SF_ACCEPTING);
+		cs->OnAccept();
+		return SF_ACCEPTED;
+	}
 }
 
 void SSLSocketIO::Connect(ConnectionSocket *s, const Anope::string &target, int port)
@@ -279,62 +289,78 @@ void SSLSocketIO::Connect(ConnectionSocket *s, const Anope::string &target, int
 	if (s->IO == &normalSocketIO)
 		throw SocketException("Attempting to connect uninitialized socket with SSL");
 
-	normalSocketIO.Connect(s, target, port);
-
-	SSLSocketIO *IO = debug_cast<SSLSocketIO *>(s->IO);
+	s->UnsetFlag(SF_CONNECTING);
+	s->UnsetFlag(SF_CONNECTED);
 
-	IO->sslsock = SSL_new(client_ctx);
-	if (!IO->sslsock)
-		throw SocketException("Unable to initialize SSL socket");
-
-	if (!SSL_set_fd(IO->sslsock, s->GetFD()))
-		throw SocketException("Unable to set SSL fd");
-
-	int ret = SSL_connect(IO->sslsock);
-	if (ret <= 0)
+	s->conaddr.pton(s->IsIPv6() ? AF_INET6 : AF_INET, target, port);
+	int c = connect(s->GetFD(), &s->conaddr.sa, s->conaddr.size());
+	if (c == -1)
 	{
-		IO->connected = 0;
-		int error = SSL_get_error(IO->sslsock, ret);
-		if (ret == -1 && (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE))
+		if (Anope::LastErrorCode() != EINPROGRESS)
+		{
+			s->OnError(Anope::LastError());
+			s->SetFlag(SF_DEAD);
+			return;
+		}
+		else
 		{
 			SocketEngine::MarkWritable(s);
+			s->SetFlag(SF_CONNECTING);
 			return;
 		}
-
-		s->ProcessError();
 	}
-
-	IO->connected = 1;
+	else
+	{
+		s->SetFlag(SF_CONNECTING);
+		this->FinishConnect(s);
+	}
 }
 
-int SSLSocketIO::Connected(ConnectionSocket *s)
+SocketFlag SSLSocketIO::FinishConnect(ConnectionSocket *s)
 {
 	if (s->IO == &normalSocketIO)
-		throw SocketException("Connected() called for non ssl socket?");
-	
-	int i = SocketIO::Connected(s);
-	if (i != 1)
-		return i;
+		throw SocketException("Attempting to finish connect uninitialized socket with SSL");
+	else if (s->HasFlag(SF_CONNECTED))
+		return SF_CONNECTED;
+	else if (!s->HasFlag(SF_CONNECTING))
+		throw SocketException("SSLSocketIO::FinishConnect called for a socket not connected nor connecting?");
 
 	SSLSocketIO *IO = debug_cast<SSLSocketIO *>(s->IO);
 
-	if (IO->connected == 0)
+	if (IO->sslsock == NULL)
 	{
-		int ret = SSL_connect(IO->sslsock);
-		if (ret <= 0)
-		{
-			int error = SSL_get_error(IO->sslsock, ret);
-			if (ret == -1 && (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE))
-				return 0;
+		IO->sslsock = SSL_new(client_ctx);
+		if (!IO->sslsock)
+			throw SocketException("Unable to initialize SSL socket");
 
-			s->ProcessError();
-			return -1;
-		}
-		IO->connected = 1;
-		return 0; // poll for next read/write (which will be real), don't assume ones available
+		if (!SSL_set_fd(IO->sslsock, s->GetFD()))
+			throw SocketException("Unable to set SSL fd");
 	}
 	
-	return IO->connected;
+	int ret = SSL_connect(IO->sslsock);
+	if (ret <= 0)
+	{
+		int error = SSL_get_error(IO->sslsock, ret);
+		if (ret == -1 && (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE))
+		{
+			SocketEngine::MarkWritable(s);
+			return SF_CONNECTING;
+		}
+		else
+		{
+			s->OnError(ERR_error_string(ERR_get_error(), NULL));
+			s->UnsetFlag(SF_CONNECTING);
+			s->SetFlag(SF_DEAD);
+			return SF_DEAD;
+		}
+	}
+	else
+	{
+		s->UnsetFlag(SF_CONNECTING);
+		s->SetFlag(SF_CONNECTED);
+		s->OnConnect();
+		return SF_CONNECTED;
+	}
 }
 
 void SSLSocketIO::Destroy()
diff --git a/modules/extra/m_xmlrpc.cpp b/modules/extra/m_xmlrpc.cpp
index 9d301d684..38a733c7c 100644
--- a/modules/extra/m_xmlrpc.cpp
+++ b/modules/extra/m_xmlrpc.cpp
@@ -9,7 +9,7 @@ class MyXMLRPCClientSocket : public XMLRPCClientSocket
 	/* Used to skip the (optional) HTTP header,  which we really don't care about */
 	bool in_query;
  public:
-	MyXMLRPCClientSocket(XMLRPCListenSocket *ls, int fd, const sockaddrs &addr) : XMLRPCClientSocket(ls, fd, addr), in_query(false)
+	MyXMLRPCClientSocket(XMLRPCListenSocket *ls, int fd, const sockaddrs &addr) : Socket(fd, ls->IsIPv6()), XMLRPCClientSocket(ls, addr), in_query(false)
 	{
 	}
 
@@ -245,9 +245,9 @@ class ModuleXMLRPC : public Module
 			Socket *s = it->second;
 			++it;
 
-			if (s->Type == SOCKTYPE_CLIENT)
+			ClientSocket *cs = dynamic_cast<ClientSocket *>(s);
+			if (cs != NULL)
 			{
-				ClientSocket *cs = debug_cast<ClientSocket *>(s);
 				for (unsigned i = 0; i < listen_sockets.size(); ++i)
 					if (cs->LS == listen_sockets[i])
 					{
diff --git a/modules/extra/xmlrpc.h b/modules/extra/xmlrpc.h
index 38f548c4f..364617ecc 100644
--- a/modules/extra/xmlrpc.h
+++ b/modules/extra/xmlrpc.h
@@ -3,13 +3,13 @@ class XMLRPCClientSocket;
 class XMLRPCListenSocket;
 class XMLRPCServiceInterface;
 
-class XMLRPCClientSocket : public ClientSocket
+class XMLRPCClientSocket : public ClientSocket, public BufferedSocket
 {
  protected:
 	Anope::string query;
 
  public:
-	XMLRPCClientSocket(ListenSocket *ls, int fd, const sockaddrs &addr) : ClientSocket(ls, fd, addr) { }
+	XMLRPCClientSocket(ListenSocket *ls, const sockaddrs &addr) : ClientSocket(ls, addr), BufferedSocket() { }
 
 	virtual ~XMLRPCClientSocket() { }
 
diff --git a/src/main.cpp b/src/main.cpp
index 3730c60ea..b00a6ee6e 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -81,7 +81,7 @@ class UpdateTimer : public Timer
 	}
 };
 
-ConnectionSocket *UplinkSock = NULL;
+UplinkSocket *UplinkSock = NULL;
 int CurrentUplink = 0;
 
 static void Connect();
@@ -105,77 +105,72 @@ class ReconnectTimer : public Timer
 	}
 };
 
-class UplinkSocket : public ConnectionSocket
+UplinkSocket::UplinkSocket() : Socket(-1, Config->Uplinks[CurrentUplink]->ipv6), ConnectionSocket(), BufferedSocket()
 {
- public:
-	UplinkSocket() : ConnectionSocket(Config->Uplinks[CurrentUplink]->ipv6)
-	{
-		UplinkSock = this;
-	}
+	UplinkSock = this;
+}
 
-	~UplinkSocket()
+UplinkSocket::~UplinkSocket()
+{
+	if (ircdproto && Me && !Me->GetLinks().empty() && Me->GetLinks()[0]->IsSynced())
 	{
-		if (ircdproto && Me && !Me->GetLinks().empty() && Me->GetLinks()[0]->IsSynced())
+		FOREACH_MOD(I_OnServerDisconnect, OnServerDisconnect());
+
+		for (Anope::insensitive_map<User *>::const_iterator it = UserListByNick.begin(); it != UserListByNick.end(); ++it)
 		{
-			FOREACH_MOD(I_OnServerDisconnect, OnServerDisconnect());
+			User *u = it->second;
 
-			for (Anope::insensitive_map<User *>::const_iterator it = UserListByNick.begin(); it != UserListByNick.end(); ++it)
+			if (u->server == Me)
 			{
-				User *u = it->second;
-
-				if (u->server == Me)
-				{
-					/* Don't use quitmsg here, it may contain information you don't want people to see */
-					ircdproto->SendQuit(u, "Shutting down");
-					BotInfo *bi = findbot(u->nick);
-					if (bi != NULL)
-						bi->introduced = false;
-				}
+				/* Don't use quitmsg here, it may contain information you don't want people to see */
+				ircdproto->SendQuit(u, "Shutting down");
+				BotInfo *bi = findbot(u->nick);
+				if (bi != NULL)
+					bi->introduced = false;
 			}
-
-			ircdproto->SendSquit(Config->ServerName, quitmsg);
-
-			this->ProcessWrite(); // Write out the last bit
 		}
 
-		for (unsigned i = Me->GetLinks().size(); i > 0; --i)
-			if (!Me->GetLinks()[i - 1]->HasFlag(SERVER_JUPED))
-				Me->GetLinks()[i - 1]->Delete(Me->GetName() + " " + Me->GetLinks()[i - 1]->GetName());
+		ircdproto->SendSquit(Config->ServerName, quitmsg);
 
-		UplinkSock = NULL;
+		this->ProcessWrite(); // Write out the last bit
+	}
 
-		Me->SetFlag(SERVER_SYNCING);
+	for (unsigned i = Me->GetLinks().size(); i > 0; --i)
+		if (!Me->GetLinks()[i - 1]->HasFlag(SERVER_JUPED))
+			Me->GetLinks()[i - 1]->Delete(Me->GetName() + " " + Me->GetLinks()[i - 1]->GetName());
 
-		if (!quitting)
-		{
-			int Retry = Config->RetryWait;
-			if (Retry <= 0)
-				Retry = 60;
+	UplinkSock = NULL;
 
-			Log() << "Retrying in " << Retry << " seconds";
-			new ReconnectTimer(Retry);
-		}
-	}
+	Me->SetFlag(SERVER_SYNCING);
 
-	bool Read(const Anope::string &buf)
+	if (!quitting)
 	{
-		process(buf);
-		return true;
-	}
+		int Retry = Config->RetryWait;
+		if (Retry <= 0)
+			Retry = 60;
 
-	void OnConnect()
-	{
-		Log() << "Successfully connected to " << Config->Uplinks[CurrentUplink]->host << ":" << Config->Uplinks[CurrentUplink]->port;
-		ircdproto->SendConnect();
-		FOREACH_MOD(I_OnServerConnect, OnServerConnect());
+		Log() << "Retrying in " << Retry << " seconds";
+		new ReconnectTimer(Retry);
 	}
+}
 
-	void OnError(const Anope::string &error)
-	{
-		Log() << "Unable to connect to server " << Config->Uplinks[CurrentUplink]->host << ":" << Config->Uplinks[CurrentUplink]->port << (!error.empty() ? (": " + error) : "");
-		this->SetFlag(SF_DEAD);
-	}
-};
+bool UplinkSocket::Read(const Anope::string &buf)
+{
+	process(buf);
+	return true;
+}
+
+void UplinkSocket::OnConnect()
+{
+	Log() << "Successfully connected to " << Config->Uplinks[CurrentUplink]->host << ":" << Config->Uplinks[CurrentUplink]->port;
+	ircdproto->SendConnect();
+	FOREACH_MOD(I_OnServerConnect, OnServerConnect());
+}
+
+void UplinkSocket::OnError(const Anope::string &error)
+{
+	Log() << "Unable to connect to server " << Config->Uplinks[CurrentUplink]->host << ":" << Config->Uplinks[CurrentUplink]->port << (!error.empty() ? (": " + error) : "");
+}
 
 static void Connect()
 {
diff --git a/src/socket_clients.cpp b/src/socket_clients.cpp
new file mode 100644
index 000000000..4a43629a4
--- /dev/null
+++ b/src/socket_clients.cpp
@@ -0,0 +1,96 @@
+/*
+ *
+ * (C) 2003-2011 Anope Team
+ * Contact us at team@anope.org
+ *
+ * Please read COPYING and README for further details.
+ *
+ * Based on the original code of Epona by Lara.
+ * Based on the original code of Services by Andy Church.
+ */
+
+#include "services.h"
+
+ConnectionSocket::ConnectionSocket() : Socket()
+{
+}
+
+void ConnectionSocket::Connect(const Anope::string &TargetHost, int Port)
+{
+	this->IO->Connect(this, TargetHost, Port);
+}
+
+bool ConnectionSocket::Process()
+{
+	try
+	{
+		if (this->HasFlag(SF_CONNECTED))
+			return true;
+		else if (this->HasFlag(SF_CONNECTING))
+			this->SetFlag(this->IO->FinishConnect(this));
+		else
+			this->SetFlag(SF_DEAD);
+	}
+	catch (const SocketException &ex)
+	{
+		Log() << ex.GetReason();
+	}
+	return false;
+}
+
+void ConnectionSocket::ProcessError()
+{
+	int optval = 0;
+	socklen_t optlen = sizeof(optval);
+	getsockopt(this->GetFD(), SOL_SOCKET, SO_ERROR, reinterpret_cast<char *>(&optval), &optlen);
+	errno = optval;
+	this->OnError(optval ? Anope::LastError() : "");
+}
+
+void ConnectionSocket::OnConnect()
+{
+}
+
+void ConnectionSocket::OnError(const Anope::string &)
+{
+}
+
+ClientSocket::ClientSocket(ListenSocket *ls, const sockaddrs &addr) : Socket(), LS(ls), clientaddr(addr)
+{
+}
+
+bool ClientSocket::Process()
+{
+	try
+	{
+		if (this->HasFlag(SF_ACCEPTED))
+			return true;
+		else if (this->HasFlag(SF_ACCEPTING))
+			this->SetFlag(this->IO->FinishAccept(this));
+		else
+			this->SetFlag(SF_DEAD);
+	}
+	catch (const SocketException &ex)
+	{
+		Log() << ex.GetReason();
+	}
+	return false;
+}
+
+void ClientSocket::ProcessError()
+{
+	int optval = 0;
+	socklen_t optlen = sizeof(optval);
+	getsockopt(this->GetFD(), SOL_SOCKET, SO_ERROR, reinterpret_cast<char *>(&optval), &optlen);
+	errno = optval;
+	this->OnError(optval ? Anope::LastError() : "");
+}
+
+void ClientSocket::OnAccept()
+{
+}
+
+void ClientSocket::OnError(const Anope::string &error)
+{
+}
+
diff --git a/src/socket_transport.cpp b/src/socket_transport.cpp
new file mode 100644
index 000000000..a7b92ff20
--- /dev/null
+++ b/src/socket_transport.cpp
@@ -0,0 +1,185 @@
+/*
+ *
+ * (C) 2003-2011 Anope Team
+ * Contact us at team@anope.org
+ *
+ * Please read COPYING and README for further details.
+ *
+ * Based on the original code of Epona by Lara.
+ * Based on the original code of Services by Andy Church.
+ */
+
+#include "services.h"
+
+BufferedSocket::BufferedSocket()
+{
+}
+
+BufferedSocket::~BufferedSocket()
+{
+}
+
+bool BufferedSocket::ProcessRead()
+{
+	char tbuffer[NET_BUFSIZE];
+
+	this->RecvLen = 0;
+
+	int len = this->IO->Recv(this, tbuffer, sizeof(tbuffer) - 1);
+	if (len <= 0)
+		return false;
+