taynpg/transm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 29 Wiki Activity

change：更改中转方式（暂未调通）。

Browse Source
This commit is contained in:
taynpg 2024-12-18 16:55:32 +08:00
parent 3a68984857
commit 2cdeb965c3
5 changed files with 126 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
client/client.cpp
View File

@ -174,6 +174,11 @@ bool CClient::down_one_file(const std::string& id, const std::string& file)
fs::path remote_file(ofen::COfPath::normalize(down_->cur_remote_file_)); fs::path remote_file(ofen::COfPath::normalize(down_->cur_remote_file_));
down_->cur_file_ = COfPath::to_full(remote_file.filename().string()); down_->cur_file_ = COfPath::to_full(remote_file.filename().string());
logger_->warn("Start Down => {} To {}", down_->cur_remote_file_, down_->cur_file_); logger_->warn("Start Down => {} To {}", down_->cur_remote_file_, down_->cur_file_);
down_->file_ = fopen(down_->cur_file_.c_str(), "wb");
if (down_->file_ == nullptr) {
logger_->error("Open {} Failed.", down_->cur_file_);
return false;
}
// 请求下载文件 // 请求下载文件
std::shared_ptr<CFrameBuffer> buf = std::make_shared<CFrameBuffer>(); std::shared_ptr<CFrameBuffer> buf = std::make_shared<CFrameBuffer>();
@ -290,15 +295,6 @@ void CClient::handle_frame(CFrameBuffer* buf)
} }
break; break;
} }
// （客户服务都不是指Server）如果是客户端啥也不干，服务端则开始发送数据
case TYPE_READY_TRANS: {
if (buf->mark_ != 0) {
break;
}
work_key_ = buf->fid_;
send_pool_->submit([&]() { send_file_data_th(); });
break;
}
// 能接收到 TRANS 一定是客户端（这里不是指Server） // 能接收到 TRANS 一定是客户端（这里不是指Server）
case TYPE_TRANS_FILE: { case TYPE_TRANS_FILE: {
auto ws = fwrite(buf->data_, 1, buf->len_, down_->file_); auto ws = fwrite(buf->data_, 1, buf->len_, down_->file_);
@ -308,34 +304,25 @@ void CClient::handle_frame(CFrameBuffer* buf)
break; break;
} }
case TYPE_OPEN_FILE: { case TYPE_OPEN_FILE: {
std::shared_ptr<TransInfomation> t = nullptr; char* keys = nullptr;
std::string id = buf->fid_; {
if (buf->mark_ == 0) {
std::lock_guard<std::mutex> lock(mutex_); std::lock_guard<std::mutex> lock(mutex_);
up_[buf->fid_] = std::make_shared<TransInfomation>(); up_[buf->fid_] = std::make_shared<TransInfomation>();
t = up_[buf->fid_];
#ifdef _WIN32 #ifdef _WIN32
t->cur_file_ = CCodec::u8ToGBK(std::string(buf->data_, buf->len_)); up_[buf->fid_]->cur_file_ = CCodec::u8ToGBK(std::string(buf->data_, buf->len_));
#else #else
t->cur_file_ = std::string(buf->data_, buf->len_); up_[buf->fid_]->cur_file_ = std::string(buf->data_, buf->len_);
#endif #endif
t->file_ = fopen(t->cur_file_.c_str(), "rb"); up_[buf->fid_]->file_ = fopen(up_[buf->fid_]->cur_file_.c_str(), "rb");
} else { if (up_[buf->fid_]->file_ == nullptr) {
t = down_; logger_->error("Ready Send File {} Open Failed.", up_[buf->fid_]->cur_file_);
t->file_ = fopen(t->cur_file_.c_str(), "wb");
}
if (t->file_ == nullptr) {
logger_->error("open file {} failed.", t->cur_file_);
report_trans_ret(TRANS_FAILED, buf->mark_ == 0 ? id : "");
break; break;
} }
std::shared_ptr<CFrameBuffer> tmp = std::make_shared<CFrameBuffer>(); keys = new char[512]();
tmp->type_ = TYPE_READY_TRANS; std::snprintf(keys, 512, "%s", buf->fid_.c_str());
tmp->mark_ = buf->mark_ == 0 ? 1 : 0; }
tmp->tid_ = buf->fid_; if (keys) {
if (!send_frame(tmp.get())) { send_pool_->submit([&]() { send_file_data_th(keys); });
logger_->error("TYPE_OPEN_FILE send ready failed.");
report_trans_ret(TRANS_FAILED, buf->mark_ == 0 ? id : "");
} }
break; break;
} }
@ -348,15 +335,27 @@ void CClient::handle_frame(CFrameBuffer* buf)
down_->trans_state_ = TRANS_DONE; down_->trans_state_ = TRANS_DONE;
break; break;
} }
case TYPE_OFFLINE: {
if (buf->mark_) {
report_trans_ret(TRANS_FAILED, buf->fid_);
break;
}
report_trans_ret(TRANS_FAILED);
break;
}
default: default:
break; break;
} }
} }
void CClient::send_file_data_th() void CClient::send_file_data_th(char* keys)
{ {
std::string str_key = work_key_; std::string str_key(keys);
std::shared_ptr<TransInfomation> t = nullptr; std::shared_ptr<TransInfomation> t = nullptr;
std::shared_ptr<int> deleter(new int(0), [&](int* p) {
delete p;
delete[] keys;
});
{ {
std::lock_guard<std::mutex> lock(mutex_); std::lock_guard<std::mutex> lock(mutex_);

12
client/client.h
View File

@ -1,12 +1,12 @@
#pragma once #pragma once
#include "file_oper.h"
#include <memory>
#include <mutex>
#include <net_base.h> #include <net_base.h>
#include <of_util.h>
#include <string>
#include <util.h> #include <util.h>
#include <vector> #include <vector>
#include <mutex>
#include <string>
#include "file_oper.h"
#include <of_util.h>
#include <memory>
using namespace ofen; using namespace ofen;
struct DownClientInfo { struct DownClientInfo {
@ -51,7 +51,7 @@ private:
private: private:
void handle_frame(CFrameBuffer* buf); void handle_frame(CFrameBuffer* buf);
void send_file_data_th(); void send_file_data_th(char* keys);
void hearts(); void hearts();
private: private:

102
server/server.cpp
View File

@ -4,21 +4,15 @@
using namespace ofen; using namespace ofen;
constexpr int g_MaxCacheLen = 1024 * 1024 * 50; constexpr int g_MaxCacheLen = 1024 * 1024 * 50;
constexpr int g_ParseThreadNum = 1;
CTcpServer::CTcpServer(asio::io_context& io_context, const std::shared_ptr<spdlog::logger>& logger) CTcpServer::CTcpServer(asio::io_context& io_context, const std::shared_ptr<spdlog::logger>& logger)
: io_context_(io_context), logger_(logger), acceptor_(io_context) : io_context_(io_context), logger_(logger), acceptor_(io_context)
{ {
th_run_ = true; th_run_ = true;
handle_pool_ = std::make_shared<CThreadPool>(g_ParseThreadNum);
handle_pool_->init();
for (int i = 0; i < g_ParseThreadNum; ++i) {
handle_pool_->submit([&]() { handle_frame(); });
}
} }
CTcpServer::~CTcpServer() CTcpServer::~CTcpServer()
{ {
th_run_ = false; th_run_ = false;
handle_pool_->close_wait_all();
} }
bool CTcpServer::start(unsigned short port) bool CTcpServer::start(unsigned short port)
@ -86,33 +80,28 @@ void CTcpServer::get_client_list(CFrameBuffer** buf)
tbuf->len_ = std::snprintf(tbuf->data_, msg.size() + 1, "%s", msg.data()); tbuf->len_ = std::snprintf(tbuf->data_, msg.size() + 1, "%s", msg.data());
} }
void CTcpServer::handle_frame() void CTcpServer::trans_data(CFrameBuffer* buf)
{ {
CFrameBuffer* buf = nullptr; std::shared_ptr<ClientCache> fcli = nullptr;
while (th_run_) { std::shared_ptr<ClientCache> tcli = nullptr;
{ {
std::lock_guard<std::mutex> lock(buf_mut_); std::lock_guard<std::mutex> lock(cli_mut_);
if (!cache_.empty()) { if (client_map_.count(buf->fid_)) {
buf = cache_.front(); fcli = client_map_[buf->fid_];
cache_.pop_front();
} }
if (client_map_.count(buf->tid_)) {
tcli = client_map_[buf->tid_];
} }
if (!buf) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
continue;
} }
switch (buf->type_) { switch (buf->type_) {
case TYPE_GET_LIST: { case TYPE_GET_LIST: {
logger_->info("[{}] GetList.", buf->fid_); logger_->info("[{}] GetList.", buf->fid_);
get_client_list(&buf); get_client_list(&buf);
std::lock_guard<std::mutex> lock(cli_mut_); if (fcli && !send_frame(fcli->socket_, buf)) {
if (client_map_.count(buf->fid_)) {
auto& cli = client_map_[buf->fid_];
if (!send_frame(cli->socket_, buf)) {
logger_->error("GetList send failed."); logger_->error("GetList send failed.");
} }
}
break; break;
} }
case TYPE_UP_LIST: { case TYPE_UP_LIST: {
@ -123,51 +112,26 @@ void CTcpServer::handle_frame()
std::string turn_files_path(files_path); std::string turn_files_path(files_path);
#endif #endif
logger_->info("[{}] UpList. {}", buf->fid_, turn_files_path); logger_->info("[{}] UpList. {}", buf->fid_, turn_files_path);
std::lock_guard<std::mutex> lock(cli_mut_); if (fcli) {
if (client_map_.count(buf->fid_)) { fcli->task_ = files_path;
auto& cli = client_map_[buf->fid_]; fcli->time_ = OfUtil::now_time();
cli->task_ = files_path;
cli->time_ = OfUtil::now_time();
} }
break; break;
} }
case TYPE_CANCEL_LIST: { case TYPE_CANCEL_LIST: {
logger_->info("[{}] Cancle Task.", buf->fid_); logger_->info("[{}] Cancle Task.", buf->fid_);
std::lock_guard<std::mutex> lock(cli_mut_); if (fcli) {
if (client_map_.count(buf->fid_)) { fcli->task_.clear();
auto& cli = client_map_[buf->fid_]; fcli->time_.clear();
cli->task_.clear();
} }
break; break;
} }
// 两边发送OPEN // 两边发送OPEN
case TYPE_OPEN_FILE: { case TYPE_OPEN_FILE:
std::lock_guard<std::mutex> lock(cli_mut_);
if (client_map_.count(buf->tid_)) {
auto& cli = client_map_[buf->tid_];
if (!send_frame(cli->socket_, buf)) {
logger_->error("[{}] turn tid_ ailed to {}", buf->fid_, buf->tid_);
}
}
if (client_map_.count(buf->fid_)) {
auto& cli = client_map_[buf->fid_];
buf->mark_ = 1;
buf->fid_ = buf->tid_;
if (!send_frame(cli->socket_, buf)) {
logger_->error("[{}] turn fid_ failed to {}", buf->fid_, buf->tid_);
}
}
break;
};
case TYPE_TRANS_DONE: case TYPE_TRANS_DONE:
case TYPE_READY_TRANS:
case TYPE_TRANS_FILE: { case TYPE_TRANS_FILE: {
std::lock_guard<std::mutex> lock(cli_mut_); if (check_double(fcli, tcli, buf) && !send_frame(tcli->socket_, buf)) {
if (client_map_.count(buf->tid_)) { logger_->error("Send from {} to {} failed Or One Offline.", buf->fid_, buf->tid_);
auto& cli = client_map_[buf->tid_];
if (!send_frame(cli->socket_, buf)) {
logger_->error("[{}] turn failed to {}", buf->fid_, buf->tid_);
}
} }
break; break;
} }
@ -175,9 +139,26 @@ void CTcpServer::handle_frame()
logger_->warn("No Mathched type."); logger_->warn("No Mathched type.");
break; break;
} }
delete buf;
buf = nullptr;
} }
bool CTcpServer::check_double(std::shared_ptr<ClientCache>& fcli, std::shared_ptr<ClientCache>& tcli,
CFrameBuffer* buf)
{
if (fcli == nullptr && tcli) {
buf->type_ = TYPE_OFFLINE;
send_frame(tcli->socket_, buf);
return false;
}
if (tcli == nullptr && fcli) {
std::swap(buf->fid_, buf->tid_);
buf->type_ = TYPE_OFFLINE;
send_frame(fcli->socket_, buf);
return false;
}
if (tcli == nullptr && fcli == nullptr) {
return false;
}
return true;
} }
void CTcpServer::accept_client() void CTcpServer::accept_client()
@ -246,8 +227,9 @@ void CTcpServer::th_client(std::shared_ptr<asio::ip::tcp::socket> socket, const
continue; continue;
} }
frame->fid_ = client_key; frame->fid_ = client_key;
std::lock_guard<std::mutex> lock(buf_mut_); // 直接转发，不加入缓存。
cache_.push_back(frame); trans_data(frame);
delete frame;
continue; continue;
} }
break; break;

