UDP client server demo

Engine/CMakeLists.txt

@@ -11,8 +11,11 @@ set(ENGINE_SOURCE_FILES
 	Systems/Renderer/Shader/Shader.cpp
 	Systems/Renderer/OGL/Renderer.cpp
 	Components/Mesh3D.cpp
+	Systems/Networks/NetworkServerService.cpp
+	Systems/Networks/NetworkClientService.cpp
 	# Renderer/CPU/RayTracer.cpp
 	Utils/AssetManager.cpp
+	Utils/Networks/WindowsSocket.cpp
 )
 
 add_library(Engine ${ENGINE_SOURCE_FILES})
@@ -23,6 +26,10 @@ target_link_libraries(Engine PRIVATE glfw)
 target_link_libraries(Engine PRIVATE assimp::assimp)
 target_include_directories(Engine PRIVATE ${Stb_INCLUDE_DIR})
 
+if (WIN32)
+	target_link_libraries(Engine PUBLIC Ws2_32)
+endif()
+
 if (OpenMP_CXX_FOUND)
 	target_link_libraries(Engine PRIVATE OpenMP::OpenMP_CXX)
 endif()

Engine/Systems/Networks/NetworkClientService.cpp

@@ -0,0 +1,135 @@
+#include "NetworkClientService.hpp"
+#include <chrono>
+#include <iostream>
+#include <mutex>
+#include <stdexcept>
+#include <thread>
+
+#ifdef _WIN32
+#include "../../Utils/Networks/WindowsSocket.hpp"
+#endif
+
+namespace Bored::Net {
+
+Client::Client(std::string compatible_id) {
+  initSocket();
+  compatible_id_ = compatible_id;
+};
+
+Client::~Client() { Disconnect(); };
+
+bool Client::Connect(int port, std::string addr, int server_port) {
+  if (running_.load())
+    return false;
+
+  running_ = true;
+
+  try {
+
+    sock_->Open(IPv4, Datagram, UDP);
+    sock_->Bind("", port);
+  } catch (const std::runtime_error &err) {
+    std::cout << "Unable to connect: " << err.what() << std::endl;
+    running_ = false;
+  }
+
+  server_addr = Conn(addr, server_port);
+  sock_->SendTo(addr, server_port, compatible_id_ + ":connect_req");
+
+  bool success = false;
+  auto start_time = std::chrono::high_resolution_clock().now();
+  while (std::chrono::duration_cast<std::chrono::milliseconds>(
+             std::chrono::high_resolution_clock::now() - start_time)
+             .count() < kConnectTimeOut * 1000) {
+    if (!sock_->HasReadable(100))
+      continue;
+    try {
+      std::string ack =
+          sock_->ReceiveFrom(server_addr.address, server_addr.port);
+      if (ack == compatible_id_) {
+        success = true;
+      }
+    } catch (const std::runtime_error &e) {
+      continue;
+    }
+  }
+
+  if (!success) {
+    running_ = false;
+    sock_->Close();
+    std::cout << "Time out while connecting with server" << std::endl;
+    return false;
+  }
+
+  listener_ = std::thread(&Client::listenLoop, this, port);
+  return true;
+};
+
+void Client::Disconnect() {
+  if (!running_.load()) {
+    return;
+  }
+
+  running_ = false;
+  sock_->Close();
+
+  if (listener_.joinable())
+    listener_.join();
+};
+
+void Client::initSocket() {
+#ifdef _WIN32
+  sock_ = std::make_shared<WindowsSocket>();
+#else
+  throw std::runtime_error("No implementation for other platform yet");
+#endif
+};
+
+void Client::listenLoop(int port) {
+  while (running_.load()) {
+    bool have_msg = false;
+#ifdef _WIN32
+    auto *ws = dynamic_cast<WindowsSocket *>(sock_.get());
+    if (ws->HasReadable(100)) {
+      have_msg = true;
+    }
+#endif
+
+    if (!have_msg) {
+      continue;
+    };
+
+    std::string from, payload;
+    int port;
+    try {
+      payload = sock_->ReceiveFrom(from, port);
+      if (Conn(from, port) != server_addr) {
+        continue;
+      };
+    } catch (const std::runtime_error &err) {
+      continue;
+    }
+
+    {
+      std::lock_guard<std::mutex> lk(q_mtx_);
+      mqueue_.push_back(payload);
+    }
+  }
+};
+
+std::vector<std::string> Client::GetAllMessage() {
+  std::lock_guard<std::mutex> lk(q_mtx_);
+  std::vector<std::string> out;
+  out.swap(mqueue_);
+  return out;
+};
+
+void Client::SendToServer(std::string payload) {
+  try {
+    sock_->SendTo(server_addr.address, server_addr.port, payload);
+  } catch (const std::runtime_error &e) {
+    std::cout << "Failed to send " << e.what() << std::endl;
+  }
+}
+
+} // namespace Bored::Net

Engine/Systems/Networks/NetworkClientService.hpp

@@ -0,0 +1,38 @@
+#pragma once
+#include "../../Utils/Networks/ISocket.hpp"
+#include <atomic>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+namespace Bored::Net {
+
+constexpr int kConnectTimeOut = 2; // ms
+
+class Client {
+public:
+  Client(std::string compatible_id = "");
+  ~Client();
+
+  bool Connect(int port, std::string addr, int server_port);
+  void Disconnect();
+
+  void SendToServer(std::string payload);
+  std::vector<std::string> GetAllMessage();
+
+private:
+  void initSocket();
+  void listenLoop(int port);
+  Conn server_addr;  
+  std::string compatible_id_;
+
+  std::shared_ptr<ISocket> sock_;
+
+  std::vector<std::string> mqueue_;
+  std::mutex q_mtx_;
+  std::atomic<bool> running_;
+  std::thread listener_;
+};
+
+} // namespace Bored::Net

