#include "client.h" #include #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_ && 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; } } for (auto& item : ths_) { if (item.joinable()) { item.join(); } } if (update_list_th_.joinable()) { update_list_th_.join(); } if (th_down_active_.joinable()) { th_down_active_.join(); } if (hearts_.joinable()) { hearts_.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(); }); CFrameBuffer* bf = new CFrameBuffer(); bf->type_ = TYPE_GET_ID; send_frame(bf); delete bf; logger_->warn("SupportCmd:Get|Up|Down|Cancel|Update"); char line[512]{}; while (std::cin.getline(line, 512)) { if (!th_run_) { break; } std::string cmd_input(line); if (cmd_input == "end") { th_run_ = false; std::this_thread::sleep_for(std::chrono::milliseconds(10)); break; } if (cmd_input == "Get") { get_task_list(); continue; } if (cmd_input == "Cancel") { cancel_task(); continue; } auto vec = COfStr::split(cmd_input, " "); if (vec.size() == 3) { if (vec[0] == "Update") { request_update_list(vec[2], std::stoi(vec[1])); continue; } logger_->error("No matched cmd, May be param size incorrect."); } else if (vec.size() == 2) { if (vec[0] == "Down") { down_task(vec[1]); continue; } if (vec[0] == "Up") { up_task(cmd_input); continue; } logger_->error("No matched cmd, May be param size incorrect."); } else { logger_->error("No matched cmd, May be param size incorrect."); } } 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) { { std::lock_guard lock(mutex_); for (const auto& item : up_) { if (item.second->trans_state_ == TRANS_REDAY || item.second->trans_state_ == TRANS_ING) { logger_->warn("Have Task Upping, Please wait!"); return false; } } } 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::lock_guard lock(mutex_); for (const auto& item : up_) { if (item.second->trans_state_ == TRANS_REDAY || item.second->trans_state_ == TRANS_ING) { logger_->warn("Have Task Upping, Please wait!"); return false; } } } 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, const std::string& local_dir) { 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_)); if (local_dir.empty()) { down_->cur_file_ = COfPath::to_full(remote_file.filename().string()); } else { down_->cur_file_ = fs::path(local_dir).append(remote_file.filename().string()).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__, down_->cur_remote_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(); } /* 清单文件,内容格式为: D:/a.txt|/home/zhangsan/ C:/Dijava|/home/zhangsan/dia 功能为,请求某个客户端,更新我所列出的文件,右侧是远端需要存储的目录(必须存在,不存在则不理会) */ bool CClient::request_update_list(const std::string& list_file, int index) { if (downloading_) { logger_->warn("Have Task Downloading, Please wait....."); return false; } if (!task_list_.count(index)) { logger_->error("No Index Found {}.", index); return false; } // 读取list文件 std::ifstream in(COfPath::to_full(list_file)); if (!in.is_open()) { logger_->error("Can't Open File:{}", COfPath::to_full(list_file)); return false; } std::istreambuf_iterator iterf(in); std::istreambuf_iterator iter; std::string content(iterf, iter); in.close(); // 校验格式是否正确 auto vec = COfStr::split(content, "\n"); bool valid = true; std::string handled_content; for (const auto& item : vec) { std::string hitem = COfStr::trim(item); auto v = COfStr::split(hitem, "|"); if (v.size() >= 2) { handled_content.append(hitem + "\n"); continue; } valid = false; break; } if (!valid) { logger_->error("Judge List File {} Format Not Passed.", list_file); return false; } list_file_ = list_file; std::shared_ptr buf = std::make_shared(); buf->type_ = TYPE_REQUEST_UPDATE_LIST; buf->data_ = new char[handled_content.size() + 1](); buf->len_ = std::snprintf(buf->data_, handled_content.size() + 1, "%s", handled_content.c_str()); buf->tid_ = task_list_[index]->id; if (!send_frame(buf.get())) { logger_->error("Send Failed {}", __LINE__); return false; } return true; } bool CClient::check_update_list(const std::string& content, std::map& files) { auto vec = COfStr::split(content, "\n"); bool valid = true; for (const auto& item : vec) { if (item.empty()) { continue; } auto vi = COfStr::split(item, "|"); if (vi.size() != 2) { logger_->error("Size not 2 {}", item); valid = true; break; } if (!fs::exists(vi[1])) { valid = false; logger_->error("Not exist {}", vi[1]); break; } files[vi[0]] = vi[1]; } return valid; } bool CClient::down_update_file(std::map files) { std::shared_ptr buf = std::make_shared(); buf->tid_ = list_serve_id_; down_ = std::make_shared(); bool suc = true; for (const auto& item : files) { if (!down_one_file(list_serve_id_, item.first, item.second)) { suc = false; break; } } if (suc) { buf->type_ = TYPE_DONE_UPDATE_LIST; logger_->info("Do Task From {} Done!", buf->tid_); } else { buf->type_ = TYPE_FAILED_UPDATE_LIST; logger_->info("Do Task From {} Failed!", buf->tid_); } send_frame(buf.get()); return suc; } 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_ID: { logger_->debug("Your ID:{}", buf->tid_); break; } 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_); report_trans_ret(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 (downloading_ && !down_->cur_remote_file_.empty()) { logger_->warn("Stop Down {} From {}.", down_->cur_remote_file_, buf->fid_); } report_trans_ret(TRANS_FAILED); break; } case TYPE_REQUEST_UPDATE_LIST: { std::string content(buf->data_, buf->len_); std::map files; if (check_update_list(content, files)) { update_list_content_ = content; buf->type_ = TYPE_CONFIRM_UPDATE_LIST; } else { buf->type_ = TYPE_UNCONFIRM_UPDATE_LIST; } std::swap(buf->tid_, buf->fid_); if (!send_frame(buf)) { logger_->error("Send Failed {}.", __LINE__); break; } list_serve_id_ = buf->tid_; logger_->debug("Do Task From {}.", buf->tid_); if (update_list_th_.joinable()) { update_list_th_.join(); } update_list_th_ = std::thread([this, files]() { down_update_file(files); }); break; } case TYPE_CONFIRM_UPDATE_LIST: { logger_->info("remote {} check {} passed!", buf->fid_, list_file_); break; } case TYPE_UNCONFIRM_UPDATE_LIST: { logger_->error("remote {} check {} not passed!", buf->fid_, list_file_); break; } case TYPE_DONE_UPDATE_LIST: { logger_->info("remote {} do task {} success!", buf->fid_, list_file_); break; } case TYPE_FAILED_UPDATE_LIST: { logger_->info("remote {} do task {} failed!", buf->fid_, list_file_); 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 (th_run_ && !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()); } }