8
server/server.h
View File

@ -1,7 +1,7 @@
#pragma once #pragma once
#include <list>
#include <net_base.h> #include <net_base.h>
#include <of_util.h> #include <of_util.h>
#include <list>
#include <string> #include <string>
#include <util.h> #include <util.h>
#include <vector> #include <vector>
@ -36,7 +36,9 @@ private:
void get_client_list(CFrameBuffer** buf); void get_client_list(CFrameBuffer** buf);
private: private:
void handle_frame(); void trans_data(CFrameBuffer* buf);
bool check_double(std::shared_ptr<ClientCache>& fcli, std::shared_ptr<ClientCache>& tcli,
CFrameBuffer* buf);
private: private:
void accept_client(); void accept_client();
@ -56,8 +58,6 @@ private:
std::map<std::string, std::shared_ptr<ClientCache>> client_map_; std::map<std::string, std::shared_ptr<ClientCache>> client_map_;
std::map<std::string, std::thread> client_threads_; std::map<std::string, std::thread> client_threads_;
std::mutex cli_mut_; std::mutex cli_mut_;
std::list<CFrameBuffer*> cache_;
std::mutex buf_mut_; std::mutex buf_mut_;
std::shared_ptr<CThreadPool> handle_pool_;
std::string server_ip_; std::string server_ip_;
}; };

4
util/util.h
View File

@ -16,11 +16,11 @@ enum FrameType : int16_t {
TYPE_OPEN_FILE, TYPE_OPEN_FILE,
TYPE_TRANS_FILE, TYPE_TRANS_FILE,
TYPE_TRANS_DONE, TYPE_TRANS_DONE,
TYPE_READY_TRANS,
TYPE_INTERRUPT, TYPE_INTERRUPT,
TYPE_NO_HIT_TASK, TYPE_NO_HIT_TASK,
TYPE_WAITTING, TYPE_WAITTING,
TYPE_HEARTS TYPE_HEARTS,
TYPE_OFFLINE
}; };
using namespace ofen; using namespace ofen;