1.Implementation for rtp session module; 2.Separate signal server from main project

application/remote_desk/remote_desk_client/remote_desk_client.cpp

@@ -100,6 +100,7 @@ inline int ProcessMouseKeyEven(SDL_Event &ev) {
 
 void GuestReceiveBuffer(const char *data, size_t size, const char *user_id,
                         size_t user_id_size) {
+  std::cout << "Receive: [" << user_id << "] " << std::endl;
   memcpy(dst_buffer, data, size);
 
   SDL_Event event;

application/signal_server/main.cpp

@@ -0,0 +1,17 @@
+
+#include <iostream>
+
+#include "signal_server.h"
+
+int main(int argc, char* argv[]) {
+  SignalServer s;
+  std::string port = "";
+  if (argc > 1) {
+    port = argv[1];
+  } else {
+    port = "9090";
+  }
+  std::cout << "Port: " << port << std::endl;
+  s.run(std::stoi(port));
+  return 0;
+}

application/signal_server/signal_server.cpp

@@ -0,0 +1,238 @@
+#include "signal_server.h"
+
+#include "common.h"
+#include "log.h"
+
+const std::string GenerateTransmissionId() {
+  static const char alphanum[] = "0123456789";
+  std::string random_id;
+  random_id.reserve(6);
+
+  for (int i = 0; i < 6; ++i) {
+    random_id += alphanum[rand() % (sizeof(alphanum) - 1)];
+  }
+
+  return "000000";
+}
+
+SignalServer::SignalServer() {
+  // Set logging settings
+  server_.set_error_channels(websocketpp::log::elevel::all);
+  server_.set_access_channels(websocketpp::log::alevel::none);
+
+  // Initialize Asio
+  server_.init_asio();
+
+  server_.set_open_handler(
+      std::bind(&SignalServer::on_open, this, std::placeholders::_1));
+
+  server_.set_close_handler(
+      std::bind(&SignalServer::on_close, this, std::placeholders::_1));
+
+  server_.set_message_handler(std::bind(&SignalServer::on_message, this,
+                                        std::placeholders::_1,
+                                        std::placeholders::_2));
+
+  server_.set_ping_handler(bind(&SignalServer::on_ping, this,
+                                std::placeholders::_1, std::placeholders::_2));
+
+  server_.set_pong_handler(bind(&SignalServer::on_pong, this,
+                                std::placeholders::_1, std::placeholders::_2));
+}
+
+SignalServer::~SignalServer() {}
+
+bool SignalServer::on_open(websocketpp::connection_hdl hdl) {
+  ws_connections_[hdl] = ws_connection_id_++;
+  LOG_INFO("New websocket connection [{}] established", ws_connection_id_);
+
+  json message = {{"type", "ws_connection_id"},
+                  {"ws_connection_id", ws_connection_id_}};
+  server_.send(hdl, message.dump(), websocketpp::frame::opcode::text);
+
+  return true;
+}
+
+bool SignalServer::on_close(websocketpp::connection_hdl hdl) {
+  LOG_INFO("Websocket onnection [{}] closed", ws_connection_id_);
+  std::string user_id = transmission_manager_.GetUserId(hdl);
+  std::string transmission_id =
+      transmission_manager_.ReleaseUserIdFromTransmission(hdl);
+
+  json message = {{"type", "user_leave_transmission"},
+                  {"transmission_id", transmission_id},
+                  {"user_id", user_id}};
+
+  std::vector<std::string> user_id_list =
+      transmission_manager_.GetAllUserIdOfTransmission(transmission_id);
+
+  if (user_id_list.empty()) {
+    transmission_list_.erase(transmission_id);
+    LOG_INFO("Release transmission [{}] due to no user in this transmission",
+             transmission_id);
+  }
+
+  for (const auto& user_id : user_id_list) {
+    send_msg(transmission_manager_.GetWsHandle(user_id), message);
+  }
+  ws_connections_.erase(hdl);
+
+  return true;
+}
+
+bool SignalServer::on_ping(websocketpp::connection_hdl hdl, std::string s) {
+  /* Do something */
+  LOG_INFO("Receive ping");
+  return true;
+}
+
+bool SignalServer::on_pong(websocketpp::connection_hdl hdl, std::string s) {
+  /* Do something */
+  LOG_INFO("pong");
+  return true;
+}
+
+void SignalServer::run(uint16_t port) {
+  server_.set_reuse_addr(true);
+  server_.listen(port);
+
+  // Queues a connection accept operation
+  server_.start_accept();
+
+  // Start the Asio io_service run loop
+  server_.run();
+}
+
+void SignalServer::send_msg(websocketpp::connection_hdl hdl, json message) {
+  server_.send(hdl, message.dump(), websocketpp::frame::opcode::text);
+}
+
+void SignalServer::on_message(websocketpp::connection_hdl hdl,
+                              server::message_ptr msg) {
+  std::string payload = msg->get_payload();
+
+  auto j = json::parse(payload);
+  std::string type = j["type"].get<std::string>();
+
+  switch (HASH_STRING_PIECE(type.c_str())) {
+    case "create_transmission"_H: {
+      std::string transmission_id = j["transmission_id"].get<std::string>();
+      std::string user_id = j["user_id"].get<std::string>();
+      LOG_INFO("Receive user id [{}] create transmission request with id [{}]",
+               user_id, transmission_id);
+      if (transmission_list_.find(transmission_id) ==
+          transmission_list_.end()) {
+        if (transmission_id.empty()) {
+          transmission_id = GenerateTransmissionId();
+          while (transmission_list_.find(transmission_id) !=
+                 transmission_list_.end()) {
+            transmission_id = GenerateTransmissionId();
+          }
+          LOG_INFO(
+              "Transmission id is empty, generate a new one for this request "
+              "[{}]",
+              transmission_id);
+        }
+        transmission_list_.insert(transmission_id);
+
+        transmission_manager_.BindUserIdToTransmission(user_id,
+                                                       transmission_id);
+        transmission_manager_.BindUserIdToWsHandle(user_id, hdl);
+
+        LOG_INFO("Create transmission id [{}]", transmission_id);
+        json message = {{"type", "transmission_id"},
+                        {"transmission_id", transmission_id},
+                        {"status", "success"}};
+        send_msg(hdl, message);
+      } else {
+        LOG_INFO("Transmission id [{}] already exist", transmission_id);
+        json message = {{"type", "transmission_id"},
+                        {"transmission_id", transmission_id},
+                        {"status", "fail"},
+                        {"reason", "Transmission id exist"}};
+        send_msg(hdl, message);
+      }
+
+      break;
+    }
+    case "query_user_id_list"_H: {
+      std::string transmission_id = j["transmission_id"].get<std::string>();
+      std::vector<std::string> user_id_list =
+          transmission_manager_.GetAllUserIdOfTransmission(transmission_id);
+
+      json message = {{"type", "user_id_list"},
+                      {"transmission_id", transmission_id},
+                      {"user_id_list", user_id_list},
+                      {"status", "success"}};
+
+      // LOG_INFO("Send member_list: [{}]", message.dump());
+      send_msg(hdl, message);
+      break;
+    }
+    case "offer"_H: {
+      std::string transmission_id = j["transmission_id"].get<std::string>();
+      std::string sdp = j["sdp"].get<std::string>();
+      std::string user_id = j["user_id"].get<std::string>();
+      std::string remote_user_id = j["remote_user_id"].get<std::string>();
+
+      transmission_manager_.BindUserIdToTransmission(user_id, transmission_id);
+      transmission_manager_.BindUserIdToWsHandle(user_id, hdl);
+
+      websocketpp::connection_hdl destination_hdl =
+          transmission_manager_.GetWsHandle(remote_user_id);
+
+      json message = {{"type", "offer"},
+                      {"sdp", sdp},
+                      {"remote_user_id", user_id},
+                      {"transmission_id", transmission_id}};
+
+      LOG_INFO("[{}] send offer to [{}]", user_id, remote_user_id);
+      send_msg(destination_hdl, message);
+
+      break;
+    }
+    case "answer"_H: {
+      std::string transmission_id = j["transmission_id"].get<std::string>();
+      std::string sdp = j["sdp"].get<std::string>();
+      std::string user_id = j["user_id"].get<std::string>();
+      std::string remote_user_id = j["remote_user_id"].get<std::string>();
+
+      websocketpp::connection_hdl destination_hdl =
+          transmission_manager_.GetWsHandle(remote_user_id);
+
+      // LOG_INFO("send answer sdp [{}]", sdp);
+      LOG_INFO("[{}] send answer to [{}]", user_id, remote_user_id);
+      json message = {{"type", "answer"},
+                      {"sdp", sdp},
+                      {"remote_user_id", user_id},
+                      {"transmission_id", transmission_id}};
+      send_msg(destination_hdl, message);
+      break;
+    }
+    case "offer_candidate"_H: {
+      std::string transmission_id = j["transmission_id"].get<std::string>();
+      std::string candidate = j["sdp"].get<std::string>();
+      LOG_INFO("send candidate [{}]", candidate.c_str());
+      json message = {{"type", "candidate"}, {"sdp", candidate}};
+      // send_msg(answer_hdl_map_[transmission_id], message);
+      break;
+    }
+    case "answer_candidate"_H: {
+      std::string transmission_id = j["transmission_id"].get<std::string>();
+      std::string candidate = j["sdp"].get<std::string>();
+      LOG_INFO("send candidate [{}]", candidate.c_str());
+      json message = {{"type", "candidate"}, {"sdp", candidate}};
+      // send_msg(offer_hdl_map_[transmission_id], message);
+      break;
+    }
+    default:
+      break;
+  }
+
+  // std::string sdp = j["sdp"];
+
+  // LOG_INFO("Message type: {}", type);
+  // LOG_INFO("Message body: {}", sdp);
+
+  // server_.send(hdl, msg->get_payload(), msg->get_opcode());
+}

application/signal_server/signal_server.h

@@ -0,0 +1,51 @@
+#ifndef _SIGNAL_SERVER_H_
+#define _SIGNAL_SERVER_H_
+
+#include <functional>
+#include <map>
+#include <nlohmann/json.hpp>
+#include <set>
+#include <string>
+#include <websocketpp/config/asio_no_tls.hpp>
+#include <websocketpp/server.hpp>
+
+#include "transmission_manager.h"
+
+using nlohmann::json;
+
+typedef websocketpp::server<websocketpp::config::asio> server;
+typedef unsigned int connection_id;
+typedef std::string room_id;
+
+class SignalServer {
+ public:
+  SignalServer();
+  ~SignalServer();
+
+  bool on_open(websocketpp::connection_hdl hdl);
+
+  bool on_close(websocketpp::connection_hdl hdl);
+
+  bool on_ping(websocketpp::connection_hdl hdl, std::string s);
+
+  bool on_pong(websocketpp::connection_hdl hdl, std::string s);
+
+  void run(uint16_t port);
+
+  void on_message(websocketpp::connection_hdl hdl, server::message_ptr msg);
+
+  void send_msg(websocketpp::connection_hdl hdl, json message);
+
+ private:
+  server server_;
+  std::map<websocketpp::connection_hdl, connection_id,
+           std::owner_less<websocketpp::connection_hdl>>
+      ws_connections_;
+  std::map<room_id, connection_id> rooms_;
+  unsigned int ws_connection_id_ = 0;
+
+  std::set<std::string> transmission_list_;
+  TransmissionManager transmission_manager_;
+};
+
+#endif

application/signal_server/transmission_manager.cpp

@@ -0,0 +1,95 @@
+#include "transmission_manager.h"
+
+#include "log.h"
+
+TransmissionManager::TransmissionManager() {}
+
+TransmissionManager::~TransmissionManager() {}
+
+std::vector<std::string> TransmissionManager::GetAllUserIdOfTransmission(
+    const std::string& transmission_id) {
+  if (transmission_user_id_list_.find(transmission_id) !=
+      transmission_user_id_list_.end()) {
+    return transmission_user_id_list_[transmission_id];
+  }
+
+  return std::vector<std::string>();
+}
+
+bool TransmissionManager::BindUserIdToTransmission(
+    const std::string& user_id, const std::string& transmission_id) {
+  if (transmission_user_id_list_.find(transmission_id) ==
+      transmission_user_id_list_.end()) {
+    transmission_user_id_list_[transmission_id].push_back(user_id);
+    LOG_INFO("Bind user id [{}]  to transmission [{}]", user_id,
+             transmission_id);
+    return true;
+  } else {
+    auto user_id_list = transmission_user_id_list_[transmission_id];
+    for (auto id : user_id_list) {
+      if (id == user_id) {
+        LOG_WARN("User id [{}] already bind to transmission [{}]", user_id,
+                 transmission_id);
+        return false;
+      }
+    }
+    transmission_user_id_list_[transmission_id].push_back(user_id);
+    LOG_INFO("Bind user id [{}]  to transmission [{}]", user_id,
+             transmission_id);
+  }
+  return true;
+}
+
+bool TransmissionManager::BindUserIdToWsHandle(
+    const std::string& user_id, websocketpp::connection_hdl hdl) {
+  if (user_id_ws_hdl_list_.find(user_id) != user_id_ws_hdl_list_.end()) {
+    LOG_WARN("User id already bind to websocket handle [{}]", user_id,
+             hdl.lock().get());
+    return false;
+  } else {
+    user_id_ws_hdl_list_[user_id] = hdl;
+  }
+  return true;
+}
+
+std::string TransmissionManager::ReleaseUserIdFromTransmission(
+    websocketpp::connection_hdl hdl) {
+  for (auto it = user_id_ws_hdl_list_.begin(); it != user_id_ws_hdl_list_.end();
+       ++it) {
+    if (it->second.lock().get() == hdl.lock().get()) {
+      for (auto trans_it = transmission_user_id_list_.begin();
+           trans_it != transmission_user_id_list_.end(); ++trans_it) {
+        auto& user_id_list = trans_it->second;
+        auto user_id_it =
+            std::find(user_id_list.begin(), user_id_list.end(), it->first);
+        if (user_id_it != user_id_list.end()) {
+          user_id_list.erase(user_id_it);
+          LOG_INFO("Remove user id [{}] from transmission [{}]", it->first,
+                   trans_it->first);
+          user_id_ws_hdl_list_.erase(it);
+          return trans_it->first;
+        }
+      }
+    }
+  }
+  return "";
+}
+
+websocketpp::connection_hdl TransmissionManager::GetWsHandle(
+    const std::string& user_id) {
+  if (user_id_ws_hdl_list_.find(user_id) != user_id_ws_hdl_list_.end()) {
+    return user_id_ws_hdl_list_[user_id];
+  } else {
+    websocketpp::connection_hdl hdl;
+    return hdl;
+  }
+}
+
+std::string TransmissionManager::GetUserId(websocketpp::connection_hdl hdl) {
+  for (auto it = user_id_ws_hdl_list_.begin(); it != user_id_ws_hdl_list_.end();
+       ++it) {
+    LOG_INFO("[{}]", it->first);
+    if (it->second.lock().get() == hdl.lock().get()) return it->first;
+  }
+  return "";
+}

application/signal_server/transmission_manager.h

@@ -0,0 +1,33 @@
+#ifndef _TRANSIMISSION_MANAGER_H_
+#define _TRANSIMISSION_MANAGER_H_
+
+#include <map>
+#include <set>
+#include <websocketpp/server.hpp>
+
+class TransmissionManager {
+ public:
+  TransmissionManager();
+  ~TransmissionManager();
+
+ public:
+  std::vector<std::string> GetAllUserIdOfTransmission(
+      const std::string& transmission_id);
+
+ public:
+  bool BindUserIdToTransmission(const std::string& user_id,
+                                const std::string& transmission_id);
+  bool BindUserIdToWsHandle(const std::string& user_id,
+                            websocketpp::connection_hdl hdl);
+
+  std::string ReleaseUserIdFromTransmission(websocketpp::connection_hdl hdl);
+
+  websocketpp::connection_hdl GetWsHandle(const std::string& user_id);
+  std::string GetUserId(websocketpp::connection_hdl hdl);
+
+ private:
+  std::map<std::string, std::vector<std::string>> transmission_user_id_list_;
+  std::map<std::string, websocketpp::connection_hdl> user_id_ws_hdl_list_;
+};
+
+#endif

application/signal_server/xmake.lua

@@ -0,0 +1,43 @@
+set_project("projectx")
+set_version("0.0.1")
+
+add_rules("mode.release", "mode.debug")
+set_languages("c++17")
+
+add_rules("mode.release", "mode.debug")
+
+add_requires("asio 1.24.0", "nlohmann_json", "spdlog 1.11.0")
+
+add_defines("ASIO_STANDALONE", "ASIO_HAS_STD_TYPE_TRAITS", "ASIO_HAS_STD_SHARED_PTR", 
+    "ASIO_HAS_STD_ADDRESSOF", "ASIO_HAS_STD_ATOMIC", "ASIO_HAS_STD_CHRONO", "ASIO_HAS_CSTDINT", "ASIO_HAS_STD_ARRAY",
+    "ASIO_HAS_STD_SYSTEM_ERROR")
+
+if is_os("windows") then
+    add_defines("_WEBSOCKETPP_CPP11_INTERNAL_")
+    add_links("ws2_32", "Bcrypt")
+    add_requires("cuda")
+elseif is_os("linux") then 
+    add_links("pthread")
+    set_config("cxxflags", "-fPIC")
+end
+
+add_packages("spdlog")
+
+includes("../../thirdparty")
+
+target("log")
+    set_kind("headeronly")
+    add_packages("spdlog")
+    add_headerfiles("../../src/log/log.h")
+    add_includedirs("../../src/log", {public = true})
+
+target("common")
+    set_kind("headeronly")
+    add_includedirs("../../src/common", {public = true})
+
+target("signal_server")
+    set_kind("binary")
+    add_deps("log", "common")
+    add_files("*.cpp")
+    add_packages("asio", "nlohmann_json", "spdlog")
+    add_includedirs("../../thirdparty/websocketpp/include")