Engine/Systems/Networks/NetworkServerService.cpp

@@ -0,0 +1,142 @@
+#include "NetworkServerService.hpp"
+#include <iostream>
+#include <memory>
+#include <mutex>
+#include <stdexcept>
+#ifdef _WIN32
+#include "../../Utils/Networks/WindowsSocket.hpp"
+#endif
+
+namespace Bored::Net {
+
+Server::Server(std::string compatible_id) {
+  initSocket();
+  compatible_id_ = compatible_id;
+};
+
+Server::~Server() { Stop(); };
+
+void Server::Start(int port) {
+  if (running_.load()) {
+    return;
+  }
+
+  running_ = true;
+  try {
+    sock_->Open(IPv4, Datagram, UDP);
+    sock_->Bind("", port);
+  } catch (const std::runtime_error &err) {
+    std::cout << "Error trying to start server: " << err.what() << std::endl;
+  }
+
+  listener_ = std::thread(&Server::listenLoop, this, port);
+};
+
+void Server::Stop() {
+  if (!running_.load())
+    return;
+  running_ = false;
+  if (listener_.joinable()) {
+    listener_.join();
+  }
+
+  sock_->Close();
+};
+
+void Server::BroadCastMessage(std::string payload) {
+  for (auto c : clients_) {
+    try {
+      sock_->SendTo(c.address, c.port, payload);
+    } catch (const std::runtime_error &err) {
+      std::cout << "Failed to broadcast: " << err.what() << std::endl;
+      handleFailedSendConn(c);
+    }
+  }
+};
+
+std::vector<Msg> Server::GetAllMessage() {
+  std::lock_guard<std::mutex> lk(q_mtx_);
+  std::vector<Msg> out;
+  out.swap(mqueue_);
+  return out;
+};
+
+void Server::handleNewConn(Conn conn) {
+  std::cout << "Recevied connect req from: " << conn.address << ":" << conn.port
+            << std::endl;
+  sock_->SendTo(conn.address, conn.port, compatible_id_);
+  clients_.insert(conn);
+};
+
+void Server::handleFailedSendConn(Conn conn) {
+  if (retried_.count(conn) >= kRetried) {
+    std::lock_guard<std::mutex> lk(c_mtx_);
+    retried_.erase(conn);
+    clients_.erase(conn);
+  }
+
+  retried_[conn]++;
+}
+
+void Server::initSocket() {
+#ifdef _WIN32
+  sock_ = std::make_shared<WindowsSocket>();
+#else
+  // TODO: extend to more than win
+  throw std::runtime_error("Haven't got support for current platform yet");
+#endif
+};
+
+void Server::listenLoop(int port) {
+  while (running_.load()) {
+
+    bool have_msg = false;
+#ifdef _WIN32
+    auto *ws = dynamic_cast<WindowsSocket *>(sock_.get());
+    if (ws->HasReadable(200)) {
+      have_msg = true;
+    }
+#endif
+
+    if (!running_.load()) {
+      break;
+    };
+    if (!have_msg) {
+      continue;
+    }
+
+    std::string from;
+    int port;
+    std::string payload;
+    try {
+      payload = sock_->ReceiveFrom(from, port);
+    } catch (const std::runtime_error &e) {
+      std::cout << "Recevied from error: " << e.what() << std::endl;
+      if (!running_.load()) {
+        break;
+      }
+      continue;
+    }
+
+    {
+      std::lock_guard<std::mutex> lk(c_mtx_);
+      Conn new_c(from, port);
+      const std::string suffix = ":connect_req";
+      if (!clients_.count(new_c) && payload.ends_with(suffix)) {
+        std::string sent_id = payload.substr(0, payload.size() - suffix.size());
+        if (sent_id == compatible_id_) {
+          handleNewConn(new_c);
+        }
+      } else {
+        if (retried_.find(new_c) != retried_.end()) {
+          retried_.erase(new_c);
+        };
+        std::lock_guard<std::mutex> lk(q_mtx_);
+        Msg new_m(from, port, payload);
+        mqueue_.push_back(new_m);
+      }
+    }
+  }
+};
+
+} // namespace Bored::Net

Engine/Systems/Networks/NetworkServerService.hpp

@@ -0,0 +1,45 @@
+#pragma once
+#include "../../Utils/Networks/ISocket.hpp"
+#include <atomic>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+namespace Bored::Net {
+
+constexpr int kRetried = 3;
+
+class Server {
+public:
+  Server(std::string compatible_id = "");
+  ~Server();
+  void Start(int port = 8080);
+  void Stop();
+  void BroadCastMessage(std::string payload);
+  std::vector<Msg> GetAllMessage();
+
+private:
+  void initSocket();
+  void listenLoop(int port);
+
+  void handleNewConn(Conn conn);
+  void handleFailedSendConn(Conn conn);
+
+private:
+  std::string compatible_id_;
+  std::shared_ptr<ISocket> sock_;
+  std::unordered_set<Conn> clients_;
+  std::mutex c_mtx_;
+
+  std::vector<Msg> mqueue_;
+  std::mutex q_mtx_;
+  std::atomic<bool> running_;
+
+  std::thread listener_;
+  std::unordered_map<Conn, int> retried_;
+};
+
+} // namespace Bored::Net