YangWX/crossdesk
1
0
Fork 0
mirror of https://github.com/kunkundi/crossdesk.git synced 2026-10-17 01:13:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

[refactor] share file transfer core across desktop and mobile

Browse Source
This commit is contained in:
kunkundi
2026-10-16 00:38:13 +08:00
parent 47e6385f9a
commit c4756b298b
13 changed files with 759 additions and 956 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/android/native/file_transfer_session.h
+4 -260
View File
@@ -7,265 +7,9 @@
#ifndef _FILE_TRANSFER_SESSION_H_
#define _FILE_TRANSFER_SESSION_H_
#include <file_transfer_format.h>
#include <stream_names.h>
#include <algorithm>
#include <array>
#include <chrono>
#include <filesystem>
#include <fstream>
#include <functional>
#include <map>
#include <mutex>
#include <optional>
#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<int(const char*, const char*, size_t)>;
using Progress = std::function<void(const std::string&, double, bool,
const std::string&)>;
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<std::mutex> 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<std::mutex> 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<std::mutex> 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<char, kFileChunkSize> payload{};
const auto size = static_cast<uint32_t>(std::min<uint64_t>(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<std::mutex> 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<std::mutex> 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<std::streamoff>(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<std::mutex> 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<uint64_t, uint64_t> 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<unsigned char>(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<double>(offset) / total : 0;
const int percent = static_cast<int>(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> outgoing_;
std::map<uint32_t, Incoming> incoming_;
uint32_t next_id_ = 0;
uint64_t receive_id_ = 0;
};
} // namespace crossdesk::android_controller
#endif
#endif
apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp
+253 -333
View File
@@ -1,370 +1,290 @@
#include "features/file_transfer/file_transfer_manager.h"
#include <algorithm>
#include <chrono>
#include <filesystem>
#include <limits>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <thread>
#include <utility>
#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<std::future<void>> 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<RemoteSession> &props) {
FileTransferManager::FileTransferState& FileTransferManager::global_state() { return global_state_; }
FileTransferManager::FileTransferState& FileTransferManager::state_for(const std::shared_ptr<RemoteSession>& props) {
return props ? props->file_transfer_ : global_state_;
}
void FileTransferManager::ProcessSelectedFile(
const std::string &path,
const std::shared_ptr<RemoteSession> &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<RemoteSession>& 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<std::mutex> 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<const char *>(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<std::mutex> 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<RemoteSession> 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<PeerEventHandler> 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<RemoteSession>(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<std::mutex> 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<std::shared_mutex> lock(file_id_to_props_mutex_);
file_id_to_props_[file_id] = props_weak;
} else {
std::lock_guard<std::shared_mutex> lock(file_id_to_state_mutex_);
file_id_to_state_[file_id] = state;
}
state->current_file_id_ = file_id;
{
std::lock_guard<std::mutex> 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<std::mutex> 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<RemoteSession> props) {
if (props && props->closing_) return;
FileTransferState &state = state_for(props);
if (state.file_sending_.load()) {
return;
}
FileTransferState::QueuedFile queued_file;
{
std::lock_guard<std::mutex> 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<std::shared_mutex> lock(file_id_to_props_mutex_);
file_id_to_props_.erase(file_id);
} else {
std::lock_guard<std::shared_mutex> 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<RemoteSession> 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<uint32_t>(bitrate * 0.99f);
const uint32_t current_rate = state->file_send_rate_bps_.load();
if (current_rate > 0) {
rate_bps = static_cast<uint32_t>(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<std::mutex> 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<double>(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<uint32_t>((std::min)(
bits_per_second,
static_cast<double>((std::numeric_limits<uint32_t>::max)())));
}
}
rate_bps =
estimated_rate > 0 && current_rate > 0
? static_cast<uint32_t>(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<std::mutex> 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<RemoteSession> 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<PeerEventHandler> 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<file_transfer::Session>(
[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<RemoteSession>& 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<uint32_t>(current * .7 + bitrate * .99 * .3)
: static_cast<uint32_t>(bitrate * .99);
} else {
const double seconds = std::chrono::duration<double>(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<uint32_t>((std::min)(measured,
static_cast<double>((std::numeric_limits<uint32_t>::max)())));
rate = current ? static_cast<uint32_t>(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<file_transfer::Session> 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<PeerEventHandler> 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<FileReceiver> 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<FileReceiver>(); 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<std::shared_ptr<FileReceiver>> 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
apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h
+41 -36
View File
@@ -1,61 +1,66 @@
#ifndef CROSSDESK_GUI_FILE_TRANSFER_MANAGER_H_
#define CROSSDESK_GUI_FILE_TRANSFER_MANAGER_H_
#include <cstddef>
#include <cstdint>
#include <filesystem>
#include <atomic>
#include <condition_variable>
#include <future>
#include <mutex>
#include <vector>
#include <memory>
#include <shared_mutex>
#include <mutex>
#include <set>
#include <string>
#include <unordered_map>
#include <vector>
#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<RemoteSession>& props);
void ProcessSelectedFile(const std::string& path, const std::shared_ptr<RemoteSession>& 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<RemoteSession> &props);
void
ProcessSelectedFile(const std::string &path,
const std::shared_ptr<RemoteSession> &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<RemoteSession> props,
const std::filesystem::path &file_path,
const std::string &file_label, const std::string &remote_id = "");
private:
struct Sender {
std::shared_ptr<file_transfer::Session> session;
std::shared_ptr<PeerEventHandler> handler;
std::string remote;
};
struct Receiver {
std::shared_ptr<FileReceiver> session;
std::shared_ptr<PeerEventHandler> handler;
std::filesystem::path default_directory;
};
void ProcessQueue(std::shared_ptr<RemoteSession> props);
void Unregister(uint32_t file_id, bool per_peer);
std::mutex workers_mutex_;
void UpdateProgress(FileTransferState& state, const std::shared_ptr<RemoteSession>& props,
const file_transfer::SendState& progress);
std::mutex workers_mutex_, sessions_mutex_, wake_mutex_;
std::condition_variable wake_;
std::vector<std::future<void>> workers_;
bool stopping_ = false;
GuiRuntime &owner_;
std::atomic<bool> stopping_{false};
std::unordered_map<uint32_t, Sender> senders_;
std::unordered_map<PeerEventHandler*, std::unordered_map<std::string, Receiver>> receivers_;
std::unordered_map<PeerEventHandler*, std::set<std::string>> disconnected_;
GuiRuntime& owner_;
FileTransferState global_state_;
std::unordered_map<uint32_t, std::weak_ptr<RemoteSession>>
file_id_to_props_;
std::shared_mutex file_id_to_props_mutex_;
std::unordered_map<uint32_t, FileTransferState *> file_id_to_state_;
std::shared_mutex file_id_to_state_mutex_;
};
} // namespace crossdesk
} // namespace crossdesk
#endif // CROSSDESK_GUI_FILE_TRANSFER_MANAGER_H_
#endif
apps/desktop/src/gui/runtime/connection_runtime.cpp
+7 -1
View File
@@ -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<RemoteSession> 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<std::string> controllers;
{
@@ -439,6 +444,7 @@ void GuiRuntime::CloseAllRemoteSessions() {
}
void GuiRuntime::HandleSessionCleanup() {
transfers_.Poll();
auto completed = [](std::future<void>& task) {
if (!task.valid() || task.wait_for(std::chrono::seconds(0)) !=
std::future_status::ready) return false;
apps/desktop/src/gui/runtime/gui_runtime.cpp
+4 -1
View File
@@ -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;
apps/desktop/src/gui/runtime/peer_data_callbacks.cpp
+2 -33
View File
@@ -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;
}
apps/desktop/src/gui/runtime/peer_event_handler.cpp
+6 -1
View File
@@ -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<GuiRuntime::RemoteSession> props;
{
std::shared_lock lock(runtime->remote_sessions_mutex_);
apps/desktop/src/gui/runtime/remote_session.h
+1 -1
View File
@@ -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<bool> file_transfer_window_visible_ = false;
std::atomic<uint32_t> current_file_id_{0};
struct QueuedFile {
apps/desktop/src/tools/file_transfer.cpp
+9 -218
View File
@@ -1,116 +1,20 @@
#include "file_transfer.h"
#include <atomic>
#include <chrono>
#include <cstdlib>
#include "rd_log.h"
namespace crossdesk {
namespace {
std::atomic<uint32_t> 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<char> buffer;
buffer.resize(chunk_size);
while (ifs && offset < total_size) {
uint64_t remaining = total_size - offset;
uint32_t to_read =
static_cast<uint32_t>(std::min<uint64_t>(remaining, chunk_size));
ifs.read(buffer.data(), static_cast<std::streamsize>(to_read));
std::streamsize bytes_read = ifs.gcount();
if (bytes_read <= 0) {
break;
}
bool is_last = (offset + static_cast<uint64_t>(bytes_read) >= total_size);
const std::string* name_ptr = is_first ? &file_name : nullptr;
std::vector<char> chunk = BuildChunk(
final_file_id, offset, total_size, buffer.data(),
static_cast<uint32_t>(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<uint64_t>(bytes_read);
is_first = false;
}
return 0;
}
std::vector<char> 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<std::chrono::milliseconds>(
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<std::streamoff>(header.offset), std::ios::beg);
ctx.ofs.write(payload, static_cast<std::streamsize>(payload_size));
if (!ctx.ofs.good()) {
LOG_ERROR("FileReceiver: write failed for file_id={}", header.file_id);
return false;
}
ctx.received += static_cast<uint64_t>(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<uint64_t>(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
apps/desktop/src/tools/file_transfer.h
+19 -72
View File
@@ -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 <filesystem>
#include <fstream>
#include <functional>
#include <string>
#include <unordered_map>
#include <vector>
#include <file_transfer_format.h>
#include <stream_names.h>
#include <file_transfer_core.h>
namespace crossdesk {
class FileSender {
public:
using SendFunc = std::function<int(const char* data, size_t size)>;
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<char> 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<void(const std::filesystem::path& saved_path)>;
using OnSendAck = std::function<int(const FileTransferAck& ack)>;
public:
// save to default desktop directory.
using OnSendAck = std::function<int(const FileTransferAck&)>;
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<uint32_t, FileContext> contexts_;
OnSendAck on_send_ack_ = nullptr;
OnSendAck on_send_ack_;
file_transfer::Session session_;
};
} // namespace crossdesk
apps/desktop/xmake/targets.lua
+1
View File
@@ -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")
libs/controller/include/file_transfer_core.h
+388
View File
@@ -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 <file_transfer_format.h>
#include <stream_names.h>
#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cerrno>
#include <cstdio>
#include <filesystem>
#include <fstream>
#include <functional>
#include <map>
#include <mutex>
#include <optional>
#include <utility>
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<int(const char*, const char*, size_t)>;
// Progress callbacks must not reenter the session. Transport callbacks run
// outside its mutex, allowing immediate/reentrant ACKs and concurrent peers.
using Progress = std::function<void(const std::string&, double, bool,
const std::string&)>;
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<uint32_t> 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<std::mutex> 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<std::mutex> 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<FileTransferAck> expired;
std::vector<char> chunk;
uint32_t id = 0;
{
std::lock_guard<std::mutex> 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<char, kFileChunkSize> payload{};
const auto size = static_cast<uint32_t>((std::min<uint64_t>)(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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> lock(mutex_);
return state_;
}
// A cancelled connection cannot be restarted by a late data callback.
void Cancel() {
std::lock_guard<std::mutex> 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<uint64_t, uint64_t> 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<uint64_t> 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<std::streamoff>(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::milliseconds>(
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<char, kFileChunkSize> buffer{};
while (ok && input) {
input.read(buffer.data(), buffer.size());
const auto count = static_cast<size_t>(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<unsigned char>(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<double>(offset) / total : 0;
const int percent = static_cast<int>(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> outgoing_;
std::map<uint32_t, Incoming> incoming_;
};
} // namespace crossdesk::file_transfer
#endif
libs/controller/include/mobile_file_transfer_session.h
+24
View File
@@ -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