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

10 files changed, 490 insertions, 548 deletions

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;
+	
+	tbuffer[len] = 0;
+	this->RecvLen = len;
+
+	Anope::string sbuffer = this->extrabuf;
+	sbuffer += tbuffer;
+	this->extrabuf.clear();
+	size_t lastnewline = sbuffer.rfind('\n');
+	if (lastnewline == Anope::string::npos)
+	{
+		this->extrabuf = sbuffer;
+		return true;
+	}
+	if (lastnewline < sbuffer.length() - 1)
+	{
+		this->extrabuf = sbuffer.substr(lastnewline);
+		this->extrabuf.trim();
+		sbuffer = sbuffer.substr(0, lastnewline);
+	}
+
+	sepstream stream(sbuffer, '\n');
+
+	Anope::string tbuf;
+	while (stream.GetToken(tbuf))
+	{
+		tbuf.trim();
+		if (!tbuf.empty() && !Read(tbuf))
+			return false;
+	}
+
+	return true;
+}
+
+bool BufferedSocket::ProcessWrite()
+{
+	int count = this->IO->Send(this, this->WriteBuffer);
+	if (count <= -1)
+		return false;
+	this->WriteBuffer = this->WriteBuffer.substr(count);
+	if (this->WriteBuffer.empty())
+		SocketEngine::ClearWritable(this);
+
+	return true;
+}
+
+bool BufferedSocket::Read(const Anope::string &buf)
+{
+	return false;
+}
+
+void BufferedSocket::Write(const char *message, ...)
+{
+	va_list vi;
+	char tbuffer[BUFSIZE];
+
+	if (!message)
+		return;
+
+	va_start(vi, message);
+	vsnprintf(tbuffer, sizeof(tbuffer), message, vi);
+	va_end(vi);
+
+	Anope::string sbuf = tbuffer;
+	Write(sbuf);
+}
+
+void BufferedSocket::Write(const Anope::string &message)
+{
+	this->WriteBuffer += message + "\r\n";
+	SocketEngine::MarkWritable(this);
+}
+
+int BufferedSocket::ReadBufferLen() const
+{
+	return RecvLen;
+}
+
+int BufferedSocket::WriteBufferLen() const
+{
+	return this->WriteBuffer.length();
+}
+
+
+BinarySocket::DataBlock::DataBlock(const char *b, size_t l)
+{
+	this->buf = new char[l];
+	memcpy(this->buf, b, l);
+	this->len = l;
+}
+
+BinarySocket::DataBlock::~DataBlock()
+{
+	delete [] this->buf;
+}
+
+BinarySocket::BinarySocket()
+{
+}
+
+BinarySocket::~BinarySocket()
+{
+}
+
+bool BinarySocket::ProcessRead()
+{
+	char tbuffer[NET_BUFSIZE];
+
+	int len = this->IO->Recv(this, tbuffer, sizeof(tbuffer));
+	if (len <= 0)
+		return false;
+	
+	return this->Read(tbuffer, len);
+}
+
+bool BinarySocket::ProcessWrite()
+{
+	if (this->WriteBuffer.empty())
+	{
+		SocketEngine::ClearWritable(this);
+		return true;
+	}
+
+	DataBlock *d = this->WriteBuffer.front();
+
+	int len = this->IO->Send(this, d->buf, d->len);
+	if (len <= -1)
+		return false;
+	else if (static_cast<size_t>(len) == d->len)
+	{
+		delete d;
+		this->WriteBuffer.pop_front();
+	}
+	else
+	{
+		d->buf += len;
+		d->len -= len;
+	}
+
+	if (this->WriteBuffer.empty())
+		SocketEngine::ClearWritable(this);
+
+	return true;
+}
+
+void BinarySocket::Write(const char *buffer, size_t l)
+{
+	this->WriteBuffer.push_back(new DataBlock(buffer, l));
+	SocketEngine::MarkWritable(this);
+}
+
+bool BinarySocket::Read(const char *buffer, size_t l)
+{
+	return true;
+}
+
diff --git a/src/socketengines/pipeengine_eventfd.cpp b/src/socketengines/pipeengine_eventfd.cpp
index aa712de3d..21931bb1b 100644
--- a/src/socketengines/pipeengine_eventfd.cpp
+++ b/src/socketengines/pipeengine_eventfd.cpp
@@ -1,64 +1,29 @@
 #include "services.h"
 #include <sys/eventfd.h>
 
