From 19fa6e3d483410d7abcb669c198e0c94e4174ada Mon Sep 17 00:00:00 2001 From: taynpg Date: Fri, 13 Dec 2024 23:03:12 +0800 Subject: [PATCH] =?UTF-8?q?update=EF=BC=9A=E5=8F=AF=E4=B8=8A=E8=BD=BD?= =?UTF-8?q?=E8=8E=B7=E5=8F=96=E6=96=87=E4=BB=B6=E5=88=97=E8=A1=A8=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 4 +- client/CMakeLists.txt | 2 +- client/client.cpp | 95 ++++++++++++++++++++++++++++--------------- client/client.h | 8 +++- client/file_oper.cpp | 25 ++++++++++++ client/file_oper.h | 15 +++++++ ofen | 2 +- server/server.cpp | 53 +++++++++++++++++++----- server/server.h | 8 +++- 9 files changed, 163 insertions(+), 49 deletions(-) create mode 100644 client/file_oper.cpp create mode 100644 client/file_oper.h diff --git a/README.md b/README.md index 79cd71b..6a5dd58 100644 --- a/README.md +++ b/README.md @@ -36,4 +36,6 @@ `type`: 198，特殊标记，下载任务。 -`type`: 197，特殊标记，上载任务。 \ No newline at end of file +`type`: 197，特殊标记，上载任务。 + +`type`: 196，特殊标记，取消上载任务。 \ No newline at end of file diff --git a/client/CMakeLists.txt b/client/CMakeLists.txt index e771b63..44e1f87 100644 --- a/client/CMakeLists.txt +++ b/client/CMakeLists.txt @@ -8,5 +8,5 @@ if (MSVC) add_compile_options(/source-charset:utf-8) endif() -add_executable(transmc main.cpp client.h client.cpp) +add_executable(transmc main.cpp client.h client.cpp file_oper.h file_oper.cpp) target_link_libraries(transmc PRIVATE trans_net trans_util) \ No newline at end of file diff --git a/client/client.cpp b/client/client.cpp index 9d87922..73ac177 100644 --- a/client/client.cpp +++ b/client/client.cpp @@ -22,6 +22,7 @@ void CClient::run() client_->register_func([&](CFrameBuffer* buf) { handle_frame(buf); }); client_->async_recv(); std::thread thread([&]() { io_context_.run(); }); + logger_->warn("SupportCmd:Get|Up|Down|Cancel"); char line[512]{}; while (std::cin.getline(line, 512)) { std::string cmd_input(line); @@ -37,8 +38,7 @@ void CClient::run() std::string param{}; if (vec.size() == 1) { cmd = vec[0]; - } - else { + } else { cmd = vec[0]; param = vec[1]; } @@ -50,13 +50,17 @@ void CClient::run() int key = param.empty() ? -1 : std::stoi(param); if (task_list_.count(key)) { down_task(); - } - else { + } else { logger_->error("no task number find."); } continue; } if (cmd == "Up") { + up_task(cmd_input); + continue; + } + if (cmd == "Cancel") { + cancel_task(); continue; } logger_->error("No matched cmd."); @@ -68,27 +72,7 @@ void CClient::run() bool CClient::get_task_list() { - char* send = nullptr; - int len{}; - std::shared_ptr buf = std::make_shared(); - buf->data_ = new char[512]{}; - auto flen = std::snprintf(buf->data_, 512, "%s", gGet); - buf->len_ = flen; - buf->type_ = 199; - if (!CTransProtocal::pack(buf.get(), &send, len)) { - logger_->error("{} pack failed.", __FUNCTION__); - return false; - } - if (!client_->send(send, len)) { - return false; - } - delete[] send; - return true; -} - -bool CClient::get_clients() -{ - return false; + return send_frame(199, gGet, std::strlen(gGet)); } bool CClient::down_task() @@ -96,9 +80,46 @@ bool CClient::down_task() return false; } -bool CClient::up_task() +bool CClient::up_task(const std::string& cmd) { - return false; + auto list = CFileOpr::get_file_list(cmd); + std::string msg; + for (const auto& item : list) { + if (msg.empty()) { + msg.append(item); + } else { + msg.append("|" + item); + } + } + return send_frame(197, msg.data(), msg.size()); +} + +bool CClient::cancel_task() +{ + char buffer[] = "None"; + return send_frame(196, buffer, sizeof(buffer)); +} + +bool CClient::send_frame(int type, const char* data, int len) +{ + std::shared_ptr buf = std::make_shared(); + buf->type_ = type; + buf->data_ = new char[len + 1]; + std::memset(buf->data_, 0x0, len + 1); + buf->len_ = std::snprintf(buf->data_, len + 1, "%s", data); + char* out_buf{}; + int out_len{}; + if (!CTransProtocal::pack(buf.get(), &out_buf, out_len)) { + logger_->error("{} pack failed.", __FUNCTION__); + return false; + } + if (!client_->send(out_buf, out_len)) { + logger_->error("{} send failed.", __FUNCTION__); + delete[] out_buf; + return false; + } + delete[] out_buf; + return true; } void CClient::handle_frame(CFrameBuffer* buf) @@ -113,12 +134,22 @@ void CClient::handle_frame(CFrameBuffer* buf) if (buf->type_ == 199) { task_list_.clear(); std::string source(buf->data_); - int index = 0; - auto vec = COfStr::split(source, "|"); + auto vec = COfStr::split(source, "\n"); for (const auto& item : vec) { - task_list_[index] = item; - ++index; - logger_->warn("{}:{}", index, item); + if (item.find("[") == std::string::npos) { + logger_->info("FILE ==> {}", item); + } + else { + logger_->debug("***********************************************"); + logger_->debug("{}", item); + } } + // int index = 0; + // auto vec = COfStr::split(source, "|"); + // for (const auto& item : vec) { + // task_list_[index] = item; + // ++index; + // logger_->warn("{}:{}", index, item); + // } } } diff --git a/client/client.h b/client/client.h index 833b57e..458cb81 100644 --- a/client/client.h +++ b/client/client.h @@ -3,6 +3,7 @@ #include #include #include +#include "file_oper.h" class CClient { @@ -15,9 +16,12 @@ public: public: bool get_task_list(); - bool get_clients(); bool down_task(); - bool up_task(); + bool up_task(const std::string& cmd); + bool cancel_task(); + +private: + bool send_frame(int type, const char* data, int len); private: void handle_frame(CFrameBuffer* buf); diff --git a/client/file_oper.cpp b/client/file_oper.cpp new file mode 100644 index 0000000..518740a --- /dev/null +++ b/client/file_oper.cpp @@ -0,0 +1,25 @@ +#include "file_oper.h" + +CFileOpr::CFileOpr() +{ +} + +CFileOpr::~CFileOpr() +{ +} + +std::vector CFileOpr::get_file_list(const std::string& input) +{ + std::vector result; + std::string backup = input; + backup.erase(0, backup.find_first_of(" ")); + backup = COfStr::trim(backup); + if (backup.empty()) { + return result; + } + auto vec = COfStr::split(backup, "|"); + for (const auto& item : vec) { + result.push_back(COfPath::to_full(item)); + } + return result; +} diff --git a/client/file_oper.h b/client/file_oper.h new file mode 100644 index 0000000..70c51e6 --- /dev/null +++ b/client/file_oper.h @@ -0,0 +1,15 @@ +#include +#include +#include +#include +#include + +using namespace ofen; +class CFileOpr +{ +public: + CFileOpr(); + ~CFileOpr(); +public: + static std::vector get_file_list(const std::string& input); +}; \ No newline at end of file diff --git a/ofen b/ofen index 44fb416..45d22c9 160000 --- a/ofen +++ b/ofen @@ -1 +1 @@ -Subproject commit 44fb416ca166aa9c4ce270ff71bea6b3df703d6e +Subproject commit 45d22c927b796181c4b53f426a4d9dd6ebc4c0ce diff --git a/server/server.cpp b/server/server.cpp index 74a0126..90f0e5f 100644 --- a/server/server.cpp +++ b/server/server.cpp @@ -1,15 +1,25 @@ #include "server.h" +#include + +using namespace ofen; + +constexpr int g_ParseThreadNum = 1; +constexpr int g_SendThreadNum = 1; CTcpServer::CTcpServer(asio::io_context& io_context, const std::shared_ptr& logger) : io_context_(io_context), logger_(logger), acceptor_(io_context) { th_run_ = true; - handle_pool_ = std::make_shared(1); - send_pool_ = std::make_shared(1); + handle_pool_ = std::make_shared(g_ParseThreadNum); + send_pool_ = std::make_shared(g_SendThreadNum); handle_pool_->init(); send_pool_->init(); - handle_pool_->submit([&]() { handle_frame(); }); - send_pool_->submit([&]() { send_simple_buf(); }); + for (int i = 0; i < g_ParseThreadNum; ++i) { + handle_pool_->submit([&]() { handle_frame(); }); + } + for (int i = 0; i < g_ParseThreadNum; ++i) { + send_pool_->submit([&]() { send_simple_buf(); }); + } } CTcpServer::~CTcpServer() { @@ -47,12 +57,16 @@ void CTcpServer::stop() client_threads_.clear(); } -std::vector> CTcpServer::get_clients() +std::vector CTcpServer::get_clients() { - std::vector> result; + std::vector result; std::lock_guard lock(cli_mut_); for (const auto& item : client_map_) { - result.push_back(std::make_pair(item.first, item.second->task_)); + TaskList t; + t.id_ = item.first; + t.task_ = item.second->task_; + t.time_ = item.second->time_; + result.push_back(t); } return result; } @@ -64,11 +78,12 @@ SimpleBuffer* CTcpServer::get_client_list() auto vec = get_clients(); std::string msg; + int index = 1; for (const auto& item : vec) { - if (msg.empty()) { - msg.append(item.first + "," + item.second); - } else { - msg.append("|" + item.first + "," + item.second); + msg.append(fmt::format("[{}][{}][{}]", index, item.id_, item.time_)); + auto files = COfStr::split(item.task_, "|"); + for (const auto& file : files) { + msg.append("\n" + file); } } buf->data_ = new char[msg.size() + 1]; @@ -109,6 +124,7 @@ void CTcpServer::handle_frame() // 拿到该包后，要看转发给谁或者处理 if (buf->type_ == 199) { // 询问在线客户端 + logger_->info("GetList."); auto* sbuf = get_client_list(); if (sbuf == nullptr) { continue; @@ -116,6 +132,21 @@ void CTcpServer::handle_frame() sbuf->id_ = buf->id_; std::lock_guard lock(sbuf_mut_); scache_.push(sbuf); + } else if (buf->type_ == 197) { + logger_->info("UpList. {}", std::string(buf->data_, buf->len_)); + std::lock_guard lock(cli_mut_); + if (client_map_.count(buf->id_)) { + auto& cli = client_map_[buf->id_]; + cli->task_ = std::string(buf->data_, buf->len_); + cli->time_ = OfUtil::now_time(); + } + } else if (buf->type_ == 196) { + logger_->info("Cancle Task."); + std::lock_guard lock(cli_mut_); + if (client_map_.count(buf->id_)) { + auto& cli = client_map_[buf->id_]; + cli->task_.clear(); + } } delete buf; buf = nullptr; diff --git a/server/server.h b/server/server.h index 43f8f4f..fd123e7 100644 --- a/server/server.h +++ b/server/server.h @@ -12,6 +12,12 @@ struct ClientCache { CMutBuffer buffer_; std::array tmp_buf_; std::string task_; + std::string time_; +}; +struct TaskList { + std::string id_; + std::string task_; + std::string time_; }; class CTcpServer @@ -25,7 +31,7 @@ public: void stop(); private: - std::vector> get_clients(); + std::vector get_clients(); SimpleBuffer* get_client_list(); private: