diff --git a/apps/android/native/file_transfer_session.h b/apps/android/native/file_transfer_session.h index 950e7140..9aeea47b 100644 --- a/apps/android/native/file_transfer_session.h +++ b/apps/android/native/file_transfer_session.h @@ -7,265 +7,9 @@ #ifndef _FILE_TRANSFER_SESSION_H_ #define _FILE_TRANSFER_SESSION_H_ -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - +#include "../../../libs/controller/include/mobile_file_transfer_session.h" namespace crossdesk::android_controller { +using mobile_controller::FileTransferSession; +} -// File I/O is bounded to one chunk per pump. Feedback is delivered by MiniRTC -// callback threads; no callback waits for the RTC executor or for a remote ACK. -class FileTransferSession { - public: - using Clock = std::chrono::steady_clock; - using Send = std::function; - using Progress = std::function; - FileTransferSession(Send send, Progress progress) - : send_(std::move(send)), progress_(std::move(progress)) {} - ~FileTransferSession() { - if (outgoing_) { - outgoing_->input.close(); - std::error_code ec; - std::filesystem::remove(outgoing_->path, ec); - } - for (auto& entry : incoming_) Remove(entry.second); - } - - void SetDirectory(std::filesystem::path directory) { - std::lock_guard lock(mutex_); - directory_ = std::move(directory); - } - - bool Start(const std::filesystem::path& path, const std::string& name, - Clock::time_point now = Clock::now()) { - std::lock_guard lock(mutex_); - if (outgoing_ || name.empty() || name.size() > UINT16_MAX) return false; - std::error_code ec; - if (!std::filesystem::is_regular_file(path, ec)) return false; - const auto size = std::filesystem::file_size(path, ec); - if (ec) return false; - Outgoing file; - file.input.open(path, std::ios::binary); - if (!file.input) return false; - file.id = ++next_id_; - file.name = name; - file.path = path; - file.total = size; - file.updated = now; - outgoing_ = std::move(file); - progress_(name, 0, true, ""); - return true; - } - - void Pump(Clock::time_point now = Clock::now()) { - std::lock_guard lock(mutex_); - for (auto it = incoming_.begin(); it != incoming_.end();) { - if (now - it->second.updated < timeout_) { ++it; continue; } - Reject(it->first, it->second.total, it->second.received); - progress_(it->second.name, -1, false, ""); - Remove(it->second); - it = incoming_.erase(it); - } - if (!outgoing_) return; - auto& file = *outgoing_; - if (now - file.updated >= timeout_) { FinishSend(false); return; } - if (file.last || file.sent - file.acked >= 1024 * 1024) return; - std::array payload{}; - const auto size = static_cast(std::min(payload.size(), file.total - file.sent)); - if (size) { - file.input.read(payload.data(), size); - if (file.input.gcount() != size) { FinishSend(false); return; } - } - const bool first = file.sent == 0; - const bool last = file.sent + size == file.total; - const auto chunk = EncodeFileChunk(file.id, file.sent, file.total, - payload.data(), size, first ? &file.name : nullptr, first, last); - if (chunk.empty() || send_(kFileStream, chunk.data(), chunk.size()) != 0) { - FinishSend(false); return; - } - file.sent += size; - file.last = last; - } - - void Ack(const char* data, size_t size, Clock::time_point now = Clock::now()) { - FileTransferAck ack{}; - if (!DecodeFileTransferAck(data, size, &ack)) return; - std::lock_guard lock(mutex_); - if (!outgoing_ || ack.file_id != outgoing_->id) return; - auto& file = *outgoing_; - if (ack.total_size != file.total || ack.acked_offset > file.sent || - (ack.flags & ~3u)) return; - if (ack.flags & 2) { FinishSend(false); return; } - if (ack.flags & 1) { - if (file.last && ack.acked_offset == file.total) FinishSend(true); - return; - } - if (ack.acked_offset <= file.acked) return; - file.acked = ack.acked_offset; - file.updated = now; - Report(file.name, file.acked, file.total, true, file.percent); - } - - void Receive(const char* data, size_t size, Clock::time_point now = Clock::now()) { - FileChunkView chunk; - if (!DecodeFileChunk(data, size, &chunk)) return; - const auto& h = chunk.header; - std::lock_guard lock(mutex_); - auto it = incoming_.find(h.file_id); - const bool first = h.flags & 1, last = h.flags & 2; - const bool invalid = (h.flags & ~3u) || h.total_size > INT64_MAX || - (first && h.offset != 0) || chunk.payload_size > kFileChunkSize || - size != sizeof(FileChunkHeader) + h.name_len + chunk.payload_size || - (!first && h.name_len) || (!chunk.payload_size && !last) || - last != (h.offset + chunk.payload_size == h.total_size); - if (invalid || (it == incoming_.end() && incoming_.size() >= 4)) { - Reject(h.file_id, h.total_size, h.offset); - if (it != incoming_.end()) { progress_(it->second.name, -1, false, ""); Remove(it->second); incoming_.erase(it); } - return; - } - if (it == incoming_.end()) { - Incoming file; - file.name = SafeName(chunk.file_name); - file.total = h.total_size; - std::error_code ec; - if (!directory_.empty()) std::filesystem::create_directories(directory_, ec); - // A per-transfer directory avoids overwrites, even for duplicate names. - auto folder = directory_ / (std::to_string(h.file_id) + "-" + std::to_string(++receive_id_)); - if (directory_.empty() || ec || !std::filesystem::create_directory(folder, ec)) { - Reject(h.file_id, h.total_size, 0); progress_(file.name, -1, false, ""); return; - } - file.path = folder / file.name; - file.partial = folder / ".partial"; - file.output.open(file.partial, std::ios::binary | std::ios::trunc); - if (!file.output) { - Reject(h.file_id, h.total_size, 0); progress_(file.name, -1, false, ""); Remove(file); return; - } - it = incoming_.emplace(h.file_id, std::move(file)).first; - } - auto& file = it->second; - if (h.total_size != file.total || (first && file.first) || - !AddRange(file, h.offset, h.offset + chunk.payload_size)) { - Reject(h.file_id, h.total_size, h.offset); - progress_(file.name, -1, false, ""); Remove(file); incoming_.erase(it); return; - } - if (first) { - file.first = true; - file.name = SafeName(chunk.file_name); - file.path = file.path.parent_path() / file.name; - } - file.last |= last; - // KCP removes messages in order, but delivers batches outside its lock on - // both RTP and timer threads. Track coverage instead of assuming callbacks - // arrive in order; ACK only the contiguous prefix and never expose holes. - file.received = !file.ranges.empty() && file.ranges.begin()->first == 0 - ? file.ranges.begin()->second : 0; - const bool complete = file.first && file.last && file.received == file.total; - file.output.seekp(static_cast(h.offset)); - file.output.write(chunk.payload, chunk.payload_size); - if (complete) { file.output.flush(); file.output.close(); } - std::error_code ec; - if (complete && !file.output.fail()) std::filesystem::rename(file.partial, file.path, ec); - if (file.output.fail() || ec) { - Reject(h.file_id, h.total_size, h.offset); - progress_(file.name, -1, false, ""); Remove(file); incoming_.erase(it); return; - } - file.updated = now; - const FileTransferAck ack{kFileAckMagic, h.file_id, file.received, file.total, complete ? 1u : 0u}; - const auto packet = EncodeFileTransferAck(ack); - send_(kFileFeedbackStream, packet.data(), packet.size()); - if (complete) { - progress_(file.name, 1, false, file.path.string()); - incoming_.erase(it); - } else Report(file.name, file.received, file.total, false, file.percent); - } - - bool Sending() { - std::lock_guard lock(mutex_); - return outgoing_.has_value(); - } - - private: - struct Outgoing { - uint32_t id = 0; - std::string name; - std::filesystem::path path; - std::ifstream input; - uint64_t total = 0, sent = 0, acked = 0; - bool last = false; - int percent = -1; - Clock::time_point updated; - }; - struct Incoming { - std::string name; - std::filesystem::path path, partial; - std::ofstream output; - uint64_t total = 0, received = 0; - bool first = false, last = false; - std::map ranges; - int percent = -1; - Clock::time_point updated; - }; - static bool AddRange(Incoming& file, uint64_t start, uint64_t end) { - if (start == end) return true; - auto next = file.ranges.lower_bound(start); - if (next != file.ranges.end() && next->first < end) return false; - if (next != file.ranges.begin()) { - auto previous = std::prev(next); - if (previous->second > start) return false; - if (previous->second == start) { start = previous->first; file.ranges.erase(previous); } - } - if (next != file.ranges.end() && next->first == end) { end = next->second; file.ranges.erase(next); } - file.ranges.emplace(start, end); - return file.ranges.size() <= 1024; - } - static std::string SafeName(std::string name) { - const auto slash = name.find_last_of("/\\"); - if (slash != std::string::npos) name.erase(0, slash + 1); - for (auto& ch : name) if (static_cast(ch) < 32 || ch == ':') ch = '_'; - if (name.empty() || name == "." || name == ".." || name == ".partial" || name.size() > 240) return "received_file"; - return name; - } - static void Remove(Incoming& file) { - file.output.close(); - std::error_code ec; - std::filesystem::remove(file.partial, ec); - std::filesystem::remove(file.path.parent_path(), ec); - } - void Reject(uint32_t id, uint64_t total, uint64_t offset) { - const auto packet = EncodeFileTransferAck({kFileAckMagic, id, offset, total, 2}); - send_(kFileFeedbackStream, packet.data(), packet.size()); - } - void Report(const std::string& name, uint64_t offset, uint64_t total, bool sending, int& previous) { - const double progress = total ? static_cast(offset) / total : 0; - const int percent = static_cast(progress * 100); - if (percent != previous) { previous = percent; progress_(name, std::min(progress, .99), sending, ""); } - } - void FinishSend(bool success) { - progress_(outgoing_->name, success ? 1 : -1, true, ""); - outgoing_->input.close(); - std::error_code ec; - std::filesystem::remove(outgoing_->path, ec); - outgoing_.reset(); - } - const std::chrono::seconds timeout_{30}; - std::mutex mutex_; - Send send_; - Progress progress_; - std::filesystem::path directory_; - std::optional outgoing_; - std::map incoming_; - uint32_t next_id_ = 0; - uint64_t receive_id_ = 0; -}; -} // namespace crossdesk::android_controller - -#endif \ No newline at end of file +#endif diff --git a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp index 628f4ff2..56fff86b 100644 --- a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp +++ b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp @@ -1,370 +1,290 @@ #include "features/file_transfer/file_transfer_manager.h" #include -#include -#include #include -#include -#include -#include -#include #include -#include -#include "file_transfer.h" #include "filesystem_utf8.h" #include "rd_log.h" #include "runtime/gui_runtime.h" namespace crossdesk { -FileTransferManager::FileTransferManager(GuiRuntime &owner) : owner_(owner) {} +FileTransferManager::FileTransferManager(GuiRuntime& owner) : owner_(owner) {} +FileTransferManager::~FileTransferManager() { Shutdown(); } -FileTransferManager::~FileTransferManager() { +void FileTransferManager::Shutdown() { std::vector> workers; { std::lock_guard lock(workers_mutex_); stopping_ = true; workers.swap(workers_); } + { + std::lock_guard lock(sessions_mutex_); + for (auto& [id, sender] : senders_) sender.session->Cancel(); + for (auto& [handler, peers] : receivers_) + for (auto& [id, receiver] : peers) receiver.session->Cancel(); + receivers_.clear(); + } + wake_.notify_all(); for (auto& worker : workers) worker.wait(); } -FileTransferManager::FileTransferState &FileTransferManager::global_state() { - return global_state_; -} - -FileTransferManager::FileTransferState &FileTransferManager::state_for( - const std::shared_ptr &props) { +FileTransferManager::FileTransferState& FileTransferManager::global_state() { return global_state_; } +FileTransferManager::FileTransferState& FileTransferManager::state_for(const std::shared_ptr& props) { return props ? props->file_transfer_ : global_state_; } -void FileTransferManager::ProcessSelectedFile( - const std::string &path, - const std::shared_ptr &props, - const std::string &file_label, const std::string &remote_id) { - if (path.empty()) { - return; - } - - FileTransferState &state = state_for(props); - LOG_INFO("Selected file: {}", path.c_str()); - - const std::filesystem::path file_path = PathFromUtf8(path); +void FileTransferManager::ProcessSelectedFile(const std::string& path, + const std::shared_ptr& props, const std::string& file_label, + const std::string& remote_id) { + if (path.empty() || stopping_ || (props && props->closing_)) return; + const auto file_path = PathFromUtf8(path); std::error_code ec; - if (!std::filesystem::is_regular_file(file_path, ec)) { - LOG_ERROR("Selected path is not a regular file: {}", path); - return; - } - const uint64_t file_size = std::filesystem::file_size(file_path, ec); - if (ec) { - LOG_ERROR("Failed to get file size: {}", ec.message()); - return; - } - + if (!std::filesystem::is_regular_file(file_path, ec)) return; + const auto size = std::filesystem::file_size(file_path, ec); + if (ec) return; + auto& state = state_for(props); { - std::lock_guard lock(state.file_transfer_list_mutex_); + std::lock_guard lock(state.file_transfer_list_mutex_); FileTransferState::FileTransferInfo info; - const auto utf8_name = file_path.filename().u8string(); - info.file_name.assign(reinterpret_cast(utf8_name.data()), - utf8_name.size()); - info.file_path = file_path; - info.file_size = file_size; + const auto name = file_path.filename().u8string(); + info.file_name.assign(name.begin(), name.end()); + info.file_path = file_path; info.file_size = size; info.status = FileTransferState::FileTransferStatus::Queued; state.file_transfer_list_.push_back(std::move(info)); } state.file_transfer_window_visible_ = true; - - auto enqueue = [&]() { - size_t queue_size = 0; - { - std::lock_guard lock(state.file_queue_mutex_); - state.file_send_queue_.push( - FileTransferState::QueuedFile{file_path, file_label, remote_id}); - queue_size = state.file_send_queue_.size(); - } - LOG_INFO("File added to queue: {} ({} files in queue)", - file_path.filename().string().c_str(), queue_size); - }; - - if (state.file_sending_.load()) { - enqueue(); - return; - } - - Start(props, file_path, file_label, remote_id); - if (!state.file_sending_.load()) { - enqueue(); - } -} - -void FileTransferManager::Start( - std::shared_ptr props, - const std::filesystem::path &file_path, const std::string &file_label, - const std::string &remote_id) { - if (props && props->closing_) return; - const bool is_global = !props; - PeerPtr* peer = nullptr; - std::shared_ptr peer_events; { - std::shared_lock lock(owner_.remote_sessions_mutex_); - if (props && props->closing_) return; - peer = is_global ? owner_.peer_ : props->peer_; - peer_events = is_global ? owner_.peer_events_ : props->peer_events_; + std::lock_guard lock(state.file_queue_mutex_); + state.file_send_queue_.push({file_path, file_label, props ? props->remote_id_ : remote_id}); } - if (!peer) { - LOG_ERROR("StartFileTransfer: invalid peer"); - return; - } - - FileTransferState &initial_state = state_for(props); - bool expected = false; - if (!initial_state.file_sending_.compare_exchange_strong(expected, true)) { - LOG_WARN("StartFileTransfer called while another file is active: {}", - file_path.filename().string().c_str()); - return; - } - - const auto props_weak = std::weak_ptr(props); - std::lock_guard workers_lock(workers_mutex_); - if (stopping_) return; - // Reap finished senders without blocking the UI. - std::erase_if(workers_, [](auto& worker) { - return worker.wait_for(std::chrono::seconds(0)) == std::future_status::ready; - }); - workers_.push_back(std::async(std::launch::async, - [this, peer, peer_events, file_path, file_label, props_weak, remote_id, - is_global]() { - auto props_locked = props_weak.lock(); - FileTransferState *state = nullptr; - if (props_locked) { - state = &props_locked->file_transfer_; - } else if (is_global) { - state = &global_state_; - } else { - return; - } - - std::error_code ec; - const uint64_t total_size = std::filesystem::file_size(file_path, ec); - if (ec) { - LOG_ERROR("Failed to get file size: {}", ec.message().c_str()); - state->file_sending_ = false; - return; - } - - state->file_sent_bytes_ = 0; - state->file_total_bytes_ = total_size; - state->file_send_rate_bps_ = 0; - state->file_transfer_window_visible_ = true; - { - std::lock_guard lock(state->file_transfer_mutex_); - state->file_send_start_time_ = std::chrono::steady_clock::now(); - state->file_send_last_update_time_ = state->file_send_start_time_; - state->file_send_last_bytes_ = 0; - } - - FileSender sender; - const uint32_t file_id = FileSender::NextFileId(); - if (props_locked) { - std::lock_guard lock(file_id_to_props_mutex_); - file_id_to_props_[file_id] = props_weak; - } else { - std::lock_guard lock(file_id_to_state_mutex_); - file_id_to_state_[file_id] = state; - } - state->current_file_id_ = file_id; - - { - std::lock_guard lock(state->file_transfer_list_mutex_); - for (auto &info : state->file_transfer_list_) { - if (info.file_path == file_path && - info.status == FileTransferState::FileTransferStatus::Queued) { - info.status = FileTransferState::FileTransferStatus::Sending; - info.file_id = file_id; - info.file_size = total_size; - info.sent_bytes = 0; - break; - } - } - } - - const int ret = sender.SendFile( - file_path, file_path.filename().string(), - [peer, peer_events, file_label, remote_id, props_locked](const char *buffer, size_t size) { - // Close drains each in-flight send before detaching the peer. - if (props_locked && props_locked->closing_) return -1; - auto callback = peer_events->EnterCallback(); - if (!callback) return -1; - if (remote_id.empty()) { - return SendReliableDataFrame(peer, buffer, size, - file_label.c_str()); - } - return SendReliableDataFrameToPeer( - peer, buffer, size, file_label.c_str(), remote_id.c_str(), - remote_id.size()); - }, - 64 * 1024, file_id); - - if (ret == 0) { - return; - } - - state->file_sending_ = false; - state->file_transfer_window_visible_ = false; - state->file_sent_bytes_ = 0; - state->file_total_bytes_ = 0; - state->file_send_rate_bps_ = 0; - state->current_file_id_ = 0; - Unregister(file_id, props_locked != nullptr); - - { - std::lock_guard lock(state->file_transfer_list_mutex_); - for (auto &info : state->file_transfer_list_) { - if (info.file_id == file_id) { - info.status = FileTransferState::FileTransferStatus::Failed; - break; - } - } - } - LOG_ERROR("FileSender::SendFile failed for [{}], ret={}", - file_path.string().c_str(), ret); - if (peer_events->IsActive()) ProcessQueue(props_locked); - })); -} - -void FileTransferManager::ProcessQueue( - std::shared_ptr props) { - if (props && props->closing_) return; - FileTransferState &state = state_for(props); - if (state.file_sending_.load()) { - return; - } - - FileTransferState::QueuedFile queued_file; - { - std::lock_guard lock(state.file_queue_mutex_); - if (state.file_send_queue_.empty()) { - return; - } - queued_file = state.file_send_queue_.front(); - state.file_send_queue_.pop(); - } - Start(props, queued_file.file_path, queued_file.file_label, - queued_file.remote_id); -} - -void FileTransferManager::Unregister(uint32_t file_id, bool per_peer) { - if (per_peer) { - std::lock_guard lock(file_id_to_props_mutex_); - file_id_to_props_.erase(file_id); - } else { - std::lock_guard lock(file_id_to_state_mutex_); - file_id_to_state_.erase(file_id); - } -} - -void FileTransferManager::HandleAck(const char *data, size_t size) { - FileTransferAck ack{}; - if (!DecodeFileTransferAck(data, size, &ack)) { - LOG_ERROR("FileTransferAck: invalid payload, size={}", size); - return; - } - - std::shared_ptr props; - { - std::shared_lock lock(file_id_to_props_mutex_); - const auto it = file_id_to_props_.find(ack.file_id); - if (it != file_id_to_props_.end()) { - props = it->second.lock(); - } - } - - FileTransferState *state = props ? &props->file_transfer_ : nullptr; - if (!state) { - std::shared_lock lock(file_id_to_state_mutex_); - const auto it = file_id_to_state_.find(ack.file_id); - if (it != file_id_to_state_.end()) { - state = it->second; - } - } - if (!state) { - LOG_WARN("FileTransferAck: no state found for file_id={}", ack.file_id); - return; - } - - state->file_sent_bytes_ = ack.acked_offset; - state->file_total_bytes_ = ack.total_size; - uint32_t rate_bps = 0; - if (props) { - const uint32_t bitrate = - props->net_traffic_stats_.DataOutboundBitrate(); - if (bitrate > 0 && state->file_sending_.load()) { - rate_bps = static_cast(bitrate * 0.99f); - const uint32_t current_rate = state->file_send_rate_bps_.load(); - if (current_rate > 0) { - rate_bps = static_cast(current_rate * 0.7 + rate_bps * 0.3); - } - } else { - rate_bps = state->file_send_rate_bps_.load(); - } - } else { - const uint32_t current_rate = state->file_send_rate_bps_.load(); - uint32_t estimated_rate = 0; - const auto now = std::chrono::steady_clock::now(); - uint64_t last_bytes = 0; - std::chrono::steady_clock::time_point last_time; - { - std::lock_guard lock(state->file_transfer_mutex_); - last_bytes = state->file_send_last_bytes_; - last_time = state->file_send_last_update_time_; - } - if (state->file_sending_.load() && ack.acked_offset >= last_bytes) { - const auto delta_bytes = ack.acked_offset - last_bytes; - const double seconds = - std::chrono::duration(now - last_time).count(); - if (seconds > 0.0 && delta_bytes > 0) { - const double bits_per_second = delta_bytes * 8.0 / seconds; - estimated_rate = static_cast((std::min)( - bits_per_second, - static_cast((std::numeric_limits::max)()))); - } - } - rate_bps = - estimated_rate > 0 && current_rate > 0 - ? static_cast(current_rate * 0.7 + estimated_rate * 0.3) - : estimated_rate > 0 ? estimated_rate - : current_rate; - } - - state->file_send_rate_bps_ = rate_bps; - state->file_send_last_bytes_ = ack.acked_offset; - state->file_send_last_update_time_ = std::chrono::steady_clock::now(); - { - std::lock_guard lock(state->file_transfer_list_mutex_); - for (auto &info : state->file_transfer_list_) { - if (info.file_id == ack.file_id) { - info.sent_bytes = ack.acked_offset; - info.file_size = ack.total_size; - info.rate_bps = rate_bps; - if ((ack.flags & 0x01) != 0) { - info.status = FileTransferState::FileTransferStatus::Completed; - info.sent_bytes = ack.total_size; - } - break; - } - } - } - - if ((ack.flags & 0x01) == 0) { - return; - } - LOG_INFO("File transfer completed: file_id={}, bytes={}", ack.file_id, - ack.total_size); - state->file_transfer_window_visible_ = true; - state->file_sending_ = false; - Unregister(ack.file_id, props != nullptr); ProcessQueue(props); } -} // namespace crossdesk +void FileTransferManager::ProcessQueue(std::shared_ptr props) { + std::lock_guard workers_lock(workers_mutex_); + auto& state = state_for(props); + if (stopping_ || state.file_sending_) return; + FileTransferState::QueuedFile file; + { + std::lock_guard lock(state.file_queue_mutex_); + if (state.file_send_queue_.empty()) return; + file = state.file_send_queue_.front(); state.file_send_queue_.pop(); + state.file_sending_ = true; + } + std::erase_if(workers_, [](auto& worker) { + return worker.wait_for(std::chrono::seconds(0)) == std::future_status::ready; + }); + workers_.push_back(std::async(std::launch::async, [this, props, file] { + auto& state = state_for(props); + const auto id = file_transfer::Session::NextFileId(); + PeerPtr* peer; + std::shared_ptr events; + { + std::shared_lock lock(owner_.remote_sessions_mutex_); + peer = props ? props->peer_ : owner_.peer_; + events = props ? props->peer_events_ : owner_.peer_events_; + } + auto session = std::make_shared( + [peer, events, file, props](const char*, const char* data, size_t size) { + if (!peer || !events || (props && props->closing_)) return -1; + auto callback = events->EnterCallback(); + if (!callback) return -1; + return SendReliableDataFrameToPeer(peer, data, size, file.file_label.c_str(), + file.remote_id.c_str(), file.remote_id.size()); + }, file_transfer::Session::Progress{}); + state.file_sent_bytes_ = 0; state.file_total_bytes_ = 0; state.file_send_rate_bps_ = 0; + state.current_file_id_ = id; + { + std::lock_guard lock(state.file_transfer_mutex_); + state.file_send_start_time_ = state.file_send_last_update_time_ = std::chrono::steady_clock::now(); + state.file_send_last_bytes_ = 0; + } + { + std::lock_guard lock(state.file_transfer_list_mutex_); + for (auto& info : state.file_transfer_list_) { + if (info.file_path == file.file_path && info.status == FileTransferState::FileTransferStatus::Queued) { + info.status = FileTransferState::FileTransferStatus::Sending; + info.file_id = id; info.sent_bytes = 0; + break; + } + } + } + const auto name = file.file_path.filename().u8string(); + bool started = false; + { + std::lock_guard lock(sessions_mutex_); + const auto closed = disconnected_.find(events.get()); + const bool disconnected = closed != disconnected_.end() && + (closed->second.count("") || closed->second.count(file.remote_id)); + if (!stopping_ && peer && events && events->IsActive() && !disconnected && + (!props || !props->closing_) && !file.remote_id.empty()) { + started = session->Start(file.file_path, std::string(name.begin(), name.end()), + file_transfer::Session::Clock::now(), id); + if (started) senders_.emplace(id, Sender{session, events, file.remote_id}); + } + } + uint64_t previous = UINT64_MAX; + auto progress = file_transfer::SendState{id, 0, 0, file_transfer::Status::Failed}; + if (started) { + do { + if (stopping_ || !events->IsActive() || (props && props->closing_)) session->Cancel(); + const bool pumped = session->Pump(); + progress = session->Snapshot(); + if (progress.acknowledged != previous || progress.status != file_transfer::Status::Sending) { + UpdateProgress(state, props, progress); previous = progress.acknowledged; + } + if (progress.status != file_transfer::Status::Sending) break; + if (!pumped) { + std::unique_lock lock(wake_mutex_); + wake_.wait_for(lock, std::chrono::milliseconds(5)); + } + } while (true); + } + { + std::lock_guard lock(sessions_mutex_); + senders_.erase(id); + } + { + std::lock_guard lock(state.file_transfer_list_mutex_); + for (auto& info : state.file_transfer_list_) if (info.file_id == id) { + info.status = progress.status == file_transfer::Status::Completed + ? FileTransferState::FileTransferStatus::Completed : FileTransferState::FileTransferStatus::Failed; + if (info.status == FileTransferState::FileTransferStatus::Completed) info.sent_bytes = info.file_size; + break; + } + } + if (progress.status != file_transfer::Status::Completed) { + state.file_transfer_window_visible_ = false; + state.file_sent_bytes_ = 0; state.file_total_bytes_ = 0; + state.file_send_rate_bps_ = 0; state.current_file_id_ = 0; + } else state.file_transfer_window_visible_ = true; + state.file_sending_ = false; + // Only the worker advances the queue, after all sends for this file drained. + ProcessQueue(props); + })); +} + +void FileTransferManager::UpdateProgress(FileTransferState& state, + const std::shared_ptr& props, const file_transfer::SendState& progress) { + state.file_sent_bytes_ = progress.acknowledged; state.file_total_bytes_ = progress.total; + const auto now = std::chrono::steady_clock::now(); + const uint32_t current = state.file_send_rate_bps_.load(); + uint32_t rate = current; + { + std::lock_guard lock(state.file_transfer_mutex_); + if (props) { + const auto bitrate = props->net_traffic_stats_.DataOutboundBitrate(); + if (bitrate) rate = current ? static_cast(current * .7 + bitrate * .99 * .3) + : static_cast(bitrate * .99); + } else { + const double seconds = std::chrono::duration(now - state.file_send_last_update_time_).count(); + if (seconds > 0 && progress.acknowledged > state.file_send_last_bytes_) { + const double measured = (progress.acknowledged - state.file_send_last_bytes_) * 8.0 / seconds; + const auto value = static_cast((std::min)(measured, + static_cast((std::numeric_limits::max)()))); + rate = current ? static_cast(current * .7 + value * .3) : value; + } + } + state.file_send_last_bytes_ = progress.acknowledged; state.file_send_last_update_time_ = now; + } + state.file_send_rate_bps_ = rate; + std::lock_guard lock(state.file_transfer_list_mutex_); + for (auto& info : state.file_transfer_list_) if (info.file_id == progress.id) { + info.sent_bytes = progress.acknowledged; info.file_size = progress.total; info.rate_bps = rate; + break; + } +} + +void FileTransferManager::HandleAck(const char* data, size_t size, + PeerEventHandler* handler, const std::string& remote) { + FileTransferAck ack{}; + if (!DecodeFileTransferAck(data, size, &ack)) return; + std::shared_ptr session; + { + std::lock_guard lock(sessions_mutex_); + const auto it = senders_.find(ack.file_id); + if (it == senders_.end() || it->second.handler.get() != handler || it->second.remote != remote) return; + session = it->second.session; + } + session->Ack(data, size); + wake_.notify_all(); +} + +void FileTransferManager::Receive(const char* data, size_t size, + PeerEventHandler* handler, const std::string& remote) { + const auto props = owner_.FindRemoteSession(remote); + PeerPtr* peer; + std::shared_ptr events; + { + std::shared_lock lock(owner_.remote_sessions_mutex_); + peer = props ? props->peer_ : owner_.peer_; + events = props ? props->peer_events_ : owner_.peer_events_; + } + if (!peer || !events || events.get() != handler || (props && props->closing_)) return; + std::shared_ptr receiver; + std::filesystem::path default_directory; + { + std::lock_guard lock(sessions_mutex_); + const auto closed = disconnected_.find(handler); + if (stopping_ || (closed != disconnected_.end() && + (closed->second.count("") || closed->second.count(remote)))) return; + auto& entry = receivers_[handler][remote]; + if (!entry.session) { + entry.session = std::make_shared(); entry.handler = events; + entry.default_directory = entry.session->OutputDir(); + entry.session->SetOnSendAck([peer, events, remote](const FileTransferAck& ack) { + auto callback = events->EnterCallback(); + if (!callback) return -1; + const auto bytes = EncodeFileTransferAck(ack); + return SendReliableDataFrameToPeer(peer, bytes.data(), bytes.size(), kFileFeedbackStream, + remote.c_str(), remote.size()); + }); + } + receiver = entry.session; + default_directory = entry.default_directory; + } + const auto configured = owner_.config_center_->GetFileTransferSavePath(); + receiver->SetOutputDir(configured.empty() ? default_directory : PathFromUtf8(configured)); + receiver->OnData(data, size); +} + +void FileTransferManager::Connected(PeerEventHandler* handler, const std::string& remote) { + std::lock_guard lock(sessions_mutex_); + if (remote.empty()) disconnected_.erase(handler); + else if (const auto it = disconnected_.find(handler); it != disconnected_.end()) { + it->second.erase(remote); + if (it->second.empty()) disconnected_.erase(it); + } +} + +void FileTransferManager::Cancel(PeerEventHandler* handler, const std::string& remote) { + std::lock_guard lock(sessions_mutex_); + disconnected_[handler].insert(remote); + for (auto& [id, sender] : senders_) + if (sender.handler.get() == handler && (remote.empty() || sender.remote == remote)) sender.session->Cancel(); + const auto it = receivers_.find(handler); + if (it != receivers_.end()) { + for (auto entry = it->second.begin(); entry != it->second.end();) { + if (remote.empty() || entry->first == remote) { + entry->second.session->Cancel(); entry = it->second.erase(entry); + } else ++entry; + } + if (it->second.empty()) receivers_.erase(it); + } + wake_.notify_all(); +} + +void FileTransferManager::Poll() { + std::vector> receivers; + { + std::lock_guard lock(sessions_mutex_); + for (const auto& [handler, peers] : receivers_) + for (const auto& [remote, receiver] : peers) receivers.push_back(receiver.session); + } + for (const auto& receiver : receivers) receiver->Poll(); +} + +} // namespace crossdesk diff --git a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h index fa143a47..9fbd8b92 100644 --- a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h +++ b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h @@ -1,61 +1,66 @@ #ifndef CROSSDESK_GUI_FILE_TRANSFER_MANAGER_H_ #define CROSSDESK_GUI_FILE_TRANSFER_MANAGER_H_ -#include -#include -#include +#include +#include #include -#include -#include #include -#include +#include +#include #include #include +#include +#include "file_transfer.h" #include "runtime/remote_session.h" namespace crossdesk { class GuiRuntime; +class PeerEventHandler; class FileTransferManager { -public: + public: using FileTransferState = gui_detail::FileTransferState; using RemoteSession = gui_detail::RemoteSession; - - explicit FileTransferManager(GuiRuntime &owner); + explicit FileTransferManager(GuiRuntime& owner); ~FileTransferManager(); + FileTransferState& global_state(); + FileTransferState& state_for(const std::shared_ptr& props); + void ProcessSelectedFile(const std::string& path, const std::shared_ptr& props, + const std::string& file_label, const std::string& remote_id = ""); + void HandleAck(const char* data, size_t size, PeerEventHandler* handler, const std::string& remote); + void Receive(const char* data, size_t size, PeerEventHandler* handler, const std::string& remote); + void Connected(PeerEventHandler* handler, const std::string& remote); + void Cancel(PeerEventHandler* handler, const std::string& remote = ""); + void Poll(); + void Shutdown(); - FileTransferState &global_state(); - FileTransferState & - state_for(const std::shared_ptr &props); - - void - ProcessSelectedFile(const std::string &path, - const std::shared_ptr &props, - const std::string &file_label, - const std::string &remote_id = ""); - void HandleAck(const char *data, size_t size); - -private: - void Start(std::shared_ptr props, - const std::filesystem::path &file_path, - const std::string &file_label, const std::string &remote_id = ""); + private: + struct Sender { + std::shared_ptr session; + std::shared_ptr handler; + std::string remote; + }; + struct Receiver { + std::shared_ptr session; + std::shared_ptr handler; + std::filesystem::path default_directory; + }; void ProcessQueue(std::shared_ptr props); - void Unregister(uint32_t file_id, bool per_peer); - - std::mutex workers_mutex_; + void UpdateProgress(FileTransferState& state, const std::shared_ptr& props, + const file_transfer::SendState& progress); + std::mutex workers_mutex_, sessions_mutex_, wake_mutex_; + std::condition_variable wake_; std::vector> workers_; - bool stopping_ = false; - GuiRuntime &owner_; + std::atomic stopping_{false}; + std::unordered_map senders_; + std::unordered_map> receivers_; + std::unordered_map> disconnected_; + GuiRuntime& owner_; FileTransferState global_state_; - std::unordered_map> - file_id_to_props_; - std::shared_mutex file_id_to_props_mutex_; - std::unordered_map file_id_to_state_; - std::shared_mutex file_id_to_state_mutex_; }; -} // namespace crossdesk +} // namespace crossdesk -#endif // CROSSDESK_GUI_FILE_TRANSFER_MANAGER_H_ +#endif diff --git a/apps/desktop/src/gui/runtime/connection_runtime.cpp b/apps/desktop/src/gui/runtime/connection_runtime.cpp index 946b544c..798f7137 100644 --- a/apps/desktop/src/gui/runtime/connection_runtime.cpp +++ b/apps/desktop/src/gui/runtime/connection_runtime.cpp @@ -188,6 +188,7 @@ void GuiRuntime::HandlePresenceProbeTimeout() { void GuiRuntime::HandleServerControllerDisconnected( const std::string& remote_id, const char* reason) { + transfers_.Cancel(peer_events_.get(), remote_id); preview_subscriptions_.Remove(remote_id); keyboard_.ReleaseRemotePressedKeys(remote_id, reason); #ifdef _WIN32 @@ -298,7 +299,10 @@ int GuiRuntime::ForceSignalReconnect() { void GuiRuntime::CloseRemoteSession(std::shared_ptr props) { if (!props || props->closing_.exchange(true)) return; MuteRemoteAudio(props); - if (props->peer_events_) props->peer_events_->Deactivate(); + if (props->peer_events_) { + props->peer_events_->Deactivate(); + transfers_.Cancel(props->peer_events_.get()); + } props->connection_established_ = false; props->streaming_ = false; props->connection_status_.store(ConnectionStatus::Closed); @@ -392,6 +396,7 @@ void GuiRuntime::CloseConnectionPeer() { return; } peer_events_->Deactivate(); + transfers_.Cancel(peer_events_.get()); announcements_.Reset(); std::vector controllers; { @@ -439,6 +444,7 @@ void GuiRuntime::CloseAllRemoteSessions() { } void GuiRuntime::HandleSessionCleanup() { + transfers_.Poll(); auto completed = [](std::future& task) { if (!task.valid() || task.wait_for(std::chrono::seconds(0)) != std::future_status::ready) return false; diff --git a/apps/desktop/src/gui/runtime/gui_runtime.cpp b/apps/desktop/src/gui/runtime/gui_runtime.cpp index abdf4c0a..75a0d093 100644 --- a/apps/desktop/src/gui/runtime/gui_runtime.cpp +++ b/apps/desktop/src/gui/runtime/gui_runtime.cpp @@ -26,7 +26,10 @@ GuiRuntime::GuiRuntime() : clipboard_(*this), devices_(*this), transfers_(*this), settings_(*this), keyboard_(*this) {} -GuiRuntime::~GuiRuntime() { WaitForSessionCleanup(); } +GuiRuntime::~GuiRuntime() { + transfers_.Shutdown(); + WaitForSessionCleanup(); +} void GuiRuntime::SendRemoteInput(const RemoteAction& action) { const bool mouse = action.type == ControlType::mouse; diff --git a/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp b/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp index 927ffcb2..39198419 100644 --- a/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp +++ b/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp @@ -101,38 +101,7 @@ void PeerEventHandler::OnReceiveDataBuffer( source_id == runtime->clipboard_label_)) return; if (source_id == runtime->file_label_) { - std::string remote_user_id = std::string(user_id, user_id_size); - - static FileReceiver receiver; - // Update output directory from config - std::string configured_path = - runtime->config_center_->GetFileTransferSavePath(); - if (!configured_path.empty()) { - receiver.SetOutputDir(PathFromUtf8(configured_path)); - } else if (receiver.OutputDir().empty()) { - receiver = FileReceiver(); // re-init with default desktop path - } - receiver.SetOnSendAck([runtime, - remote_user_id](const FileTransferAck &ack) -> int { - const auto encoded_ack = EncodeFileTransferAck(ack); - bool is_server_sending = remote_user_id.rfind("C-", 0) != 0; - if (is_server_sending) { - auto props = - runtime->FindRemoteSession(remote_user_id); - if (props) { - PeerPtr *peer = props->peer_; - return SendReliableDataFrame( - peer, encoded_ack.data(), encoded_ack.size(), - runtime->file_feedback_label_.c_str()); - } - } - - return SendReliableDataFrame( - runtime->peer_, encoded_ack.data(), encoded_ack.size(), - runtime->file_feedback_label_.c_str()); - }); - - receiver.OnData(data, size); + runtime->transfers_.Receive(data, size, handler, std::string(user_id, user_id_size)); return; } else if (source_id == runtime->clipboard_label_) { if (size > 0) { @@ -148,7 +117,7 @@ void PeerEventHandler::OnReceiveDataBuffer( } return; } else if (source_id == runtime->file_feedback_label_) { - runtime->transfers_.HandleAck(data, size); + runtime->transfers_.HandleAck(data, size, handler, std::string(user_id, user_id_size)); return; } diff --git a/apps/desktop/src/gui/runtime/peer_event_handler.cpp b/apps/desktop/src/gui/runtime/peer_event_handler.cpp index bcfe4e42..17ae7e44 100644 --- a/apps/desktop/src/gui/runtime/peer_event_handler.cpp +++ b/apps/desktop/src/gui/runtime/peer_event_handler.cpp @@ -61,7 +61,9 @@ void PeerEventHandler::SendClientInfo(PeerPtr* peer, } } -PeerEventHandler::PeerEventHandler(GuiRuntime& owner) : owner_(owner) {} +PeerEventHandler::PeerEventHandler(GuiRuntime& owner) : owner_(owner) { + owner_.transfers_.Connected(this, ""); +} void PeerEventHandler::OnSignalMessage(const char* message, size_t size, void* user_data) { @@ -257,6 +259,9 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status, if (!runtime) return; std::string remote_id(user_id, user_id_size); + if (status == ConnectionStatus::Connected) runtime->transfers_.Connected(handler, remote_id); + else if (status == ConnectionStatus::Disconnected || status == ConnectionStatus::Failed || + status == ConnectionStatus::Closed) runtime->transfers_.Cancel(handler, remote_id); std::shared_ptr props; { std::shared_lock lock(runtime->remote_sessions_mutex_); diff --git a/apps/desktop/src/gui/runtime/remote_session.h b/apps/desktop/src/gui/runtime/remote_session.h index 8ea173e7..961ba4c6 100644 --- a/apps/desktop/src/gui/runtime/remote_session.h +++ b/apps/desktop/src/gui/runtime/remote_session.h @@ -44,7 +44,7 @@ struct FileTransferState { std::chrono::steady_clock::time_point file_send_start_time_; std::chrono::steady_clock::time_point file_send_last_update_time_; uint64_t file_send_last_bytes_ = 0; - bool file_transfer_window_visible_ = false; + std::atomic file_transfer_window_visible_ = false; std::atomic current_file_id_{0}; struct QueuedFile { diff --git a/apps/desktop/src/tools/file_transfer.cpp b/apps/desktop/src/tools/file_transfer.cpp index dbf75482..99df886c 100644 --- a/apps/desktop/src/tools/file_transfer.cpp +++ b/apps/desktop/src/tools/file_transfer.cpp @@ -1,116 +1,20 @@ #include "file_transfer.h" -#include -#include - +#include #include "rd_log.h" namespace crossdesk { -namespace { -std::atomic g_next_file_id{1}; -} // namespace - -uint32_t FileSender::NextFileId() { return g_next_file_id.fetch_add(1); } - -int FileSender::SendFile(const std::filesystem::path& path, - const std::string& label, const SendFunc& send, - std::size_t chunk_size, uint32_t file_id) { - if (!send) { - LOG_ERROR("FileSender::SendFile: send function is empty"); - return -1; - } - - std::error_code ec; - if (!std::filesystem::exists(path, ec) || - !std::filesystem::is_regular_file(path, ec)) { - LOG_ERROR("FileSender::SendFile: file [{}] not found or not regular", - path.string().c_str()); - return -1; - } - - uint64_t total_size = std::filesystem::file_size(path, ec); - if (ec) { - LOG_ERROR("FileSender::SendFile: failed to get size of [{}]: {}", - path.string().c_str(), ec.message().c_str()); - return -1; - } - - std::ifstream ifs(path, std::ios::binary); - if (!ifs.is_open()) { - LOG_ERROR("FileSender::SendFile: failed to open [{}]", - path.string().c_str()); - return -1; - } - LOG_INFO("FileSender send file {}, total size {}, file_id={}", - path.string().c_str(), total_size, file_id); - - if (file_id == 0) { - file_id = NextFileId(); - } - const uint32_t final_file_id = file_id; - uint64_t offset = 0; - bool is_first = true; - std::string file_name = label.empty() ? path.filename().string() : label; - - std::vector buffer; - buffer.resize(chunk_size); - - while (ifs && offset < total_size) { - uint64_t remaining = total_size - offset; - uint32_t to_read = - static_cast(std::min(remaining, chunk_size)); - - ifs.read(buffer.data(), static_cast(to_read)); - std::streamsize bytes_read = ifs.gcount(); - if (bytes_read <= 0) { - break; - } - - bool is_last = (offset + static_cast(bytes_read) >= total_size); - const std::string* name_ptr = is_first ? &file_name : nullptr; - - std::vector chunk = BuildChunk( - final_file_id, offset, total_size, buffer.data(), - static_cast(bytes_read), name_ptr, is_first, is_last); - - int ret = send(chunk.data(), chunk.size()); - if (ret != 0) { - LOG_ERROR("FileSender::SendFile: send failed for [{}], ret={}", - path.string().c_str(), ret); - return ret; - } - - offset += static_cast(bytes_read); - is_first = false; - } - - return 0; -} - -std::vector FileSender::BuildChunk(uint32_t file_id, uint64_t offset, - uint64_t total_size, const char* data, - uint32_t data_size, - const std::string* file_name, - bool is_first, bool is_last) { - return EncodeFileChunk(file_id, offset, total_size, data, - data_size, file_name, is_first, is_last); -} - -// ---------- FileReceiver ---------- - -FileReceiver::FileReceiver() : output_dir_(GetDefaultDesktopPath()) {} +FileReceiver::FileReceiver() : FileReceiver(GetDefaultDesktopPath()) {} FileReceiver::FileReceiver(const std::filesystem::path& output_dir) - : output_dir_(output_dir) { - std::error_code ec; - if (!output_dir_.empty()) { - std::filesystem::create_directories(output_dir_, ec); - if (ec) { - LOG_ERROR("FileReceiver: failed to create output dir [{}]: {}", - output_dir_.string().c_str(), ec.message().c_str()); - } - } + : output_dir_(output_dir), session_( + [this](const char*, const char* data, size_t size) { + FileTransferAck ack{}; + return on_send_ack_ && DecodeFileTransferAck(data, size, &ack) + ? on_send_ack_(ack) : 0; + }, {}) { + session_.SetDirectory(output_dir); } std::filesystem::path FileReceiver::GetDefaultDesktopPath() { @@ -136,117 +40,4 @@ std::filesystem::path FileReceiver::GetDefaultDesktopPath() { return desktop_path; } -bool FileReceiver::OnData(const char* data, size_t size) { - FileChunkView chunk; - if (!DecodeFileChunk(data, size, &chunk)) { - LOG_ERROR("FileReceiver::OnData: invalid buffer"); - return false; - } - const std::string* file_name = - chunk.file_name.empty() ? nullptr : &chunk.file_name; - return HandleChunk(chunk.header, chunk.payload, chunk.payload_size, - file_name); -} - -bool FileReceiver::HandleChunk(const FileChunkHeader& header, - const char* payload, size_t payload_size, - const std::string* file_name) { - auto it = contexts_.find(header.file_id); - if (it == contexts_.end()) { - // new file context must start with first chunk. - if ((header.flags & 0x01) == 0) { - LOG_ERROR("FileReceiver: received non-first chunk for unknown file_id={}", - header.file_id); - return false; - } - - FileContext ctx; - ctx.total_size = header.total_size; - - std::string filename; - if (file_name && !file_name->empty()) { - filename = *file_name; - } else { - filename = "received_" + std::to_string(header.file_id); - } - - ctx.file_name = filename; - - std::filesystem::path save_path = output_dir_.empty() - ? std::filesystem::path(filename) - : output_dir_ / filename; - - // if file exists, append timestamp. - std::error_code ec; - if (std::filesystem::exists(save_path, ec)) { - auto now = std::chrono::system_clock::now(); - auto ts = std::chrono::duration_cast( - now.time_since_epoch()) - .count(); - save_path = save_path.parent_path() / - (save_path.stem().string() + "_" + std::to_string(ts) + - save_path.extension().string()); - } - - ctx.ofs.open(save_path, std::ios::binary | std::ios::trunc); - if (!ctx.ofs.is_open()) { - LOG_ERROR("FileReceiver: failed to open [{}] for writing", - save_path.string().c_str()); - return false; - } - - contexts_.emplace(header.file_id, std::move(ctx)); - it = contexts_.find(header.file_id); - } - - FileContext& ctx = it->second; - - if (payload_size > 0 && payload) { - ctx.ofs.seekp(static_cast(header.offset), std::ios::beg); - ctx.ofs.write(payload, static_cast(payload_size)); - if (!ctx.ofs.good()) { - LOG_ERROR("FileReceiver: write failed for file_id={}", header.file_id); - return false; - } - ctx.received += static_cast(payload_size); - } - - // Send ACK after processing chunk - if (on_send_ack_) { - FileTransferAck ack{}; - ack.magic = kFileAckMagic; - ack.file_id = header.file_id; - ack.acked_offset = header.offset + static_cast(payload_size); - ack.total_size = header.total_size; - ack.flags = 0; - - bool is_last = (header.flags & 0x02) != 0; - if (is_last || ctx.received >= ctx.total_size) { - ack.flags |= 0x01; // completed - } - - int ret = on_send_ack_(ack); - if (ret != 0) { - LOG_ERROR("FileReceiver: failed to send ACK for file_id={}, ret={}", - header.file_id, ret); - } - } - - bool is_last = (header.flags & 0x02) != 0; - if (is_last || ctx.received >= ctx.total_size) { - ctx.ofs.close(); - - std::filesystem::path saved_path = - output_dir_.empty() ? std::filesystem::path(ctx.file_name) - : output_dir_ / ctx.file_name; - - LOG_INFO("FileReceiver: file received complete, file_id={}, size={}", - header.file_id, ctx.received); - - contexts_.erase(header.file_id); - } - - return true; -} - } // namespace crossdesk diff --git a/apps/desktop/src/tools/file_transfer.h b/apps/desktop/src/tools/file_transfer.h index 6d6e4877..2a070cf0 100644 --- a/apps/desktop/src/tools/file_transfer.h +++ b/apps/desktop/src/tools/file_transfer.h @@ -1,99 +1,46 @@ /* * @Author: DI JUNKUN * @Date: 2025-12-18 - * Copyright (c) 2025 by DI JUNKUN, All Rights Reserved. + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. */ #ifndef _FILE_TRANSFER_H_ #define _FILE_TRANSFER_H_ #include -#include #include -#include -#include -#include - -#include -#include +#include namespace crossdesk { -class FileSender { - public: - using SendFunc = std::function; - - public: - FileSender() = default; - - // generate a new file id - static uint32_t NextFileId(); - - // synchronously send a file using the provided send function. - // `path` : full path to the local file. - // `label` : logical filename to send (usually path.filename()). - // `send` : callback that pushes one encoded chunk into the data channel. - // `file_id` : file id to use (0 means auto-generate). - // Return 0 on success, <0 on error. - int SendFile(const std::filesystem::path& path, const std::string& label, - const SendFunc& send, - std::size_t chunk_size = kFileChunkSize, - uint32_t file_id = 0); - - // Build a single encoded chunk buffer using the FileChunkHeader wire format. - static std::vector BuildChunk(uint32_t file_id, uint64_t offset, - uint64_t total_size, const char* data, - uint32_t data_size, - const std::string* file_name, - bool is_first, bool is_last); -}; - +// Desktop destination policy; protocol, assembly, ACKs and cleanup live in the +// same engine used by the mobile adapter. class FileReceiver { public: - struct FileContext { - std::string file_name; - uint64_t total_size = 0; - uint64_t received = 0; - std::ofstream ofs; - }; - - using OnFileComplete = - std::function; - using OnSendAck = std::function; - - public: - // save to default desktop directory. + using OnSendAck = std::function; FileReceiver(); - - // save to a specified directory. explicit FileReceiver(const std::filesystem::path& output_dir); - - // process one received data buffer (one chunk). - // return true if parsed and processed successfully, false otherwise. - bool OnData(const char* data, size_t size); - - void SetOnSendAck(OnSendAck cb) { on_send_ack_ = cb; } - - const std::filesystem::path& OutputDir() const { return output_dir_; } - + bool OnData(const char* data, size_t size) { return session_.Receive(data, size); } + void Poll() { session_.Pump(); } + void Cancel() { session_.Cancel(); } + // Set once, before the receiver is published to transport callbacks. + void SetOnSendAck(OnSendAck cb) { on_send_ack_ = std::move(cb); } + std::filesystem::path OutputDir() const { + std::lock_guard lock(directory_mutex_); + return output_dir_; + } void SetOutputDir(const std::filesystem::path& dir) { + std::lock_guard lock(directory_mutex_); output_dir_ = dir; - if (!output_dir_.empty()) { - std::error_code ec; - std::filesystem::create_directories(output_dir_, ec); - } + session_.SetDirectory(dir); } private: static std::filesystem::path GetDefaultDesktopPath(); - - bool HandleChunk(const FileChunkHeader& header, const char* payload, - size_t payload_size, const std::string* file_name); - - private: + mutable std::mutex directory_mutex_; std::filesystem::path output_dir_; - std::unordered_map contexts_; - OnSendAck on_send_ack_ = nullptr; + OnSendAck on_send_ack_; + file_transfer::Session session_; }; } // namespace crossdesk diff --git a/apps/desktop/xmake/targets.lua b/apps/desktop/xmake/targets.lua index da95327d..3ff3d883 100644 --- a/apps/desktop/xmake/targets.lua +++ b/apps/desktop/xmake/targets.lua @@ -485,6 +485,7 @@ function setup_targets() add_files("apps/desktop/src/platform/linux/clipboard.cpp") end add_includedirs("apps/desktop/src", "apps/desktop/src/tools", {public = true}) + add_includedirs("libs/controller/include", {public = true}) target("gui") set_kind("object") diff --git a/libs/controller/include/file_transfer_core.h b/libs/controller/include/file_transfer_core.h new file mode 100644 index 00000000..a0c7a795 --- /dev/null +++ b/libs/controller/include/file_transfer_core.h @@ -0,0 +1,388 @@ +/* + * @Author: DI JUNKUN + * @Date: 2026-10-16 + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. + */ + +#ifndef _FILE_TRANSFER_CORE_H_ +#define _FILE_TRANSFER_CORE_H_ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace crossdesk::file_transfer { + +// UI, RTC peers and platform permissions stay in the adapters. Only mobile +// adapters transfer ownership of disposable picker copies to this engine. +struct Options { + bool remove_source = false; + bool receive_subdirectory = false; + size_t max_incoming = 0; +}; + +enum class Status { Idle, Sending, Completed, Failed }; +struct SendState { + uint32_t id = 0; + uint64_t total = 0, acknowledged = 0; + Status status = Status::Idle; +}; + +class Session { + public: + using Clock = std::chrono::steady_clock; + using Send = std::function; + // Progress callbacks must not reenter the session. Transport callbacks run + // outside its mutex, allowing immediate/reentrant ACKs and concurrent peers. + using Progress = std::function; + Session(Send send, Progress progress, Options options = {}) + : send_(std::move(send)), progress_(std::move(progress)), options_(options) {} + ~Session() { Clear(); } + + static uint32_t NextFileId() { + static std::atomic next{1}; + uint32_t id; + do { id = next.fetch_add(1); } while (id == 0); + return id; + } + void SetDirectory(std::filesystem::path directory) { + std::lock_guard lock(mutex_); + directory_ = std::move(directory); + } + bool Start(const std::filesystem::path& path, const std::string& name, + Clock::time_point now = Clock::now(), uint32_t id = 0) { + std::lock_guard lock(mutex_); + if (closed_ || outgoing_ || !send_ || name.empty() || name.size() > UINT16_MAX) return false; + std::error_code ec; + if (!std::filesystem::is_regular_file(path, ec)) return false; + const auto size = std::filesystem::file_size(path, ec); + if (ec || size > INT64_MAX) return false; + Outgoing file; + file.input.open(path, std::ios::binary); + if (!file.input) return false; + file.id = id ? id : NextFileId(); + file.name = name; file.path = path; file.total = size; file.updated = now; + state_ = {file.id, size, 0, Status::Sending}; + outgoing_ = std::move(file); + Notify(name, 0, true, ""); + return true; + } + + // At most one chunk per pump; bound queued but unacknowledged bytes to 1 MiB. + bool Pump(Clock::time_point now = Clock::now()) { + std::vector expired; + std::vector chunk; + uint32_t id = 0; + { + std::lock_guard lock(mutex_); + if (closed_) return false; + for (auto it = incoming_.begin(); it != incoming_.end();) { + if (now - it->second.updated < timeout_) { ++it; continue; } + expired.push_back({kFileAckMagic, it->first, it->second.received, it->second.total, 2}); + Notify(it->second.name, -1, false, ""); + Remove(it->second); it = incoming_.erase(it); + } + if (outgoing_) { + auto& file = *outgoing_; + if (now - file.updated >= timeout_) FinishSend(false); + else if (!file.last && file.sent - file.acked < 1024 * 1024) { + std::array payload{}; + const auto size = static_cast((std::min)(payload.size(), file.total - file.sent)); + if (size) file.input.read(payload.data(), size); + if (size && file.input.gcount() != size) FinishSend(false); + else { + const bool first = file.sent == 0; + const bool last = file.sent + size == file.total; + chunk = EncodeFileChunk(file.id, file.sent, file.total, payload.data(), size, + first ? &file.name : nullptr, first, last); + if (chunk.empty()) FinishSend(false); + else { + id = file.id; + // Reserve before calling transport: an ACK can arrive before Send returns. + file.sent += size; file.last = last; + } + } + } + } + } + for (const auto& ack : expired) SendAck(ack); + if (!chunk.empty() && send_(kFileStream, chunk.data(), chunk.size()) != 0) { + std::lock_guard lock(mutex_); + if (outgoing_ && outgoing_->id == id) FinishSend(false); + } + return !chunk.empty(); + } + + void Ack(const char* data, size_t size, Clock::time_point now = Clock::now()) { + FileTransferAck ack{}; + if (!DecodeFileTransferAck(data, size, &ack)) return; + std::lock_guard lock(mutex_); + if (!outgoing_ || ack.file_id != outgoing_->id) return; + auto& file = *outgoing_; + if (ack.total_size != file.total || ack.acked_offset > file.sent || (ack.flags & ~3u)) return; + if (ack.flags & 2) { FinishSend(false); return; } + if (ack.flags & 1) { + if (file.last && ack.acked_offset == file.total) { + state_.acknowledged = file.total; + FinishSend(true); + } + return; + } + if (ack.acked_offset <= file.acked) return; + file.acked = ack.acked_offset; file.updated = now; + state_.acknowledged = file.acked; + Report(file.name, file.acked, file.total, true, file.percent); + } + + bool Receive(const char* data, size_t size, Clock::time_point now = Clock::now()) { + FileChunkView chunk; + if (!DecodeFileChunk(data, size, &chunk)) return false; + FileTransferAck ack; + { + std::lock_guard lock(mutex_); + if (closed_) return false; + ack = ReceiveLocked(chunk, size, now); + } + SendAck(ack); + return (ack.flags & 2) == 0; + } + bool Sending() const { return Snapshot().status == Status::Sending; } + SendState Snapshot() const { + std::lock_guard lock(mutex_); + return state_; + } + // A cancelled connection cannot be restarted by a late data callback. + void Cancel() { + std::lock_guard lock(mutex_); + closed_ = true; + if (outgoing_) FinishSend(false); + Clear(); + } + + private: + struct Outgoing { + uint32_t id = 0; + std::string name; + std::filesystem::path path; + std::ifstream input; + uint64_t total = 0, sent = 0, acked = 0; + bool last = false; + int percent = -1; + Clock::time_point updated; + }; + struct Incoming { + std::string name; + std::filesystem::path path, partial, folder, directory; + std::ofstream output; + uint64_t total = 0, received = 0; + bool first = false, last = false; + std::map ranges; + int percent = -1; + Clock::time_point updated; + }; + FileTransferAck ReceiveLocked(const FileChunkView& chunk, size_t size, Clock::time_point now) { + const auto& h = chunk.header; + auto it = incoming_.find(h.file_id); + auto reject = [&]() { + if (it != incoming_.end()) { + Notify(it->second.name, -1, false, ""); Remove(it->second); incoming_.erase(it); + } else Notify(SafeName(chunk.file_name), -1, false, ""); + return FileTransferAck{kFileAckMagic, h.file_id, h.offset, h.total_size, 2}; + }; + const bool first = h.flags & 1, last = h.flags & 2; + if ((h.flags & ~3u) || h.total_size > INT64_MAX || (first && h.offset != 0) || + chunk.payload_size > kFileChunkSize || size != sizeof(FileChunkHeader) + h.name_len + chunk.payload_size || + (!first && h.name_len) || (!chunk.payload_size && !last) || + last != (h.offset + chunk.payload_size == h.total_size) || + (it == incoming_.end() && options_.max_incoming && incoming_.size() >= options_.max_incoming)) + return reject(); + if (it == incoming_.end()) { + Incoming file; + file.name = SafeName(chunk.file_name); file.total = h.total_size; + file.directory = directory_.empty() ? std::filesystem::path(".") : directory_; + std::error_code ec; + std::filesystem::create_directories(file.directory, ec); + if (ec || (directory_.empty() && options_.receive_subdirectory)) return reject(); + // Reserve a unique staging directory even when two peers use the same ID. + static std::atomic serial{0}; + for (int attempt = 0; attempt < 100; ++attempt) { + const auto token = std::to_string(h.file_id) + "-" + + std::to_string(Clock::now().time_since_epoch().count()) + "-" + std::to_string(++serial); + file.folder = file.directory / (options_.receive_subdirectory ? token : ".crossdesk-" + token); + if (std::filesystem::create_directory(file.folder, ec)) break; + if (ec || attempt == 99) return reject(); + } + file.path = file.folder / NamePath(file.name); + file.partial = file.folder / ".partial"; + file.output.open(file.partial, std::ios::binary | std::ios::trunc); + if (!file.output) { Remove(file); return reject(); } + it = incoming_.emplace(h.file_id, std::move(file)).first; + } + auto& file = it->second; + if (h.total_size != file.total || (first && file.first) || + !AddRange(file, h.offset, h.offset + chunk.payload_size)) return reject(); + if (first) { + file.first = true; file.name = SafeName(chunk.file_name); + file.path = file.folder / NamePath(file.name); + } + file.last |= last; + // KCP's RTP and timer callbacks can race: ACK only contiguous coverage. + file.received = !file.ranges.empty() && file.ranges.begin()->first == 0 ? file.ranges.begin()->second : 0; + const bool complete = file.first && file.last && file.received == file.total; + file.output.seekp(static_cast(h.offset)); + file.output.write(chunk.payload, chunk.payload_size); + if (complete) { file.output.flush(); file.output.close(); } + if (file.output.fail() || (complete && !Publish(file))) return reject(); + file.updated = now; + const FileTransferAck ack{kFileAckMagic, h.file_id, file.received, file.total, complete ? 1u : 0u}; + if (complete) { + const auto path = file.path.u8string(); + Notify(file.name, 1, false, std::string(path.begin(), path.end())); + incoming_.erase(it); + } else Report(file.name, file.received, file.total, false, file.percent); + return ack; + } + bool Publish(Incoming& file) { + std::error_code ec; + if (options_.receive_subdirectory) { + std::filesystem::rename(file.partial, file.path, ec); + return !ec; + } + // Preserve the desktop's flat destination and timestamp suffix on conflict. + // Hard-link publication is atomic and never overwrites another peer's file. + const auto original = NamePath(file.name); + const auto stamp = std::to_string(std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count()); + for (unsigned attempt = 0; attempt < 1000; ++attempt) { + auto name = original; + if (attempt) { + name = original.stem(); name += "_" + stamp; + if (attempt > 1) name += "_" + std::to_string(attempt); + name += original.extension(); + } + file.path = file.directory / name; + ec.clear(); + std::filesystem::create_hard_link(file.partial, file.path, ec); + if (!ec) { Remove(file); return true; } + if (ec == std::errc::file_exists) continue; + // Some public folder providers disallow links. Exclusively create the + // destination so a failed copy can remove only the file we just created. +#if defined(_WIN32) + std::FILE* output = _wfopen(file.path.c_str(), L"wbx"); +#else + std::FILE* output = std::fopen(file.path.c_str(), "wbx"); +#endif + if (!output) { + if (errno == EEXIST) continue; + return false; + } + std::ifstream input(file.partial, std::ios::binary); + bool ok = input.is_open(); + std::array buffer{}; + while (ok && input) { + input.read(buffer.data(), buffer.size()); + const auto count = static_cast(input.gcount()); + if (count && std::fwrite(buffer.data(), 1, count, output) != count) ok = false; + } + ok = ok && input.eof() && !input.bad(); + input.close(); + if (std::fclose(output) != 0) ok = false; + if (!ok) { std::filesystem::remove(file.path, ec); return false; } + Remove(file); return true; + } + return false; + } + static bool AddRange(Incoming& file, uint64_t start, uint64_t end) { + if (start == end) return true; + auto next = file.ranges.lower_bound(start); + if (next != file.ranges.end() && next->first < end) return false; + if (next != file.ranges.begin()) { + auto previous = std::prev(next); + if (previous->second > start) return false; + if (previous->second == start) { start = previous->first; file.ranges.erase(previous); } + } + if (next != file.ranges.end() && next->first == end) { end = next->second; file.ranges.erase(next); } + file.ranges.emplace(start, end); + return file.ranges.size() <= 1024; + } + static std::filesystem::path NamePath(const std::string& name) { +#if defined(__cpp_char8_t) + return std::filesystem::path(std::u8string(name.begin(), name.end())); +#else + return std::filesystem::u8path(name); +#endif + } + static std::string SafeName(std::string name) { + const auto slash = name.find_last_of("/\\"); + if (slash != std::string::npos) name.erase(0, slash + 1); + for (auto& ch : name) + if (static_cast(ch) < 32 || std::string(":*?\"<>|").find(ch) != std::string::npos) ch = '_'; + while (!name.empty() && (name.back() == '.' || name.back() == ' ')) name.pop_back(); + if (name.empty() || name == ".partial" || name.size() > 240) return "received_file"; + return name; + } + static void Remove(Incoming& file) { + if (file.output.is_open()) file.output.close(); + std::error_code ec; + std::filesystem::remove(file.partial, ec); + std::filesystem::remove(file.folder, ec); + } + void SendAck(const FileTransferAck& ack) { + if (!send_) return; + const auto packet = EncodeFileTransferAck(ack); + send_(kFileFeedbackStream, packet.data(), packet.size()); + } + void Notify(const std::string& name, double value, bool sending, const std::string& path) { + if (progress_) progress_(name, value, sending, path); + } + void Report(const std::string& name, uint64_t offset, uint64_t total, bool sending, int& previous) { + const double progress = total ? static_cast(offset) / total : 0; + const int percent = static_cast(progress * 100); + if (percent != previous) { previous = percent; Notify(name, (std::min)(progress, .99), sending, ""); } + } + void RemoveSource() { + outgoing_->input.close(); + if (options_.remove_source) { + std::error_code ec; + std::filesystem::remove(outgoing_->path, ec); + } + outgoing_.reset(); + } + void FinishSend(bool success) { + state_.status = success ? Status::Completed : Status::Failed; + Notify(outgoing_->name, success ? 1 : -1, true, ""); + RemoveSource(); + } + void Clear() { + if (outgoing_) RemoveSource(); + for (auto& entry : incoming_) Remove(entry.second); + incoming_.clear(); + } + const std::chrono::seconds timeout_{30}; + mutable std::mutex mutex_; + Send send_; + Progress progress_; + Options options_; + bool closed_ = false; + SendState state_; + std::filesystem::path directory_; + std::optional outgoing_; + std::map incoming_; +}; + +} // namespace crossdesk::file_transfer + +#endif diff --git a/libs/controller/include/mobile_file_transfer_session.h b/libs/controller/include/mobile_file_transfer_session.h new file mode 100644 index 00000000..677c513a --- /dev/null +++ b/libs/controller/include/mobile_file_transfer_session.h @@ -0,0 +1,24 @@ +/* + * @Author: DI JUNKUN + * @Date: 2026-10-09 + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. + */ + +#ifndef _MOBILE_FILE_TRANSFER_SESSION_H_ +#define _MOBILE_FILE_TRANSFER_SESSION_H_ + +#include "file_transfer_core.h" + +namespace crossdesk::mobile_controller { + +// Mobile pickers provide disposable copies and expose completed files through +// their existing per-transfer directories. The transfer engine is shared with PC. +class FileTransferSession : public file_transfer::Session { + public: + FileTransferSession(Send send, Progress progress) + : Session(std::move(send), std::move(progress), {true, true, 4}) {} +}; + +} // namespace crossdesk::mobile_controller + +#endif