-class PipeIO : public SocketIO
+Pipe::Pipe() : Socket(eventfd(0, EFD_NONBLOCK))
 {
- public:
-	/** Receive something from the buffer
- 	 * @param s The socket
-	 * @param buf The buf to read to
-	 * @param sz How much to read
-	 * @return Number of bytes received
-	 */
-	int Recv(Socket *s, char *buf, size_t sz)
-	{
-		static eventfd_t dummy;
-		return !eventfd_read(s->GetFD(), &dummy);
-	}
-
-	/** Write something to the socket
- 	 * @param s The socket
-	 * @param buf What to write
-	 * @return Number of bytes written
-	 */
-	int Send(Socket *s, const Anope::string &buf)
-	{
-		return !eventfd_write(s->GetFD(), 1);
-	}
-} pipeSocketIO;
-
-Pipe::Pipe() : BufferedSocket()
-{
-	this->IO = &pipeSocketIO;
-	this->Sock = eventfd(0, EFD_NONBLOCK);
 	if (this->Sock < 0)
-		throw CoreException(Anope::string("Could not create pipe: ") + Anope::LastError());
-
-	this->IPv6 = false;
+		throw CoreException("Could not create pipe: " + Anope::LastError());
 
 	SocketEngine::AddSocket(this);
 }
 
-bool Pipe::ProcessRead()
+Pipe::~Pipe()
 {
-	this->IO->Recv(this, NULL, 0);
-	return this->Read("");
 }
 
-bool Pipe::Read(const Anope::string &)
+bool Pipe::ProcessRead()
 {
+	eventfd_t dummy;
+	eventfd_read(this->GetFD(), &dummy);
 	this->OnNotify();
 	return true;
 }
 
 void Pipe::Notify()
 {
-	/* Note we send this immediatly. If use use Socket::Write and if this functions is called
-	 * from a thread, only epoll is able to pick up the change to this sockets want flags immediately
-	 * Other engines time out then pick up and write the change then read it back, which
-	 * is too slow for most things.
-	 */
-	this->IO->Send(this, "");
+	eventfd_write(this->GetFD(), 1);
 }
 
 void Pipe::OnNotify()
diff --git a/src/socketengines/pipeengine_pipe.cpp b/src/socketengines/pipeengine_pipe.cpp
index b3ca15d87..ceaded8ea 100644
--- a/src/socketengines/pipeengine_pipe.cpp
+++ b/src/socketengines/pipeengine_pipe.cpp
@@ -1,67 +1,40 @@
 #include "services.h"
 
-class PipeIO : public SocketIO
-{
- public:
-	/** Receive something from the buffer
- 	 * @param s The socket
-	 * @param buf The buf to read to
-	 * @param sz How much to read
-	 * @return Number of bytes received
-	 */
-	int Recv(Socket *s, char *buf, size_t sz)
-	{
-		static char dummy[512];
-		while (read(s->GetFD(), &dummy, 512) == 512);
-		return 0;
-	}
-
-	/** Write something to the socket
- 	 * @param s The socket
-	 * @param buf What to write
-	 * @return Number of bytes written
-	 */
-	int Send(Socket *s, const Anope::string &buf)
-	{
-		static const char dummy = '*';
-		Pipe *pipe = debug_cast<Pipe *>(s);
-		return write(pipe->WritePipe, &dummy, 1);
-	}
-} pipeSocketIO;
-
-Pipe::Pipe() : BufferedSocket()
+Pipe::Pipe() : Socket(-1)
 {
 	int fds[2];
 	if (pipe(fds))
-		throw CoreException(Anope::string("Could not create pipe: ") + Anope::LastError());
+		throw CoreException("Could not create pipe: " + Anope::LastError());
 	int flags = fcntl(fds[0], F_GETFL, 0);
 	fcntl(fds[0], F_SETFL, flags | O_NONBLOCK);
 	flags = fcntl(fds[1], F_GETFL, 0);
 	fcntl(fds[1], F_SETFL, flags | O_NONBLOCK);
 
-	this->IO = &pipeSocketIO;
+	this->~Socket();
+
 	this->Sock = fds[0];
 	this->WritePipe = fds[1];
-	this->IPv6 = false;
 
 	SocketEngine::AddSocket(this);
 }
 
-bool Pipe::ProcessRead()
+Pipe::~Pipe()
 {
-	this->IO->Recv(this, NULL, 0);
-	return this->Read("");
+	CloseSocket(this->WritePipe);
 }
 
-bool Pipe::Read(const Anope::string &)
+bool Pipe::ProcessRead()
 {
+	char dummy[512];
+	while (read(this->GetFD(), &dummy, 512) == 512);
 	this->OnNotify();
 	return true;
 }
 
 void Pipe::Notify()
 {
-	this->IO->Send(this, "");
+	const char dummy = '*';
+	write(this->WritePipe, &dummy, 1);
 }
 
 void Pipe::OnNotify()
diff --git a/src/socketengines/pipeengine_win32.cpp b/src/socketengines/pipeengine_win32.cpp
index 3bd18d0cd..f34472870 100644
--- a/src/socketengines/pipeengine_win32.cpp
+++ b/src/socketengines/pipeengine_win32.cpp
@@ -1,89 +1,50 @@
 #include "services.h"
 
