Basic relay link connection

common/CMakeLists.txt

@@ -76,6 +76,10 @@ SET(common_sources
 	net/daybreak_connection.cpp
 	net/eqstream.cpp
 	net/packet.cpp
+	net/relay.cpp
+	net/relay_link.cpp
+	net/tcp_connection.cpp
+	net/tcp_server.cpp
 	patches/patches.cpp
 	patches/sod.cpp
 	patches/sod_limits.cpp
@@ -105,6 +109,7 @@ SET(common_sources
 	tinyxml/tinyxml.cpp
 	tinyxml/tinyxmlerror.cpp
 	tinyxml/tinyxmlparser.cpp
+	util/uuid.cpp
 )
 
 SET(common_headers
@@ -213,9 +218,14 @@ SET(common_headers
 	net/crc32.h
 	net/daybreak_connection.h
 	net/daybreak_structs.h
+	net/dns.h
 	net/endian.h
 	net/eqstream.h
 	net/packet.h
+	net/relay.h
+	net/relay_link.h
+	net/tcp_connection.h
+	net/tcp_server.h
 	patches/patches.h
 	patches/sod.h
 #	patches/sod_itemfields.h
@@ -268,6 +278,7 @@ SET(common_headers
 	tinyxml/tinystr.h
 	tinyxml/tinyxml.h
 	util/memory_stream.h
+	util/uuid.h
 )
 
 SOURCE_GROUP(Event FILES
@@ -282,11 +293,22 @@ SOURCE_GROUP(Net FILES
 	net/daybreak_connection.cpp
 	net/daybreak_connection.h
 	net/daybreak_structs.h
+	net/dns.h
 	net/endian.h
+	net/eqmq.cpp
+	net/eqmq.h
 	net/eqstream.h
 	net/eqstream.cpp
 	net/packet.cpp
 	net/packet.h
+	net/relay.cpp
+	net/relay.h
+	net/relay_link.cpp
+	net/relay_link.h
+	net/tcp_connection.cpp
+	net/tcp_connection.h
+	net/tcp_server.cpp
+	net/tcp_server.h
 )
 
 SOURCE_GROUP(Patches FILES
@@ -383,6 +405,8 @@ SOURCE_GROUP(TinyXML FILES
 
 SOURCE_GROUP(Util FILES
 	util/memory_stream.h
+	util/uuid.cpp
+	util/uuid.h
 )
 
 INCLUDE_DIRECTORIES(Patches SocketLib StackWalker TinyXML)

common/emu_tcp_connection.cpp

@@ -283,6 +283,7 @@ bool EmuTCPConnection::LineOutQueuePush(char* line) {
 		}
 	#endif
 	if(line[0] == '*') {
+		printf("LineOutQueuePush %s\n", line);
 		if (strcmp(line, "**PACKETMODE**") == 0) {
 			MSendQueue.lock();
 			safe_delete_array(sendbuf);

common/event/timer.h

@@ -6,58 +6,62 @@ namespace EQ {
 	class Timer
 	{
 	public:
-		Timer(uint64_t duration_ms, bool repeats, std::function<void(void)> cb)
+		Timer(std::function<void(Timer *)> cb)
 		{
-			memset(&m_timer, 0, sizeof(uv_timer_t));
+			m_timer = nullptr;
 			m_cb = cb;
-			m_duration_ms = duration_ms;
-			m_repeats = repeats;
-			m_attached = nullptr;
-			Attach(EventLoop::Get().Handle());
 		}
 
+		Timer(uint64_t duration_ms, bool repeats, std::function<void(Timer *)> cb)
+		{
+			m_timer = nullptr;
+			m_cb = cb;
+			Start(duration_ms, repeats);
+		}
+	
 		~Timer()
 		{
-			Detach();
+			Stop();
+		}
+
+		void Start(uint64_t duration_ms, bool repeats) {
+			auto loop = EventLoop::Get().Handle();
+			if (!m_timer) {
+				m_timer = new uv_timer_t;
+				memset(m_timer, 0, sizeof(uv_timer_t));
+				uv_timer_init(loop, m_timer);
+				m_timer->data = this;
+
+				if (repeats) {
+					uv_timer_start(m_timer, [](uv_timer_t *handle) {
+						Timer *t = (Timer*)handle->data;
+						t->Execute();
+					}, duration_ms, duration_ms);
+				}
+				else {
+					uv_timer_start(m_timer, [](uv_timer_t *handle) {
+						Timer *t = (Timer*)handle->data;
+						t->Stop();
+						t->Execute();
+					}, duration_ms, 0);
+				}
+			}
+		}
+
+		void Stop() {
+			if (m_timer) {
+				uv_close((uv_handle_t*)m_timer, [](uv_handle_t* handle) {
+					delete handle;
+				});
+				m_timer = nullptr;
+			}
 		}
 	private:
 		void Execute() {
-			m_cb();
+			m_cb(this);
 		}
-
-		virtual void Attach(uv_loop_t *loop) {
-			if (!m_attached) {
-				uv_timer_init(loop, &m_timer);
-				m_timer.data = this;
-
-				if (m_repeats) {
-					uv_timer_start(&m_timer, [](uv_timer_t *handle) {
-						Timer *t = (Timer*)handle->data;
-						t->Execute();
-					}, m_duration_ms, m_duration_ms);
-				}
-				else {
-					uv_timer_start(&m_timer, [](uv_timer_t *handle) {
-						Timer *t = (Timer*)handle->data;
-						t->Execute();
-					}, m_duration_ms, 0);
-				}
-
-				m_attached = loop;
-			}
-		}
-
-		virtual void Detach() {
-			if (m_attached) {
-				uv_timer_stop(&m_timer);
-				m_attached = nullptr;
-			}
-		}
-
-		uv_timer_t m_timer;
-		std::function<void(void)> m_cb;
-		uint64_t m_duration_ms;
-		bool m_repeats;
-		uv_loop_t *m_attached;
+	
+		uv_timer_t *m_timer;
+		std::function<void(Timer*)> m_cb;
 	};
 }

common/net/dns.h

@@ -0,0 +1,54 @@
+#pragma once
+
+#include "../event/event_loop.h"
+#include <string>
+#include <functional>
+
+namespace EQ
+{
+	namespace Net
+	{
+		static void DNSLookup(const std::string &addr, int port, bool ipv6, std::function<void(const std::string&)> cb) {
+			struct DNSBaton
+			{
+				std::function<void(const std::string&)> cb;
+				bool ipv6;
+			};
+
+			auto loop = EQ::EventLoop::Get().Handle();
+			uv_getaddrinfo_t *resolver = new uv_getaddrinfo_t();
+			memset(resolver, 0, sizeof(uv_getaddrinfo_t));
+			auto port_str = std::to_string(port);
+			DNSBaton *baton = new DNSBaton();
+			baton->cb = cb;
+			baton->ipv6 = ipv6;
+			resolver->data = baton;
+
+			uv_getaddrinfo(loop, resolver, [](uv_getaddrinfo_t* req, int status, addrinfo* res) {
+				DNSBaton *baton = (DNSBaton*)req->data;
+				if (status < 0) {
+					auto cb = baton->cb;
+					delete baton;
+					delete req;
+					cb("");
+					return;
+				}
+
+				char addr[40] = { 0 };
+
+				if (baton->ipv6) {
+					uv_ip6_name((struct sockaddr_in6*)res->ai_addr, addr, 40);
+				}
+				else {
+					uv_ip4_name((struct sockaddr_in*)res->ai_addr, addr, 40);
+				}
+
+				auto cb = baton->cb;
+				delete baton;
+				delete req;
+				uv_freeaddrinfo(res);
+				cb(addr);
+			}, addr.c_str(), port_str.c_str(), nullptr);
+		}
+	}
+}

common/net/packet.cpp

@@ -255,7 +255,7 @@ std::string EQ::Net::Packet::GetString(size_t offset, size_t length) const
 		throw std::out_of_range("Packet read out of range.");
 	}
 
-	return std::string((char*)Data(), (char*)Data() + length);
+	return std::string((char*)Data() + offset, (char*)Data() + offset + length);
 }
 
 std::string EQ::Net::Packet::GetCString(size_t offset) const

common/net/relay_link.cpp

@@ -0,0 +1,100 @@
+#include "relay_link.h"
+#include "dns.h"
+#include "../eqemu_logsys.h"
+#include <algorithm>
+
+EQ::Net::RelayLink::RelayLink(const std::string &addr, int port, const std::string &identifier)
+	: m_timer(std::unique_ptr<EQ::Timer>(new EQ::Timer(250, true, std::bind(&EQ::Net::RelayLink::Connect, this))))
+{
+	m_established = false;
+	m_connecting = false;
+	m_port = port;
+	m_identifier = identifier;
+	DNSLookup(addr, port, false, [this](const std::string &address) {
+		m_addr = address;
+	});
+}
+
+EQ::Net::RelayLink::~RelayLink()
+{
+}
+
+void EQ::Net::RelayLink::Connect()
+{
+	if (m_addr.length() == 0 || m_port == 0 || m_connection || m_connecting) {
+		return;
+	}
+
+	m_connecting = true;
+	EQ::Net::TCPConnection::Connect(m_addr, m_port, false, [this](std::shared_ptr<EQ::Net::TCPConnection> connection) {
+		if (connection == nullptr) {
+			Log.OutF(Logs::General, Logs::Debug, "Error connecting to {0}:{1}, attempting to reconnect...", m_addr, m_port);
+			m_connecting = false;
+			return;
+		}
+
+		Log.OutF(Logs::General, Logs::Debug, "Connected to {0}:{1}", m_addr, m_port);
+		m_connection = connection;
+		m_connection->OnDisconnect([this](EQ::Net::TCPConnection *c) {
+			Log.OutF(Logs::General, Logs::Debug, "Connection lost to {0}:{1}, attempting to reconnect...", m_addr, m_port);
+			m_established = false;
+			m_connection.reset();
+		});
+
+		m_connection->OnRead(std::bind(&EQ::Net::RelayLink::ProcessData, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
+		m_connection->Start();
+
+		SendIdentifier();
+		m_connecting = false;
+	});
+}
+
+void EQ::Net::RelayLink::ProcessData(EQ::Net::TCPConnection *c, const unsigned char *data, size_t length)
+{
+	EQ::Net::ReadOnlyPacket p((void*)data, length);
+	try {
+		Log.OutF(Logs::General, Logs::Debug, "Process data:\n{0}", p.ToString());
+
+		if (m_established) {
+			//process raw packet
+		}
+		else {
+			auto msg = fmt::format("**PACKETMODE{0}**", m_identifier);
+			std::string cmp_msg;
+			if (p.GetInt8(0) == '*') {
+				cmp_msg = p.GetString(0, msg.length());
+			}
+			else if (p.GetInt8(1) == '*') {
+				cmp_msg = p.GetString(1, msg.length());
+			}
+			else {
+				return;
+			}
+
+			if (cmp_msg.compare(msg) == 0) {
+				m_established = true;
+				Log.OutF(Logs::General, Logs::Debug, "Established connection of type {0}", m_identifier);
+			}
+		}
+	}
+	catch (std::exception &ex) {
+		Log.OutF(Logs::General, Logs::Debug, "Error parsing relay link packet: {0}", ex.what());
+	}
+}
+
+void EQ::Net::RelayLink::SendIdentifier()
+{
+	auto msg = fmt::format("**PACKETMODE{0}**\r", m_identifier);
+	EQ::Net::WritablePacket packet;
+	packet.PutData(0, (void*)msg.c_str(), msg.length());
+	SendInternal(packet);
+}
+
+void EQ::Net::RelayLink::SendInternal(const EQ::Net::Packet &p)
+{
+	if (m_connection == nullptr) {
+		return;
+	}
+
+	m_connection->Write((const char*)p.Data(), p.Length());
+}

common/net/relay_link.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include "tcp_server.h"
+#include "packet.h"
+#include "../event/timer.h"
+#include "../event/event_loop.h"
+
+namespace EQ
+{
+	namespace Net {
+		class RelayLink
+		{
+		public:
+			RelayLink(const std::string &addr, int port, const std::string &identifier);
+			~RelayLink();
+
+		private:
+			void Connect();
+			void ProcessData(EQ::Net::TCPConnection *c, const unsigned char *data, size_t length);
+			void SendIdentifier();
+			void SendInternal(const EQ::Net::Packet &p);
+
+			std::unique_ptr<EQ::Timer> m_timer;
+			std::string m_addr; 
+			std::string m_identifier;
+			int m_port;
+			std::shared_ptr<EQ::Net::TCPConnection> m_connection;
+			bool m_established;
+			bool m_connecting;
+		};
+	}
+}

common/net/tcp_connection.cpp

@@ -0,0 +1,175 @@
+#include "tcp_connection.h"
+#include "../event/event_loop.h"
+
+void on_close_handle(uv_handle_t* handle) {
+	delete handle;
+}
+
+EQ::Net::TCPConnection::TCPConnection(uv_tcp_t *socket)
+{
+	m_socket = socket;
+	m_socket->data = this;
+}
+
+EQ::Net::TCPConnection::~TCPConnection() {
+	Disconnect();
+}
+
+void EQ::Net::TCPConnection::Connect(const std::string &addr, int port, bool ipv6, std::function<void(std::shared_ptr<TCPConnection>)> cb)
+{
+	struct EQTCPConnectBaton
+	{
+		uv_tcp_t *socket;
+		std::function<void(std::shared_ptr<EQ::Net::TCPConnection>)> cb;
+	};
+
+	auto loop = EQ::EventLoop::Get().Handle();
+	uv_tcp_t *socket = new uv_tcp_t;
+	memset(socket, 0, sizeof(uv_tcp_t));
+	uv_tcp_init(loop, socket);
+
+	if (ipv6) {
+		sockaddr_in6 iaddr;
+		uv_ip6_addr(addr.c_str(), port, &iaddr);
+
+		uv_connect_t *connect = new uv_connect_t;
+		memset(connect, 0, sizeof(uv_connect_t));
+
+		EQTCPConnectBaton *baton = new EQTCPConnectBaton;
+		baton->cb = cb;
+		baton->socket = socket;
+		connect->data = baton;
+		uv_tcp_connect(connect, socket, (sockaddr*)&iaddr,
+			[](uv_connect_t* req, int status) {
+			EQTCPConnectBaton *baton = (EQTCPConnectBaton*)req->data;
+			auto socket = baton->socket;
+			auto cb = baton->cb;
+
+			delete baton;
+
+			if (status < 0) {
+				uv_close((uv_handle_t*)socket, on_close_handle);
+				delete req;
+				cb(nullptr);
+			}
+			else {
+				delete req;
+				std::shared_ptr<EQ::Net::TCPConnection> connection(new EQ::Net::TCPConnection(socket));
+				cb(connection);
+			}
+		});
+	}
+	else {
+		sockaddr_in iaddr;
+		uv_ip4_addr(addr.c_str(), port, &iaddr);
+
+		uv_connect_t *connect = new uv_connect_t;
+		memset(connect, 0, sizeof(uv_connect_t));
+
+		EQTCPConnectBaton *baton = new EQTCPConnectBaton;
+		baton->cb = cb;
+		baton->socket = socket;
+		connect->data = baton;
+		uv_tcp_connect(connect, socket, (sockaddr*)&iaddr,
+			[](uv_connect_t* req, int status) {
+			EQTCPConnectBaton *baton = (EQTCPConnectBaton*)req->data;
+			auto socket = baton->socket;
+			auto cb = baton->cb;
+
+			delete baton;
+
+			if (status < 0) {
+				uv_close((uv_handle_t*)socket, on_close_handle);
+				delete req;
+				cb(nullptr);
+			}
+			else {
+				delete req;
+				std::shared_ptr<EQ::Net::TCPConnection> connection(new EQ::Net::TCPConnection(socket));
+				cb(connection);
+			}
+		});
+	}
+}
+
+void EQ::Net::TCPConnection::Start() {
+	uv_read_start((uv_stream_t*)m_socket, [](uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
+		buf->base = new char[suggested_size];
+		buf->len = suggested_size;
+	}, [](uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) {
+
+		TCPConnection *connection = (TCPConnection*)stream->data;
+
+		if (nread > 0) {
+			connection->Read(buf->base, nread);
+
+			if (buf->base) {
+				delete[] buf->base;
+			}
+		}
+		else if (nread == UV_EOF) {
+			if (buf->base) {
+				delete[] buf->base;
+			}
+		}
+		else if (nread < 0) {
+			connection->Disconnect();
+
+			if (buf->base) {
+				delete[] buf->base;
+			}
+		}
+	});
+}
+
+void EQ::Net::TCPConnection::OnRead(std::function<void(TCPConnection*, const unsigned char*, size_t)> cb)
+{
+	m_on_read_cb = cb;
+}
+
+void EQ::Net::TCPConnection::OnDisconnect(std::function<void(TCPConnection*)> cb)
+{
+	m_on_disconnect_cb = cb;
+}
+
+void EQ::Net::TCPConnection::Disconnect()
+{
+	if (m_socket) {
+		uv_close((uv_handle_t*)m_socket, on_close_handle);
+		m_socket = nullptr;
+	
+		if (m_on_disconnect_cb) {
+			m_on_disconnect_cb(this);
+		}
+	}
+}
+
+void EQ::Net::TCPConnection::Read(const char *data, size_t count)
+{
+	if (m_on_read_cb) {
+		m_on_read_cb(this, (unsigned char*)data, count);
+	}
+}
+
+void EQ::Net::TCPConnection::Write(const char *data, size_t count)
+{
+	if (!m_socket) {
+		return;
+	}
+
+	uv_write_t *write_req = new uv_write_t;
+	memset(write_req, 0, sizeof(uv_write_t));
+	write_req->data = this;
+	uv_buf_t send_buffers[1];
+	send_buffers[0].base = (char*)data;
+	send_buffers[0].len = count;
+
+	uv_write(write_req, (uv_stream_t*)m_socket, send_buffers, 1, [](uv_write_t* req, int status) {
+		EQ::Net::TCPConnection *connection = (EQ::Net::TCPConnection*)req->data;
+		delete req;
+
+		if (status < 0) {
+			connection->Disconnect();
+		}
+	});
+}

common/net/tcp_connection.h

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <uv.h>
+#include <functional>
+#include <string>
+#include <memory>
+#include <evt_tls.h>
+
+namespace EQ
+{
+	namespace Net
+	{
+		class TCPConnection
+		{
+		public:
+			TCPConnection(uv_tcp_t *socket);
+			~TCPConnection();
+
+			static void Connect(const std::string &addr, int port, bool ipv6, std::function<void(std::shared_ptr<TCPConnection>)> cb);
+			
+			void Start();
+			void OnRead(std::function<void(TCPConnection*, const unsigned char *, size_t)> cb);
+			void OnDisconnect(std::function<void(TCPConnection*)> cb);
+			void Disconnect();
+			void Read(const char *data, size_t count);
+			void Write(const char *data, size_t count);
+		private:
+			TCPConnection();
+
+			uv_tcp_t *m_socket;
+			std::function<void(TCPConnection*, const unsigned char *, size_t)> m_on_read_cb;
+			std::function<void(TCPConnection*)> m_on_disconnect_cb;
+		};
+	}
+}

common/net/tcp_server.cpp

@@ -0,0 +1,75 @@
+#include "tcp_server.h"
+#include "../event/event_loop.h"
+
+void on_close_tcp_server_handle(uv_handle_t* handle) {
+	delete handle;
+}
+
+EQ::Net::TCPServer::TCPServer()
+{
+	m_socket = nullptr;
+}
+
+EQ::Net::TCPServer::~TCPServer() {
+	Close();
+}
+
+void EQ::Net::TCPServer::Listen(int port, bool ipv6, std::function<void(std::shared_ptr<TCPConnection>)> cb)
+{
+	if (m_socket) {
+		return;
+	}
+
+	m_on_new_connection = cb;
+
+	auto loop = EQ::EventLoop::Get().Handle();
+	m_socket = new uv_tcp_t;
+	memset(m_socket, 0, sizeof(uv_tcp_t));
+	uv_tcp_init(loop, m_socket);
+
+	sockaddr iaddr;
+	if (ipv6) {
+		uv_ip6_addr("::", port, (sockaddr_in6*)&iaddr);
+	}
+	else {
+		uv_ip4_addr("0.0.0.0", port, (sockaddr_in*)&iaddr);
+	}
+
+	m_socket->data = this;
+	uv_tcp_bind(m_socket, &iaddr, 0);
+
+	uv_listen((uv_stream_t*)m_socket, 128, [](uv_stream_t* server, int status) {
+		if (status < 0) {
+			return;
+		}
+
+		auto loop = EQ::EventLoop::Get().Handle();
+		uv_tcp_t *client = new uv_tcp_t;
+		memset(client, 0, sizeof(uv_tcp_t));
+		uv_tcp_init(loop, client);
+
+		if (uv_accept(server, (uv_stream_t*)client) < 0) {
+			delete client;
+			return;
+		}
+
+		EQ::Net::TCPServer *s = (EQ::Net::TCPServer*)server->data;
+		s->AddClient(client);
+	});
+}
+
+void EQ::Net::TCPServer::Close()
+{
+	if (m_socket) {
+		uv_close((uv_handle_t*)m_socket, on_close_tcp_server_handle);
+		m_socket = nullptr;
+	}
+}
+
+void EQ::Net::TCPServer::AddClient(uv_tcp_t *c)
+{
+	std::shared_ptr<TCPConnection> client(new TCPConnection(c));
+	if (m_on_new_connection) {
+		m_on_new_connection(client);
+	}
+}

common/net/tcp_server.h

@@ -0,0 +1,24 @@
+#pragma once
+
+#include "tcp_connection.h"
+
+namespace EQ
+{
+	namespace Net
+	{
+		class TCPServer
+		{
+		public:
+			TCPServer();
+			~TCPServer();
+
+			void Listen(int port, bool ipv6, std::function<void(std::shared_ptr<TCPConnection>)> cb);
+			void Close();
+			void AddClient(uv_tcp_t *c);
+
+		private:
+			std::function<void(std::shared_ptr<TCPConnection>)> m_on_new_connection;
+			uv_tcp_t *m_socket;
+		};
+	}
+}

common/util/uuid.cpp

@@ -1,6 +1,6 @@
 #include "uuid.h"
 
-#include <core/logsys.h>
+#include <fmt/format.h>
 
 #ifdef _WIN32
 #include <objbase.h>