server：初步添加获取客户端列表功能。

.vscode/settings.json

@@ -121,6 +121,8 @@
         "source_location": "cpp",
         "span": "cpp",
         "stop_token": "cpp",
-        "*.ipp": "cpp"
+        "*.ipp": "cpp",
+        "queue": "cpp",
+        "resumable": "cpp"
     }
 }

README.md

@@ -7,3 +7,35 @@
 - `client`和`server`均为命令行端程序，无GUI。
 - `client`从`server`下载文件，如果本地有重复则覆盖。
 - 工作方式为`client A`端提交待传输的文件列表到`server`，`client B`端从`server`获取有哪些客户机提交的哪些任务，可以从中下载。
+
+## 格式（开发用）
+
+通讯协议中的`DATA`部分，对于`type`为`1`的类型来讲，统一以下格式（`command`和`param`内容中不能含有`|`）：
+
+**command|param|data** （传输格式）
+
+**command(空格)param1,param2,param3..** （cmd输入格式）
+
+### 1.命令
+
+`type`：1
+
+**GetTaskList：** 获取当前挂载到服务器的任务单。
+
+**DownTask：** 下载指定的任务清单，`param`为`GetTaskList`中列出的名称。
+
+**UpTask：** 上载任务单，`param`为文件或者文件夹路径，可多个，使用`,`分隔。
+
+### 2.数据
+
+`type`：2
+
+当`mark`为`0`时表示数据的最后一包，其他数据表示非最后一包。
+
+`type`: 199
+
+特殊标记，表示询问在线客户端。
+
+`type`: 198
+
+特殊标记，表示询问挂载任务。

client/CMakeLists.txt

@@ -1,6 +1,7 @@
 cmake_minimum_required(VERSION 3.16)
 
 project(transmc LANGUAGES CXX)
+set(CMAKE_CXX_STANDARD 17)
 
 if (MSVC)
     add_definitions(-D_WIN32_WINNT=0x0601)

client/client.cpp

@@ -19,7 +19,7 @@ void CClient::run()
         logger_->info("{} connect err.", __FUNCTION__);
         return;
     }
-    client_->register_func([&](CFrameBuffer* buf) { handle_data(buf); });
+    client_->register_func([&](CFrameBuffer* buf) { handle_frame(buf); });
     client_->async_recv();
     std::thread thread([&]() { io_context_.run(); });
     char line[512]{};
@@ -53,7 +53,7 @@ bool CClient::get_task_list()
     buf->data_ = new char[512]{};
     auto flen = std::snprintf(buf->data_, 512, "%s", gGetTaskList);
     buf->len_ = flen;
-    buf->type_ = 1;
+    buf->type_ = 199;
     if (!CTransProtocal::pack(buf.get(), &send, len)) {
         logger_->error("{} pack failed.", __FUNCTION__);
         return false;
@@ -65,7 +65,12 @@ bool CClient::get_task_list()
     return true;
 }
 
-void CClient::handle_data(CFrameBuffer* buf)
+bool CClient::get_clients()
+{
+    return false;
+}
+
+void CClient::handle_frame(CFrameBuffer* buf)
 {
     if (buf == nullptr) {
         logger_->error("{} nullptr.", __FUNCTION__);
@@ -73,4 +78,8 @@ void CClient::handle_data(CFrameBuffer* buf)
     }
     logger_->debug("type: {}", buf->type_);
     logger_->debug("len: {}", buf->len_);
+
+    if (buf->type_ == 199) {
+        logger_->debug("data: {}", buf->data_);
+    }
 }

client/client.h

@@ -15,9 +15,10 @@ public:
 
 public:
     bool get_task_list();
+    bool get_clients();
 
 private:
-    void handle_data(CFrameBuffer* buf);
+    void handle_frame(CFrameBuffer* buf);
 
 private:
     std::shared_ptr<spdlog::logger> logger_;

net/net_base.cpp