-static ClientSocket *newsocket = NULL;
-
-class LSocket : public ListenSocket
-{
- public:
-	LSocket(const Anope::string &host, int port) : ListenSocket(host, port, false) { }
-
-	ClientSocket *OnAccept(int fd, const sockaddrs &addr)
-	{
-		newsocket = new ClientSocket(this, fd, addr);
-		return newsocket;
-	}
-};
-
-class PipeIO : public SocketIO
+Pipe::Pipe() : Socket(-1)
 {
- public:
-	/** Receive something from the buffer
- 	 * @param s The socket
-	 * @param buf The buf to read to
-	 * @param sz How much to read
-	 * @return Number of bytes received
-	 */
-	int Recv(Socket *s, char *buf, size_t sz)
-	{
-		static char dummy[512];
-		return recv(s->GetFD(), dummy, 512, 0);
-	}
-
-	/** Write something to the socket
- 	 * @param s The socket
-	 * @param buf What to write
-	 * @return Number of bytes written
-	 */
-	int Send(Socket *s, const Anope::string &buf)
-	{
-		static const char dummy = '*';
-		Pipe *pipe = debug_cast<Pipe *>(s);
-		return send(pipe->WritePipe, &dummy, 1, 0);
-	}
-} pipeSocketIO;
+	sockaddrs localhost;
 
-Pipe::Pipe() : BufferedSocket()
-{
-	LSocket lfs("127.0.0.1", 0);
+	localhost.pton(AF_INET, "127.0.0.1");
 
-	int cfd = socket(AF_INET, SOCK_STREAM, 0);
+	int cfd = socket(AF_INET, SOCK_STREAM, 0), lfd = socket(AF_INET, SOCK_STREAM, 0);
 	if (cfd == -1)
 		throw CoreException("Error accepting new socket for Pipe");
+	
+	if (bind(lfd, &localhost.sa, localhost.size()) == -1)
+		throw CoreException("Error accepting new socket for Pipe");
+	if (listen(lfd, 1) == -1)
+		throw CoreException("Error accepting new socket for Pipe");
 
-	sockaddr_in addr;
-	socklen_t sz = sizeof(addr);
-	getsockname(lfs.GetFD(), reinterpret_cast<sockaddr *>(&addr), &sz);
+	sockaddrs lfd_addr;
+	socklen_t sz = sizeof(lfd_addr);
+	getsockname(lfd, &lfd_addr.sa, &sz);
 
-	if (connect(cfd, reinterpret_cast<sockaddr *>(&addr), sz))
-		throw CoreException("Error accepting new socket for Pipe");
-	lfs.ProcessRead();
-	if (!newsocket)
+	if (connect(cfd, &lfd_addr.sa, lfd_addr.size()))
 		throw CoreException("Error accepting new socket for Pipe");
+	CloseSocket(lfd);
 	
-	this->IO = &pipeSocketIO;
-	this->Sock = cfd;
-	this->WritePipe = newsocket->GetFD();
-	this->IPv6 = false;
+	this->WritePipe = cfd;
 
 	SocketEngine::AddSocket(this);
-	newsocket = NULL;
 }
 
-bool Pipe::ProcessRead()
+Pipe::~Pipe()
 {
-	this->IO->Recv(this, NULL, 0);
-	return this->Read("");
+	CloseSocket(this->WritePipe);
 }
 
-bool Pipe::Read(const Anope::string &)
+bool Pipe::ProcessRead()
 {
+	char dummy[512];
+	while (recv(this->GetFD(), dummy, 512, 0) == 512);
 	this->OnNotify();
 	return true;
 }
 
 void Pipe::Notify()
 {
-	this->IO->Send(this, "");
+	const char dummy = '*';
+	send(this->WritePipe, &dummy, 1, 0);
 }
 
 void Pipe::OnNotify()
diff --git a/src/socketengines/socketengine_epoll.cpp b/src/socketengines/socketengine_epoll.cpp
index ad9bef3ce..73f7e10d6 100644
--- a/src/socketengines/socketengine_epoll.cpp
+++ b/src/socketengines/socketengine_epoll.cpp
@@ -11,18 +11,12 @@ void SocketEngine::Init()
 	max = ulimit(4, 0);
 
 	if (max <= 0)
-	{
-		Log() << "Can't determine maximum number of open sockets";
-		throw CoreException("Can't determine maximum number of open sockets");
-	}
+		throw SocketException("Can't determine maximum number of open sockets");
 
 	EngineHandle = epoll_create(max / 4);
 
 	if (EngineHandle == -1)
-	{
-		Log() << "Could not initialize epoll socket engine: " << Anope::LastError();
-		throw CoreException(Anope::string("Could not initialize epoll socket engine: ") + Anope::LastError());
-	}
+		throw SocketException("Could not initialize epoll socket engine: " + Anope::LastError());
 
 	events = new epoll_event[max];
 	memset(events, 0, sizeof(epoll_event) * max);
