#include "client.h" #include #include #include #include #include namespace fs = std::filesystem; CClient::CClient(const std::shared_ptr& logger) : logger_(logger) { client_ = std::make_shared(io_context_, logger_); supported_.push_back("Get"); sleep_.set_timeout(2000); } CClient::~CClient() { th_run_ = false; sleep_.contiune(); if (down_->file_) { fclose(down_->file_); down_->file_ = nullptr; } std::lock_guard lock(mutex_); for (const auto& item : up_) { if (item.second->file_) { fclose(item.second->file_); item.second->file_ = nullptr; } } if (hearts_.joinable()) { hearts_.join(); } for (auto& item : ths_) { if (item.joinable()) { item.join(); } } } void CClient::run(const std::string& ip, const std::string& port) { th_run_ = true; if (!client_->connect(ip, port)) { logger_->info("{} connect err.", __FUNCTION__); return; } client_->register_func([&](CFrameBuffer* buf) { handle_frame(buf); }); client_->async_recv(); hearts_ = std::thread([&]() { hearts(); }); std::thread thread([&]() { io_context_.run(); }); logger_->warn("SupportCmd:Get|Up|Down|Cancel"); char line[512]{}; while (std::cin.getline(line, 512)) { if (!th_run_) { break; } std::string cmd_input(line); if (cmd_input == "end") { break; } auto vec = COfStr::split(cmd_input, " "); if (vec.size() < 1) { logger_->error("input's invalid format."); continue; } std::string cmd{}; std::string param{}; if (vec.size() == 1) { cmd = vec[0]; } else { cmd = vec[0]; param = vec[1]; } if (cmd == "Get") { get_task_list(); continue; } if (cmd == "Down") { down_task(vec[1]); continue; } if (cmd == "Up") { up_task(cmd_input); continue; } if (cmd == "Cancel") { cancel_task(); continue; } logger_->error("No matched cmd."); } client_->disconnect(); thread.join(); logger_->info("{} exit.", __FUNCTION__); } bool CClient::get_task_list() { std::shared_ptr buf = std::make_shared(); buf->type_ = TYPE_GET_LIST; return send_frame(buf.get()); } bool CClient::down_task(const std::string& param) { if (downloading_) { logger_->warn("Have Task Downloading, Please wait....."); return false; } int id = std::stoi(param); if (!task_list_.count(id)) { logger_->error("No matched id[{}] in task list.", id); return false; } down_ = std::make_shared(); const auto& vec = task_list_[id]->files; // 开始传输文件 for (const auto& item : vec) { if (!down_one_file(task_list_[id]->id, item)) { break; } } return true; } bool CClient::up_task(const std::string& cmd) { auto list = CFileOpr::get_file_list(cmd); std::string msg; for (const auto& item : list) { if (!fs::exists(item)) { logger_->error("File {} not exist, please check!", item); return false; } if (msg.empty()) { msg.append(item); } else { msg.append("|" + item); } } if (msg.empty()) { logger_->warn("{} msg empty.", __FUNCTION__); return false; } #ifdef _WIN32 msg = CCodec::GBKTou8(msg); #endif std::shared_ptr buf = std::make_shared(); buf->type_ = TYPE_UP_LIST; buf->data_ = new char[msg.size() + 1]; std::memset(buf->data_, 0x0, msg.size() + 1); buf->len_ = std::snprintf(buf->data_, msg.size() + 1, "%s", msg.data()); return send_frame(buf.get()); } bool CClient::cancel_task() { std::shared_ptr buf = std::make_shared(); buf->type_ = TYPE_CANCEL_LIST; return send_frame(buf.get()); } bool CClient::down_one_file(const std::string& id, const std::string& file) { down_->cur_remote_id_ = id; #ifdef _WIN32 down_->cur_remote_file_ = CCodec::u8ToGBK(file); #else down_->cur_remote_file_ = file; #endif fs::path remote_file(ofen::COfPath::normalize(down_->cur_remote_file_)); down_->cur_file_ = COfPath::to_full(remote_file.filename().string()); 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 buf = std::make_shared(); buf->type_ = TYPE_OPEN_FILE; buf->tid_ = id; buf->data_ = new char[file.size() + 1]; buf->len_ = std::snprintf(buf->data_, file.size() + 1, "%s", file.data()); if (!send_frame(buf.get())) { logger_->error("{} request open file [{}] send failed.", __FUNCTION__, file); down_->cur_remote_id_.clear(); down_->cur_remote_file_.clear(); return false; } down_->trans_state_ = TRANS_REDAY; while (down_->trans_state_ != TRANS_DONE) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); if (!th_run_) { logger_->error("Interrup When Receive File."); report_trans_ret(TRANS_FAILED); return false; } } return true; } void CClient::report_trans_ret(TransState state, const std::string& key) { std::shared_ptr t = nullptr; if (key.empty()) { t = down_; downloading_ = false; if (th_down_active_.joinable()) { th_down_active_.join(); } } else { std::lock_guard lock(mutex_); if (up_.count(key)) { t = up_[key]; } } if (t == nullptr) { return; } t->trans_state_ = state; if (t->file_) { fclose(t->file_); t->file_ = nullptr; if (key.empty() && t->trans_state_ == TRANS_FAILED) { fs::remove(t->cur_file_); } } t->cur_remote_file_.clear(); t->cur_remote_id_.clear(); } bool CClient::send_frame(CFrameBuffer* buf) { char* out_buf{}; int out_len{}; if (!CTransProtocal::pack(buf, &out_buf, out_len)) { logger_->error("{} pack failed.", __FUNCTION__); return false; } std::lock_guard lock(send_mut_); if (!client_->send(out_buf, out_len)) { delete[] out_buf; return false; } delete[] out_buf; return true; } void CClient::handle_frame(CFrameBuffer* buf) { if (buf == nullptr) { logger_->error("{} nullptr.", __FUNCTION__); return; } switch (buf->type_) { case TYPE_GET_LIST: { task_list_.clear(); std::string source(buf->data_, buf->len_); auto vec = COfStr::split(source, "\n"); int index = -1; for (const auto& item : vec) { std::string real = COfStr::trim(item); if (real.empty()) { continue; } if (real.find("[") == std::string::npos) { #ifdef _WIN32 logger_->info("FILE ==> {}", CCodec::u8ToGBK(real)); #else logger_->info("FILE ==> {}", real); #endif task_list_[index]->files.push_back(real); } else { auto a = real.find_first_of("[") + 1; auto b = real.find_first_of("]"); std::string str_index = real.substr(a, b - a); index = std::stoi(str_index); std::string backup = real; backup.erase(0, b + 1); auto aa = backup.find_first_of("[") + 1; auto bb = backup.find_first_of("]"); std::string id = backup.substr(aa, bb - aa); if (!task_list_.count(index)) { task_list_[index] = std::make_shared(); task_list_[index]->id = id; } logger_->debug("***********************************************"); logger_->info("{}", real); } } break; } // 能接收到 TRANS 一定是客户端(这里不是指Server) case TYPE_TRANS_FILE: { if (!downloading_) { downloading_ = true; th_down_active_ = std::thread([&]() { judget_down_active(); }); } auto ws = fwrite(buf->data_, 1, buf->len_, down_->file_); if (static_cast(ws) != buf->len_) { logger_->warn("no matched write and data."); } break; } case TYPE_OPEN_FILE: { std::string keys{}; { std::lock_guard lock(mutex_); up_[buf->fid_] = std::make_shared(); #ifdef _WIN32 up_[buf->fid_]->cur_file_ = CCodec::u8ToGBK(std::string(buf->data_, buf->len_)); #else up_[buf->fid_]->cur_file_ = std::string(buf->data_, buf->len_); #endif up_[buf->fid_]->file_ = fopen(up_[buf->fid_]->cur_file_.c_str(), "rb"); up_[buf->fid_]->trans_state_ = TRANS_REDAY; if (up_[buf->fid_]->file_ == nullptr) { logger_->error("Ready Send File {} Open Failed.", up_[buf->fid_]->cur_file_); break; } keys = buf->fid_; } if (!keys.empty()) { ths_.emplace_back(std::thread([this, keys]() { send_file_data_th(keys.c_str()); })); } break; } case TYPE_TRANS_DONE: { logger_->warn("Trans done, close file {}.", down_->cur_file_); if (down_->file_) { fclose(down_->file_); down_->file_ = nullptr; } down_->trans_state_ = TRANS_DONE; break; } case TYPE_OFFLINE: { if (buf->mark_) { std::lock_guard lock(mutex_); if (!up_.count(buf->fid_)) { logger_->warn("Offline no match."); break; } auto t = up_[buf->fid_]; t->trans_state_ = TRANS_BREAK; break; } if (!down_->cur_remote_file_.empty()) { logger_->warn("Stop Down {} From {}.", down_->cur_remote_file_, buf->fid_); } report_trans_ret(TRANS_FAILED); break; } default: break; } } void CClient::send_file_data_th(const char* keys) { std::string str_key(keys); std::shared_ptr t = nullptr; { std::lock_guard lock(mutex_); if (!up_.count(str_key)) { logger_->error("{} no matched key.", __FUNCTION__); return; } t = up_[str_key]; } logger_->info("Start Trans File {} To {}", t->cur_file_, str_key); std::shared_ptr buf = std::make_shared(); buf->type_ = TYPE_TRANS_FILE; buf->tid_ = str_key; buf->data_ = new char[g_BuffSize]{}; buf->mark_ = 1; while (!feof(t->file_)) { if (t->trans_state_ == TRANS_BREAK) { logger_->warn("Stop Trans {} To {} failed.", t->cur_file_, str_key); report_trans_ret(TRANS_FAILED, str_key); return; } buf->len_ = fread(buf->data_, 1, g_BuffSize, t->file_); if (!send_frame(buf.get())) { report_trans_ret(TRANS_FAILED, str_key); logger_->error("Stop Trans {} To {} failed.", t->cur_file_, str_key); return; } //std::this_thread::sleep_for(std::chrono::milliseconds(10)); } buf->type_ = TYPE_TRANS_DONE; if (!send_frame(buf.get())) { logger_->error("send_file_data_th send DONE failed."); } report_trans_ret(TRANS_DONE, str_key); logger_->debug("Trans File {} To {} Done !!!", t->cur_file_, str_key); } void CClient::hearts() { std::shared_ptr buf = std::make_shared(); buf->type_ = TYPE_HEARTS; while (th_run_) { sleep_.sleep(); if (!send_frame(buf.get())) { logger_->error("{} send failed.", __FUNCTION__); th_run_ = false; } } } void CClient::judget_down_active() { std::shared_ptr buf = std::make_shared(); buf->type_ = TYPE_JUDGE_ACTIVE; buf->tid_ = down_->cur_remote_id_; while (downloading_ && th_run_) { std::this_thread::sleep_for(std::chrono::milliseconds(2000)); send_frame(buf.get()); } }