@@ -57,7 +57,13 @@ void CTcpClient::async_recv()
     auto self(shared_from_this());
     socket_.async_read_some(asio::buffer(tmp_buf_), [this, self](std::error_code ec, std::size_t length) {
         // logger_->debug("{} {}", __FUNCTION__, ec.message());
-        if (!ec) {
+        if (ec) {
+            if (ec == asio::error::eof) {
+                logger_->error("Remote Server Closed.");
+                return;
+            }
+            async_recv();
+        } else {
             std::lock_guard<std::mutex> lock(mutex_);
             buffer_.push(tmp_buf_.data(), length);
             auto* frame = CTransProtocal::parse(buffer_);

ofen

@@ -1 +1 @@
-Subproject commit 1427bf4bca2c0f52429fe751c4c205acfd78b326
+Subproject commit 44fb416ca166aa9c4ce270ff71bea6b3df703d6e

server/CMakeLists.txt

@@ -1,6 +1,7 @@
 cmake_minimum_required(VERSION 3.16)
 
 project(transms LANGUAGES CXX)
+set(CMAKE_CXX_STANDARD 17)
 
 if (MSVC)
     add_compile_options(/source-charset:utf-8)

server/main.cpp

@@ -7,7 +7,7 @@ int main()
     g_Logger = get_logger("server", "server.log");
     asio::io_context io_context;
     CTcpServer server(io_context, g_Logger);
-    if (!server.Start(8080)) {
+    if (!server.start(8080)) {
         return -1;
     }
     io_context.run();

server/server.cpp

@@ -3,12 +3,22 @@
 CTcpServer::CTcpServer(asio::io_context& io_context, const std::shared_ptr<spdlog::logger>& logger)
     : io_context_(io_context), logger_(logger), acceptor_(io_context)
 {
+    th_run_ = true;
+    handle_pool_ = std::make_shared<CThreadPool>(1);
+    send_pool_ = std::make_shared<CThreadPool>(1);
+    handle_pool_->init();
+    send_pool_->init();
+    handle_pool_->submit([&]() { handle_frame(); });
+    send_pool_->submit([&]() { send_simple_buf(); });
 }
 CTcpServer::~CTcpServer()
 {
+    th_run_ = false;
+    handle_pool_->close_wait_all();
+    send_pool_->close_wait_all();
 }
 
-bool CTcpServer::Start(unsigned short port)
+bool CTcpServer::start(unsigned short port)
 {
     asio::ip::tcp::endpoint endpoint(asio::ip::tcp::v4(), port);
     try {
@@ -20,16 +30,15 @@ bool CTcpServer::Start(unsigned short port)
         logger_->error("Failed to bind to {}: {}", endpoint.address().to_string(), e.what());
         return false;
     }
-
+    accept_client();
-    Accept();
     logger_->info("Server started on port {}", port);
     return true;
 }
 
-void CTcpServer::Stop()
+void CTcpServer::stop()
 {
     acceptor_.close();
-    std::lock_guard<std::mutex> lock(mutex_);
+    std::lock_guard<std::mutex> lock(cli_mut_);
     for (auto& [key, thread] : client_threads_) {
         if (thread.joinable()) {
             thread.join();
@@ -38,15 +47,110 @@ void CTcpServer::Stop()
     client_threads_.clear();
 }
 
-void CTcpServer::handle_data(CFrameBuffer* buf)
+std::vector<std::string> CTcpServer::get_clients()
 {
-    if (buf == nullptr) {
+    std::vector<std::string> result;
-        logger_->error("{} nullptr.", __FUNCTION__);
+    std::lock_guard<std::mutex> lock(cli_mut_);
-        return;
+    for (const auto& item : client_map_) {
+        result.push_back(item.first);
+    }
+    return result;
+}
+
+SimpleBuffer* CTcpServer::get_client_list()
+{
+    CFrameBuffer* buf = new CFrameBuffer();
+    buf->type_ = 199;
+
+    auto vec = get_clients();
+    std::string msg;
+    for (const auto& item : vec) {
+        if (msg.empty()) {
+            msg.append(item);
+        } else {
+            msg.append("|" + item);
+        }
+    }
+    buf->data_ = new char[msg.size() + 1];
+    buf->len_ = static_cast<int>(msg.size() + 1);
+    std::snprintf(buf->data_, buf->len_, "%s", msg.data());
+
+    SimpleBuffer* sbuf = new SimpleBuffer();
+    if (!CTransProtocal::pack(buf, &sbuf->data_, sbuf->len_)) {
+        logger_->error("{} pack faile.", __FUNCTION__);
+        delete sbuf;
+        return nullptr;
+    }
+    return sbuf;
+}
+
+bool CTcpServer::push_frame(CFrameBuffer* buf)
+{
+    std::lock_guard<std::mutex> lock(buf_mut_);
+    cache_.push(buf);
+    return true;
+}
+
+void CTcpServer::handle_frame()
+{
+    CFrameBuffer* buf = nullptr;
+    while (th_run_) {
+        {
+            std::lock_guard<std::mutex> lock(buf_mut_);
+            if (cache_.size() > 0) {
+                buf = cache_.front();
+                cache_.pop();
+            } else {
+                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+                continue;
             }
         }
 
-void CTcpServer::Accept()
+        // 拿到该包后，要看转发给谁或者处理
+        if (buf->type_ == 199) {   // 询问在线客户端
+            auto* sbuf = get_client_list();
+            if (sbuf == nullptr) {
+                continue;
+            }
+            sbuf->id_ = buf->id_;
+            std::lock_guard<std::mutex> lock(sbuf_mut_);
+            scache_.push(sbuf);
+            continue;
+        }
+        delete buf;
+    }
+}
+
+void CTcpServer::send_simple_buf()
+{
+    SimpleBuffer* buf = nullptr;
+    while (th_run_) {
+        {
+            std::lock_guard<std::mutex> slock(sbuf_mut_);
+            if (scache_.size() > 0) {
+                buf = scache_.front();
+                scache_.pop();
+            } else {
+                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+                continue;
+            }
+        }
+        std::shared_ptr<asio::ip::tcp::socket> socket = nullptr;
+        {
+            std::lock_guard<std::mutex> clock(cli_mut_);
+            if (!client_map_.count(buf->id_)) {
+                logger_->warn("{} abandon {}'s data.", __FUNCTION__, buf->id_);
+                delete buf;
+                continue;
+            }
+            socket = client_map_[buf->id_]->socket_;
+        }
+        socket->send(asio::buffer(buf->data_, buf->len_));
+        delete buf;
+    }
+}
+
+void CTcpServer::accept_client()
 {
     auto socket = std::make_shared<asio::ip::tcp::socket>(io_context_);
     acceptor_.async_accept(*socket, [this, socket](const asio::error_code& error) {
@@ -56,22 +160,46 @@ void CTcpServer::Accept()
             logger_->info("New connection from {}", client_key);
 
             {
-                std::lock_guard<std::mutex> lock(mutex_);
+                std::lock_guard<std::mutex> lock(cli_mut_);
-                client_map_[client_key] = std::make_shared<ClientCache>();
+                auto cache = std::make_shared<ClientCache>();
+                cache->socket_ = socket;
+                client_map_[client_key] = cache;
             }
 
-            client_threads_[client_key] = std::thread(&CTcpServer::HandleClient, this, socket, client_key);
+            client_threads_[client_key] = std::thread(&CTcpServer::th_client, this, socket, client_key);
         }
-        Accept();
+        accept_client();
     });
 }
 
-void CTcpServer::HandleClient(std::shared_ptr<asio::ip::tcp::socket> socket, const std::string& client_key)
+void CTcpServer::th_client(std::shared_ptr<asio::ip::tcp::socket> socket, const std::string& client_key)
 {
+    std::shared_ptr<int> deleter(new int(0), [&](int* p) {
+        std::lock_guard<std::mutex> lock(cli_mut_);
+        delete p;
+        client_map_.erase(client_key);
+        if (client_threads_.find(client_key) != client_threads_.end()) {
+            client_threads_.at(client_key).detach();
+            client_threads_.erase(client_key);
+        }
+        logger_->warn("{} client {} exit.", __FUNCTION__, client_key);
+    });
+
     try {
+        std::shared_ptr<ClientCache> cache = nullptr;
+
+        {
+            std::lock_guard<std::mutex> lock(cli_mut_);
+            if (!client_map_.count(client_key)) {
+                logger_->error("Not Find Client{} in cache.", client_key);
+                return;
+            }
+            cache = client_map_[client_key];
+        }
+
         while (true) {
             asio::error_code error;
-            size_t length = socket->read_some(asio::buffer(client_map_[client_key]->tmp_buf_), error);
+            size_t length = socket->read_some(asio::buffer(cache->tmp_buf_), error);
             if (error == asio::error::eof) {
                 logger_->info("Connection closed by client: {}", client_key);
                 break;
@@ -79,25 +207,14 @@ void CTcpServer::HandleClient(std::shared_ptr<asio::ip::tcp::socket> socket, con
                 throw asio::system_error(error);
             }
 
-            client_map_[client_key]->buffer_.push(client_map_[client_key]->tmp_buf_.data(), length);
+            cache->buffer_.push(cache->tmp_buf_.data(), length);
-            auto* frame = CTransProtocal::parse(client_map_[client_key]->buffer_);
+            auto* frame = CTransProtocal::parse(cache->buffer_);
-            frame->id_ = client_key;
             if (frame) {
-                handle_data(frame);
+                frame->id_ = client_key;
-                delete frame;
+                push_frame(frame);
             }
-
-            // auto relen = socket->send(asio::buffer(data, length));
-            // logger_->info("Received data from {}, len={}, relen={}", client_key, length, relen);
         }
     } catch (std::exception& e) {
         logger_->error("Error with client {}: {}", client_key, e.what());
     }
-
-    std::lock_guard<std::mutex> lock(mutex_);
-    client_map_.erase(client_key);
-    if (client_threads_.find(client_key) != client_threads_.end()) {
-        client_threads_.at(client_key).detach();
-        client_threads_.erase(client_key);
-    }
 }

server/server.h

@@ -1,10 +1,14 @@
 #pragma once
 #include <net_base.h>
+#include <of_util.h>
+#include <queue>
 #include <string>
 #include <util.h>
 #include <vector>
 
+using namespace ofen;
 struct ClientCache {
+    std::shared_ptr<asio::ip::tcp::socket> socket_;
     CMutBuffer buffer_;
     std::array<char, 1024> tmp_buf_;
 };
@@ -16,21 +20,34 @@ public:
     ~CTcpServer();
 
 public:
-    bool Start(unsigned short port);
+    bool start(unsigned short port);
-    void Stop();
+    void stop();
 
 private:
-    void handle_data(CFrameBuffer* buf);
+    std::vector<std::string> get_clients();
+    SimpleBuffer* get_client_list();
 
 private:
-    void Accept();
+    bool push_frame(CFrameBuffer* buf);
-    void HandleClient(std::shared_ptr<asio::ip::tcp::socket> socket, const std::string& client_key);
+    void handle_frame();
+    void send_simple_buf();
 
 private:
+    void accept_client();
+    void th_client(std::shared_ptr<asio::ip::tcp::socket> socket, const std::string& client_key);
+
+private:
+    bool th_run_{false};
     asio::io_context& io_context_;
     asio::ip::tcp::acceptor acceptor_;
     std::shared_ptr<spdlog::logger> logger_;
     std::map<std::string, std::shared_ptr<ClientCache>> client_map_;
     std::map<std::string, std::thread> client_threads_;
-    std::mutex mutex_;
+    std::mutex cli_mut_;
+    std::queue<CFrameBuffer*> cache_;
+    std::queue<SimpleBuffer*> scache_;
+    std::mutex buf_mut_;
+    std::mutex sbuf_mut_;
+    std::shared_ptr<CThreadPool> handle_pool_;
+    std::shared_ptr<CThreadPool> send_pool_;
 };

util/util.cpp

@@ -58,6 +58,8 @@ CFrameBuffer* CTransProtocal::parse(CMutBuffer& buffer)
     result = new CFrameBuffer();
     result->data_ = new char[len];
     result->len_ = len;
+    result->mark_ = mark;
+    result->type_ = type;
     std::memset(result->data_, 0x0, len);
     std::memcpy(result->data_, buffer.get_data() + find + 2 + 2 + 1 + 4, len);
     buffer.remove_of(0, tail_index + 2);
@@ -91,3 +93,9 @@ CFrameBuffer::~CFrameBuffer()
     delete[] data_;
     len_ = 0;
 }
+
+SimpleBuffer::~SimpleBuffer()
+{
+    delete[] data_;
+    len_ = 0;
+}

util/util.h

@@ -15,6 +15,7 @@ class CFrameBuffer
 public:
     CFrameBuffer();
     ~CFrameBuffer();
+
 public:
     std::string id_{};
 
@@ -24,6 +25,17 @@ public:
     int len_{};
     char mark_{};
 };
+class SimpleBuffer
+{
+public:
+    SimpleBuffer() = default;
+    ~SimpleBuffer();
+
+public:
+    std::string id_;
+    char* data_{};
+    int len_{};
+};
 using ExFun_t = std::function<void(CFrameBuffer* buf)>;
 /*
 【 transm TCP 数据协议 】