@@ -53,10 +47,7 @@ void SocketEngine::AddSocket(Socket *s)
 	ev.data.fd = s->GetFD();
 
 	if (epoll_ctl(EngineHandle, EPOLL_CTL_ADD, ev.data.fd, &ev) == -1)
-	{
-		Log() << "Unable to add fd " << ev.data.fd << " to socketengine epoll: " << Anope::LastError();
-		return;
-	}
+		throw SocketException("Unable to add fd " + stringify(ev.data.fd) + " to epoll: " + Anope::LastError());
 
 	Sockets.insert(std::make_pair(ev.data.fd, s));
 }
@@ -70,10 +61,7 @@ void SocketEngine::DelSocket(Socket *s)
 	ev.data.fd = s->GetFD();
 
 	if (epoll_ctl(EngineHandle, EPOLL_CTL_DEL, ev.data.fd, &ev) == -1)
-	{
-		Log() << "Unable to delete fd " << ev.data.fd << " from socketengine epoll: " << Anope::LastError();
-		return;
-	}
+		throw SocketException("Unable to remove fd " + stringify(ev.data.fd) + " from epoll: " + Anope::LastError());
 
 	Sockets.erase(ev.data.fd);
 }
@@ -91,9 +79,9 @@ void SocketEngine::MarkWritable(Socket *s)
 	ev.data.fd = s->GetFD();
 
 	if (epoll_ctl(EngineHandle, EPOLL_CTL_MOD, ev.data.fd, &ev) == -1)
-		Log() << "Unable to mark fd " << ev.data.fd << " as writable in socketengine epoll: " << Anope::LastError();
-	else
-		s->SetFlag(SF_WRITABLE);
+		throw SocketException("Unable to mark fd " + stringify(ev.data.fd) + " as writable in epoll: " + Anope::LastError());
+
+	s->SetFlag(SF_WRITABLE);
 }
 
 void SocketEngine::ClearWritable(Socket *s)
@@ -109,9 +97,9 @@ void SocketEngine::ClearWritable(Socket *s)
 	ev.data.fd = s->GetFD();
 
 	if (epoll_ctl(EngineHandle, EPOLL_CTL_MOD, ev.data.fd, &ev) == -1)
-		Log() << "Unable to mark fd " << ev.data.fd << " as unwritable in socketengine epoll: " << Anope::LastError();
-	else
-		s->UnsetFlag(SF_WRITABLE);
+		throw SocketException("Unable clear mark fd " + stringify(ev.data.fd) + " as writable in epoll: " + Anope::LastError());
+
+	s->UnsetFlag(SF_WRITABLE);
 }
 
 void SocketEngine::Process()
