sockets

Author: 1fbff5f83b23d39d38b1dfcb4cac8d9b

byond-extools/src/core/byond_structures.cpp

@@ -143,6 +143,21 @@ List::List(Value v)
 	list = GetListPointerById(id);
 }
 
+Container::Container(char type, int id) : type(type), id(id)
+{
+	IncRefCount(type, id);
+}
+
+Container::Container(Value val) : type(val.type), id(val.value)
+{
+	IncRefCount(type, id);
+}
+
+Container::~Container()
+{
+	DecRefCount(type, id);
+}
+
 unsigned int Container::length()
 {
 	return Length(type, id);

byond-extools/src/core/byond_structures.h

@@ -175,8 +175,9 @@ struct ContainerProxy
 struct Container //All kinds of lists, including magical snowflake lists like contents
 { 
 
-	Container(char type, int id) : type(type), id(id) {}
-	Container(Value val) : type(val.type), id(val.value) {}
+	Container(char type, int id);
+	Container(Value val);
+	~Container();
 	char type;
 	int id;
 

byond-extools/src/core/socket/socket.cpp

@@ -58,27 +58,21 @@ bool Socket::create(int family, int socktype, int protocol)
 	return raw_socket != INVALID_SOCKET;
 }
 
-// Listening.
-bool TcpListener::listen(const char* port, const char* iface)
+bool connect_socket(Socket& socket, const char* port, const char* remote)
 {
+
 	struct addrinfo* result = NULL;
 	struct addrinfo hints;
 	int iResult;
 
-	// Initialize Winsock
-	if (!InitOnce())
-	{
-		return false;
-	}
-
 	ZeroMemory(&hints, sizeof(hints));
 	hints.ai_family = AF_INET;
 	hints.ai_socktype = SOCK_STREAM;
 	hints.ai_protocol = IPPROTO_TCP;
 	hints.ai_flags = AI_PASSIVE;
 
 	// Resolve the server address and port
-	iResult = getaddrinfo(iface, port, &hints, &result);
+	iResult = getaddrinfo(remote, port, &hints, &result);
 	if (iResult != 0)
 	{
 		Core::Alert("getaddrinfo failed with error: " + std::to_string(iResult));
@@ -93,40 +87,21 @@ bool TcpListener::listen(const char* port, const char* iface)
 		return false;
 	}
 
-	// Set SO_REUSEADDR so if the process is killed, the port becomes reusable
-	// immediately rather than after a 60-second delay.
-	int opt = 1;
-	setsockopt(socket.raw(), SOL_SOCKET, SO_REUSEADDR, (const char*) &opt, sizeof(int));
-
 	// Setup the TCP listening socket
-	iResult = bind(socket.raw(), result->ai_addr, (int)result->ai_addrlen);
+	iResult = ::connect(socket.raw(), result->ai_addr, (int)result->ai_addrlen);
 	if (iResult == SOCKET_ERROR)
 	{
-		Core::Alert("bind failed with error: " + std::to_string(WSAGetLastError()));
+		Core::Alert("connect failed with error: " + std::to_string(WSAGetLastError()));
 		freeaddrinfo(result);
 		socket.close();
 		return false;
 	}
 
-	freeaddrinfo(result);
-
-	iResult = ::listen(socket.raw(), SOMAXCONN);
-	if (iResult == SOCKET_ERROR)
-	{
-		Core::Alert("listen failed with error: " + std::to_string(WSAGetLastError()));
-		socket.close();
-		return false;
-	}
-
 	return true;
 }
 
-JsonStream TcpListener::accept()
-{
-	return JsonStream(Socket(::accept(socket.raw(), NULL, NULL)));
-}
-
-bool JsonStream::connect(const char* port, const char* remote)
+// Listening.
+bool JsonListener::listen(const char* port, const char* iface)
 {
 	struct addrinfo* result = NULL;
 	struct addrinfo hints;
@@ -145,7 +120,7 @@ bool JsonStream::connect(const char* port, const char* remote)
 	hints.ai_flags = AI_PASSIVE;
 
 	// Resolve the server address and port
-	iResult = getaddrinfo(remote, port, &hints, &result);
+	iResult = getaddrinfo(iface, port, &hints, &result);
 	if (iResult != 0)
 	{
 		Core::Alert("getaddrinfo failed with error: " + std::to_string(iResult));
@@ -160,19 +135,50 @@ bool JsonStream::connect(const char* port, const char* remote)
 		return false;
 	}
 
+	// Set SO_REUSEADDR so if the process is killed, the port becomes reusable
+	// immediately rather than after a 60-second delay.
+	int opt = 1;
+	setsockopt(socket.raw(), SOL_SOCKET, SO_REUSEADDR, (const char*) &opt, sizeof(int));
+
 	// Setup the TCP listening socket
-	iResult = ::connect(socket.raw(), result->ai_addr, (int)result->ai_addrlen);
+	iResult = bind(socket.raw(), result->ai_addr, (int)result->ai_addrlen);
 	if (iResult == SOCKET_ERROR)
 	{
-		Core::Alert("connect failed with error: " + std::to_string(WSAGetLastError()));
+		Core::Alert("bind failed with error: " + std::to_string(WSAGetLastError()));
 		freeaddrinfo(result);
 		socket.close();
 		return false;
 	}
 
+	freeaddrinfo(result);
+
+	iResult = ::listen(socket.raw(), SOMAXCONN);
+	if (iResult == SOCKET_ERROR)
+	{
+		Core::Alert("listen failed with error: " + std::to_string(WSAGetLastError()));
+		socket.close();
+		return false;
+	}
+
 	return true;
 }
 
+JsonStream JsonListener::accept()
+{
+	return JsonStream(Socket(::accept(socket.raw(), NULL, NULL)));
+}
+
+bool JsonStream::connect(const char* port, const char* remote)
+{
+	// Initialize Winsock
+	if (!InitOnce())
+	{
+		return false;
+	}
+
+	return connect_socket(socket, port, remote);
+}
+
 bool JsonStream::send(const char* type, nlohmann::json content)
 {
 	nlohmann::json j = {
@@ -218,3 +224,38 @@ nlohmann::json JsonStream::recv_message()
 		recv_buffer.append(data.begin(), data.begin() + received_bytes);
 	}
 }
+
+bool TcpStream::connect(const char* port, const char* remote)
+{
+	if (!InitOnce())
+	{
+		return false;
+	}
+
+	return connect_socket(socket, port, remote);
+}
+
+std::string TcpStream::recv()
+{
+	std::vector<char> data(1024);
+	int received_bytes = ::recv(socket.raw(), data.data(), data.size(), 0);
+	if (received_bytes <= 0)
+	{
+		return "";
+	}
+	return std::string(data.begin(), data.begin() + received_bytes);
+}
+
+bool TcpStream::send(std::string data)
+{
+	while (!data.empty())
+	{
+		int sent_bytes = ::send(socket.raw(), data.c_str(), data.size(), 0);
+		if (sent_bytes == SOCKET_ERROR)
+		{
+			return false;
+		}
+		data.erase(data.begin(), data.begin() + sent_bytes);
+	}
+	return true;
+}

byond-extools/src/core/socket/socket.h

@@ -49,12 +49,22 @@ class JsonStream
 	void close() { socket.close(); }
 };
 