@@ -130,10 +118,15 @@ void SocketEngine::Process()
 	for (int i = 0; i < total; ++i)
 	{
 		epoll_event *ev = &events[i];
-		Socket *s = Sockets[ev->data.fd];
+
+		std::map<int, Socket *>::iterator it = Sockets.find(ev->data.fd);
+		if (it == Sockets.end())
+			continue;
+		Socket *s = it->second;
 
 		if (s->HasFlag(SF_DEAD))
 			continue;
+
 		if (ev->events & (EPOLLHUP | EPOLLERR))
 		{
 			s->ProcessError();
@@ -141,6 +134,9 @@ void SocketEngine::Process()
 			continue;
 		}
 
+		if (!s->Process())
+			continue;
+
 		if ((ev->events & EPOLLIN) && !s->ProcessRead())
 			s->SetFlag(SF_DEAD);
 
@@ -151,7 +147,10 @@ void SocketEngine::Process()
 	for (int i = 0; i < total; ++i)
 	{
 		epoll_event *ev = &events[i];
-		Socket *s = Sockets[ev->data.fd];
+		std::map<int, Socket *>::iterator it = Sockets.find(ev->data.fd);
+		if (it == Sockets.end())
+			continue;
+		Socket *s = it->second;
 
 		if (s->HasFlag(SF_DEAD))
 			delete s;
diff --git a/src/socketengines/socketengine_poll.cpp b/src/socketengines/socketengine_poll.cpp
index 278485fa7..ba111659a 100644
--- a/src/socketengines/socketengine_poll.cpp
+++ b/src/socketengines/socketengine_poll.cpp
@@ -50,10 +50,7 @@ void SocketEngine::Shutdown()
 void SocketEngine::AddSocket(Socket *s)
 {
 	if (SocketCount == max)
-	{
-		Log() << "Unable to add fd " << s->GetFD() << " to socketengine poll, engine is full";
-		return;
-	}
+		throw SocketException("Unable to add fd " + stringify(s->GetFD()) + " to poll: " + Anope::LastError());
 
 	pollfd *ev = &events[SocketCount];
 	ev->fd = s->GetFD();
@@ -70,10 +67,7 @@ void SocketEngine::DelSocket(Socket *s)
 {
 	std::map<int, int>::iterator pos = socket_positions.find(s->GetFD());
 	if (pos == socket_positions.end())
-	{
-		Log() << "Unable to delete unknown fd " << s->GetFD() << " from socketengine poll";
-		return;
-	}
+		throw SocketException("Unable to remove fd " + stringify(s->GetFD()) + " from poll");
 
 	if (pos->second != SocketCount - 1)
 	{
@@ -100,10 +94,7 @@ void SocketEngine::MarkWritable(Socket *s)
 
 	std::map<int, int>::iterator pos = socket_positions.find(s->GetFD());
 	if (pos == socket_positions.end())
-	{
-		Log() << "Unable to mark unknown fd " << s->GetFD() << " as writable";
-		return;
-	}
+		throw SocketException("Unable to mark fd " + stringify(s->GetFD()) + " as writable in poll");
 
 	pollfd *ev = &events[pos->second];
 	ev->events |= POLLOUT;
@@ -118,10 +109,7 @@ void SocketEngine::ClearWritable(Socket *s)
 
 	std::map<int, int>::iterator pos = socket_positions.find(s->GetFD());
 	if (pos == socket_positions.end())
-	{
-		Log() << "Unable to mark unknown fd " << s->GetFD() << " as writable";
-		return;
-	}
+		throw SocketException("Unable clear mark fd " + stringify(s->GetFD()) + " as writable in poll");
 
 	pollfd *ev = &events[pos->second];
 	ev->events &= ~POLLOUT;
@@ -149,10 +137,14 @@ void SocketEngine::Process()
 		if (ev->revents != 0)
 			++processed;
 
-		Socket *s = Sockets[ev->fd];
+		std::map<int, Socket *>::iterator it = Sockets.find(ev->fd);
+		if (it == Sockets.end())
+			continue;
+		Socket *s = it->second;
 
 		if (s->HasFlag(SF_DEAD))
 			continue;
+
 		if (ev->revents & (POLLERR | POLLRDHUP))
 		{
 			s->ProcessError();
@@ -160,6 +152,9 @@ void SocketEngine::Process()
 			continue;
 		}
 
+		if (!s->Process())
+			continue;
+
 		if ((ev->revents & POLLIN) && !s->ProcessRead())
 			s->SetFlag(SF_DEAD);
 
@@ -170,7 +165,10 @@ void SocketEngine::Process()
 	for (int i = 0; i < SocketCount; ++i)
 	{
 		pollfd *ev = &events[i];
-		Socket *s = Sockets[ev->fd];
+		std::map<int, Socket *>::iterator it = Sockets.find(ev->fd);
+		if (it == Sockets.end())
+			continue;
+		Socket *s = it->second;
 
 		if (s->HasFlag(SF_DEAD))
 			delete s;
diff --git a/src/socketengines/socketengine_select.cpp b/src/socketengines/socketengine_select.cpp
index 34a86f680..ad6e04ca9 100644
--- a/src/socketengines/socketengine_select.cpp
+++ b/src/socketengines/socketengine_select.cpp
@@ -96,19 +96,27 @@ void SocketEngine::Process()
 		{
 			Socket *s = it->second;
 
-			if (FD_ISSET(s->GetFD(), &efdset) || FD_ISSET(s->GetFD(), &rfdset) || FD_ISSET(s->GetFD(), &wfdset))
+			bool has_read = FD_ISSET(s->GetFD(), &rfdset), has_write = FD_ISSET(s->GetFD(), &wfdset), has_error = FD_ISSET(s->GetFD(), &efdset);
+			if (has_read || has_write || has_error)
 				++processed;
+
 			if (s->HasFlag(SF_DEAD))
 				continue;
-			if (FD_ISSET(s->GetFD(), &efdset))
+
+			if (has_error)
 			{
 				s->ProcessError();
 				s->SetFlag(SF_DEAD);
 				continue;
 			}
-			if (FD_ISSET(s->GetFD(), &rfdset) && !s->ProcessRead())
+
+			if (!s->Process())
+				continue;
+
+			if (has_read && !s->ProcessRead())
 				s->SetFlag(SF_DEAD);
-			if (FD_ISSET(s->GetFD(), &wfdset) && !s->ProcessWrite())
+
+			if (has_write && !s->ProcessWrite())
 				s->SetFlag(SF_DEAD);
 		}
 
diff --git a/src/sockets.cpp b/src/sockets.cpp
index b7f0a5f9a..541e3bb3a 100644
--- a/src/sockets.cpp
+++ b/src/sockets.cpp
@@ -260,15 +260,19 @@ int SocketIO::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 buf The data to write
+ * @param size The length of the data
  */
-int SocketIO::Send(Socket *s, const Anope::string &buf)
+int SocketIO::Send(Socket *s, const char *buf, size_t sz)
 {
-	size_t i = send(s->GetFD(), buf.c_str(), buf.length(), 0);
+	size_t i = send(s->GetFD(), buf, sz, 0);
 	TotalWritten += i;
 	return i;
 }
+int SocketIO::Send(Socket *s, const Anope::string &buf)
+{
+	return this->Send(s, buf.c_str(), buf.length());
+}
 
 /** Accept a connection from a socket
  * @param s The socket
@@ -282,22 +286,27 @@ ClientSocket *SocketIO::Accept(ListenSocket *s)
 	int newsock = accept(s->GetFD(), &conaddr.sa, &size);
 
 #ifndef INVALID_SOCKET
-# define INVALID_SOCKET -1
+	static const int INVALID_SOCKET = -1;
 #endif
 
 	if (newsock >= 0 && newsock != INVALID_SOCKET)
-		return s->OnAccept(newsock, conaddr);
+	{
+		ClientSocket *ns = s->OnAccept(newsock, conaddr);
+		ns->SetFlag(SF_ACCEPTED);
+		ns->OnAccept();
+		return ns;
+	}
 	else
 		throw SocketException("Unable to accept connection: " + Anope::LastError());
 }
 
-/** 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 SocketIO::Accepted(ClientSocket *cs)
+SocketFlag SocketIO::FinishAccept(ClientSocket *cs)
 {
-	return 1;
+	return SF_ACCEPTED;
 }
 
 /** Bind a socket
@@ -319,6 +328,8 @@ void SocketIO::Bind(Socket *s, const Anope::string &ip, int port)
   */
 void SocketIO::Connect(ConnectionSocket *s, const Anope::string &target, int port)
 {
+	s->UnsetFlag(SF_CONNECTING);
+	s->UnsetFlag(SF_CONNECTED);
 	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)
@@ -326,30 +337,51 @@ void SocketIO::Connect(ConnectionSocket *s, const Anope::string &target, int por
 		if (Anope::LastErrorCode() != EINPROGRESS)
 			s->OnError(Anope::LastError());
 		else
+		{
 			SocketEngine::MarkWritable(s);
+			s->SetFlag(SF_CONNECTING);
+		}
 	}
 	else
 	{
-		s->connected = true;
+		s->SetFlag(SF_CONNECTED);
 		s->OnConnect();
 	}
 }
 
-/** 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 SocketIO::Connected(ConnectionSocket *s)
+SocketFlag SocketIO::FinishConnect(ConnectionSocket *s)
 {
-	return s->connected == true ? 1 : -1;
+	if (s->HasFlag(SF_CONNECTED))
+		return SF_CONNECTED;
+	else if (!s->HasFlag(SF_CONNECTING))
+		throw SocketException("SocketIO::FinishConnect called for a socket not connected nor connecting?");
+
+	int optval = 0;
+	socklen_t optlen = sizeof(optval);
+	if (!getsockopt(s->GetFD(), SOL_SOCKET, SO_ERROR, reinterpret_cast<char *>(&optval), &optlen) && !optval)
+	{
+		s->SetFlag(SF_CONNECTED);
+		s->UnsetFlag(SF_CONNECTING);
+		s->OnConnect();
+		return SF_CONNECTED;
+	}
+	else
+	{
+		errno = optval;
+		s->ProcessError();
+		return SF_DEAD;
+	}
 }
 
-/** Empty constructor, used for things such as the pipe socket
+/** Empty constructor, should not be called.
  */
-Socket::Socket() : Flags<SocketFlag, 2>(SocketFlagStrings)
+Socket::Socket() : Flags<SocketFlag>(SocketFlagStrings)
 {
-	this->Type = SOCKTYPE_BASE;
-	this->IO = &normalSocketIO;
+	throw CoreException("Socket::Socket() ?");
 }
 
 /** Constructor
@@ -357,12 +389,11 @@ Socket::Socket() : Flags<SocketFlag, 2>(SocketFlagStrings)
  * @param ipv6 IPv6?
  * @param type The socket type, defaults to SOCK_STREAM
  */
-Socket::Socket(int sock, bool ipv6, int type) : Flags<SocketFlag, 2>(SocketFlagStrings)
+Socket::Socket(int sock, bool ipv6, int type) : Flags<SocketFlag>(SocketFlagStrings)
 {
-	this->Type = SOCKTYPE_BASE;
 	this->IO = &normalSocketIO;
 	this->IPv6 = ipv6;
-	if (sock == 0)
+	if (sock == -1)
 		this->Sock = socket(this->IPv6 ? AF_INET6 : AF_INET, type, 0);
 	else
 		this->Sock = sock;
@@ -432,163 +463,35 @@ void Socket::Bind(const Anope::string &ip, int port)
 	this->IO->Bind(this, ip, port);
 }
 
-/** 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.
+ * @return true to continue to call ProcessRead/ProcessWrite, false to not continue
  */
-bool Socket::ProcessRead()
+bool Socket::Process()
 {
 	return true;
 }
 
-/** Called when the socket is ready to be written to
- * @return true on success, false to drop this socket
- */
-bool Socket::ProcessWrite()
-{
-	return true;
-}
-
-/** Called when there is an error for this socket
- * @return true on success, false to drop this socket
- */
-void Socket::ProcessError()
-{
-}
-
-/** Constructor for pipe socket
- */
-BufferedSocket::BufferedSocket() : Socket()
-{
-	this->Type = SOCKTYPE_BUFFERED;
-}
-
-/** Constructor
- * @param fd FD to use
- * @param ipv6 true for ipv6
- * @param type socket type, defaults to SOCK_STREAM
- */
-BufferedSocket::BufferedSocket(int fd, bool ipv6, int type) : Socket(fd, ipv6, type)
-{
-	this->Type = SOCKTYPE_BUFFERED;
-}
-
-/** Default destructor
- */
-BufferedSocket::~BufferedSocket()
-{
-}
-
 /** Called when there is something to be received for this socket
  * @return true on success, false to drop this socket
  */
-bool BufferedSocket::ProcessRead()
+bool Socket::ProcessRead()
 {
-	char tbuffer[NET_BUFSIZE];
-
-	this->RecvLen = 0;
-
-	int len = this->IO->Recv(this, tbuffer, sizeof(tbuffer) - 1);
-	if (len <= 0)
-		return false;
-	
-	tbuffer[len] = 0;
-	this->RecvLen = len;
-
-	Anope::string sbuffer = this->extrabuf;
-	sbuffer += tbuffer;
-	this->extrabuf.clear();
-	size_t lastnewline = sbuffer.rfind('\n');
-	if (lastnewline == Anope::string::npos)
-	{
-		this->extrabuf = sbuffer;
-		return true;
-	}
-	if (lastnewline < sbuffer.length() - 1)
-	{
-		this->extrabuf = sbuffer.substr(lastnewline);
-		this->extrabuf.trim();
-		sbuffer = sbuffer.substr(0, lastnewline);
-	}
-
-	sepstream stream(sbuffer, '\n');
-
-	Anope::string tbuf;
-	while (stream.GetToken(tbuf))
-	{
-		tbuf.trim();
-		if (!tbuf.empty() && !Read(tbuf))
-			return false;
-	}
-
 	return true;
 }
 
 /** Called when the socket is ready to be written to
  * @return true on success, false to drop this socket
  */
-bool BufferedSocket::ProcessWrite()
+bool Socket::ProcessWrite()
 {
-	int count = this->IO->Send(this, this->WriteBuffer);
-	if (count <= -1)
-		return false;
-	this->WriteBuffer = this->WriteBuffer.substr(count);
-	if (this->WriteBuffer.empty())
-		SocketEngine::ClearWritable(this);
-
 	return true;
 }
 
-/** Called with a line received from the socket
- * @param buf The line
- * @return true to continue reading, false to drop the socket
- */
-bool BufferedSocket::Read(const Anope::string &buf)
-{
-	return false;
-}
-
-/** Write to the socket
- * @param message The message
- */
-void BufferedSocket::Write(const char *message, ...)
-{
-	va_list vi;
-	char tbuffer[BUFSIZE];
-
-	if (!message)
-		return;
-
-	va_start(vi, message);
-	vsnprintf(tbuffer, sizeof(tbuffer), message, vi);
-	va_end(vi);
-
-	Anope::string sbuf = tbuffer;
-	Write(sbuf);
-}
-
-/** Write to the socket
- * @param message The message
- */
-void BufferedSocket::Write(const Anope::string &message)
-{
-	this->WriteBuffer += message + "\r\n";
-	SocketEngine::MarkWritable(this);
-}
-
-/** Get the length of the read buffer
- * @return The length of the read buffer
- */
-int BufferedSocket::ReadBufferLen() const
-{
-	return RecvLen;
-}
-
-/** Get the length of the write buffer
- * @return The length of the write buffer
+/** Called when there is an error for this socket
+ * @return true on success, false to drop this socket
  */
-int BufferedSocket::WriteBufferLen() const
+void Socket::ProcessError()
 {
-	return this->WriteBuffer.length();
 }
 
 /** Constructor
@@ -596,16 +499,20 @@ int BufferedSocket::WriteBufferLen() const
  * @param port The port to listen on
  * @param ipv6 true for ipv6
  */
-ListenSocket::ListenSocket(const Anope::string &bindip, int port, bool ipv6) : Socket(0, ipv6)
+ListenSocket::ListenSocket(const Anope::string &bindip, int port, bool ipv6) : Socket(-1, ipv6)
 {
-	this->Type = SOCKTYPE_LISTEN;
 	this->SetNonBlocking();
 
+#ifndef _WIN32
+	int op = 1;
+	setsockopt(this->GetFD(), SOL_SOCKET, SO_REUSEADDR, &op, sizeof(op));
+#endif
+
 	this->bindaddr.pton(IPv6 ? AF_INET6 : AF_INET, bindip, port);
 	this->IO->Bind(this, bindip, port);
 
 	if (listen(Sock, SOMAXCONN) == -1)
-		throw SocketException(Anope::string("Unable to listen: ") + Anope::LastError());
+		throw SocketException("Unable to listen: " + Anope::LastError());
 }
 
 /** Destructor
@@ -629,148 +536,3 @@ bool ListenSocket::ProcessRead()
 	return true;
 }
 
-/** Called when a connection is accepted
- * @param fd The FD for the new connection
- * @param addr The sockaddr for where the connection came from
- * @return The new socket
- */
-ClientSocket *ListenSocket::OnAccept(int fd, const sockaddrs &addr)
-{
-	return new ClientSocket(this, fd, addr);
-}
-
-/** Constructor
- * @param ipv6 true to use IPv6
- * @param type The socket type, defaults to SOCK_STREAM
- */
-ConnectionSocket::ConnectionSocket(bool ipv6, int type) : BufferedSocket(0, ipv6, type), connected(false)
-{
-	this->Type = SOCKTYPE_CONNECTION;
-}
-
-/** Connect the socket
- * @param TargetHost The target host to connect to
- * @param Port The target port to connect to
- */
-void ConnectionSocket::Connect(const Anope::string &TargetHost, int Port)
-{
-	this->IO->Connect(this, TargetHost, Port);
-}
-
-/** Called when there is something to be received for this socket
- * @return true on success, false to drop this socket
- */
-bool ConnectionSocket::ProcessRead()
-{
-	if (!this->connected)
-	{
-		int optval = 0;
-		socklen_t optlen = sizeof(optval);
-		if (!getsockopt(this->GetFD(), SOL_SOCKET, SO_ERROR, reinterpret_cast<char *>(&optval), &optlen) && !optval)
-		{
-			this->connected = true;
-			this->OnConnect();
-		}
-		else
-			errno = optval;
-	}
-
-	int i = this->IO->Connected(this);
-	if (i == 1)
-		return BufferedSocket::ProcessRead();
-	else if (i == 0)
-		return true;
-
-	this->OnError(Anope::LastError());
-	return false;
-}
-
-/** Called when the socket is ready to be written to
- * @return true on success, false to drop this socket
- */
-bool ConnectionSocket::ProcessWrite()
-{
-	if (!this->connected)
-	{
-		int optval = 0;
-		socklen_t optlen = sizeof(optval);
-		if (!getsockopt(this->GetFD(), SOL_SOCKET, SO_ERROR, reinterpret_cast<char *>(&optval), &optlen) && !optval)
-		{
-			this->connected = true;
-			this->OnConnect();
-		}
-		else
-			errno = optval;
-	}
-
-	int i = this->IO->Connected(this);
-	if (i == 1)
-		return BufferedSocket::ProcessWrite();
-	else if (i == 0)
-		return true;
-	
-	this->OnError(Anope::LastError());
-	return false;
-}
-
-/** Called when there is an error for this socket
- * @return true on success, false to drop this socket
- */
-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() : "");
-}
-
-/** Called on a successful connect
- */
-void ConnectionSocket::OnConnect()
-{
-}
-
-/** Called when a connection is not successful
- * @param error The error
- */
-void ConnectionSocket::OnError(const Anope::string &error)
-{
-}
-
-/** Constructor
- * @param ls Listen socket this connection is from
- * @param fd New FD for this socket
- * @param addr Address the connection came from
- */
-ClientSocket::ClientSocket(ListenSocket *ls, int fd, const sockaddrs &addr) : BufferedSocket(fd, ls->IsIPv6()), LS(ls), clientaddr(addr)
-{
-	this->Type = SOCKTYPE_CLIENT;
-}
-
-/** Called when there is something to be received for this socket
- * @return true on success, false to drop this socket
- */
-bool ClientSocket::ProcessRead()
-{
-	int i = this->IO->Accepted(this);
-	if (i == 1)
-		return BufferedSocket::ProcessRead();
-	else if (i == 0)
-		return true;
-	return false;
-}
-
-/** Called when the socket is ready to be written to
- * @return true on success, false to drop this socket
- */
-bool ClientSocket::ProcessWrite()
-{
-	int i = this->IO->Accepted(this);
-	if (i == 1)
-		return BufferedSocket::ProcessWrite();
-	else if (i == 0)
-		return true;
-	return false;
-}
-