-class TcpListener
+class JsonListener
 {
 	Socket socket;
 public:
-	TcpListener() {}
+	JsonListener() {}
 	bool listen(const char* port = DBG_DEFAULT_PORT, const char* iface = "127.0.0.1");
 	JsonStream accept();
 	void close() { socket.close(); }
 };
+
+class TcpStream
+{
+	Socket socket;
+public:
+	bool connect(const char* port, const char* remote); //augh, why port first?! damn it spaceman
+	bool send(std::string data);
+	std::string recv();
+	void close() { socket.close(); }
+};

byond-extools/src/datum_socket/datum_socket.cpp

@@ -0,0 +1,143 @@
+#include "datum_socket.h"
+#include "../core/core.h"
+#include "../core/proc_management.h"
+
+std::unordered_map<unsigned int, std::unique_ptr<DatumSocket>> sockets;
+unsigned int recv_sleep_opcode = -1;
+
+DatumSocket::DatumSocket()
+{
+}
+
+DatumSocket::DatumSocket(const DatumSocket& other)
+{
+}
+
+DatumSocket::~DatumSocket()
+{
+	close();
+}
+
+bool DatumSocket::connect(std::string addr, std::string port)
+{
+	stream = TcpStream();
+	bool connected = stream.connect(port.c_str(), addr.c_str());
+	if (connected)
+	{
+		std::thread(&DatumSocket::recv_loop, this).detach();
+	}
+	return connected;
+}
+
+bool DatumSocket::send(std::string data)
+{
+	return stream.send(data);
+}
+
+std::string DatumSocket::recv(int len)
+{
+	std::lock_guard<std::mutex> lk(buffer_lock);
+	size_t nom = min(len, buffer.size());
+	std::string sub = buffer.substr(0, nom);
+	buffer.erase(buffer.begin(), buffer.begin() + nom);
+	return sub;
+}
+
+void DatumSocket::close()
+{
+	stream.close();
+	open = false;
+}
+
+void DatumSocket::recv_loop()
+{
+	while (open)
+	{
+		std::string data = stream.recv();
+		if (data.empty())
+		{
+			close();
+			return;
+		}
+		buffer_lock.lock();
+		buffer += data;
+		buffer_lock.unlock();
+		if (data_awaiter)
+		{
+			data_awaiter->time_to_resume = 0;
+			data_awaiter = nullptr;
+		}
+	}
+}
+
+trvh register_socket(unsigned int args_len, Value* args, Value src)
+{
+	//Core::Alert("register");
+	sockets[src.value] = std::make_unique<DatumSocket>();
+	return Value::Null();
+}
+
+trvh connect_socket(unsigned int args_len, Value* args, Value src)
+{
+	//Core::Alert("connect");
+	return sockets[src.value]->connect(args[0], std::to_string((int)args[1].valuef)) ? Value::True() : Value::False();
+}
+
+trvh send_socket(unsigned int args_len, Value* args, Value src)
+{
+	//Core::Alert("send");
+	return sockets[src.value]->send(args[0]) ? Value::True() : Value::False();
+}
+
+trvh check_socket(unsigned int args_len, Value* args, Value src)
+{
+	//Core::Alert("check");
+	return sockets[src.value]->has_data() ? Value::True() : Value::False();
+}
+
+trvh retrieve_socket(unsigned int args_len, Value* args, Value src)
+{
+	//Core::Alert("retrieve");
+	return Value(sockets[src.value]->recv(1024));
+}
+
+trvh deregister_socket(unsigned int args_len, Value* args, Value src)
+{
+	if (sockets.find(src.value) == sockets.end())
+	{
+		return Value::Null();
+	}
+	sockets[src.value].reset();
+	sockets.erase(src.value);
+	return Value::Null();
+}
+
+void recv_suspend(ExecutionContext* ctx)
+{
+	ctx->current_opcode++;
+	SuspendedProc* proc = Suspend(ctx, 0);
+	proc->time_to_resume = 0x7FFFFF;
+	StartTiming(proc);
+	int datum_id = ctx->constants->src.value;
+	sockets[datum_id]->set_awaiter(proc);
+	ctx->current_opcode--;
+}
+
+bool enable_sockets()
+{
+	Core::get_proc("/datum/socket/proc/__register_socket").hook(register_socket);
+	Core::get_proc("/datum/socket/proc/__check_has_data").hook(check_socket);
+	Core::get_proc("/datum/socket/proc/__retrieve_data").hook(retrieve_socket);
+	Core::get_proc("/datum/socket/proc/connect").hook(connect_socket);
+	Core::get_proc("/datum/socket/proc/send").hook(send_socket);
+	Core::get_proc("/datum/socket/proc/__deregister_socket").hook(deregister_socket);
+	recv_sleep_opcode = Core::register_opcode("RECV_SLEEP", recv_suspend);
+	Core::get_proc("/datum/socket/proc/__wait_for_data").set_bytecode(new std::vector<std::uint32_t>({ recv_sleep_opcode, 0, 0, 0 }));
+	return true;
+}
+
+extern "C" __declspec(dllexport) const char* init_sockets(int a, const char** b)
+{
+	enable_sockets();
+	return "ok";
+}