mirror of
https://github.com/kunkundi/crossdesk.git
synced 2026-10-19 02:13:03 +08:00
[feat] complete mobile session controls and file transfer
This commit is contained in:
@@ -12,6 +12,12 @@
|
||||
#include <remote_action.h>
|
||||
|
||||
namespace crossdesk::android_controller {
|
||||
inline std::string SecureAttentionMessage() {
|
||||
RemoteAction action{};
|
||||
action.type = service_command;
|
||||
action.c.flag = send_sas;
|
||||
return action.to_json();
|
||||
}
|
||||
inline std::string VideoSettingsMessage(int quality, int frame_rate, int preference,
|
||||
uint32_t request_id) {
|
||||
RemoteAction action{};
|
||||
@@ -32,4 +38,4 @@ inline nlohmann::json NetworkReport(const MiniRtcNetTrafficStats& stats, Travers
|
||||
}
|
||||
} // namespace crossdesk::android_controller
|
||||
|
||||
#endif
|
||||
#endif
|
||||
|
||||
@@ -0,0 +1,271 @@
|
||||
/*
|
||||
* @Author: DI JUNKUN
|
||||
* @Date: 2026-10-09
|
||||
* Copyright (c) 2026 by DI JUNKUN, All Rights Reserved.
|
||||
*/
|
||||
|
||||
#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>
|
||||
|
||||
namespace crossdesk::android_controller {
|
||||
|
||||
// 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
|
||||
@@ -18,6 +18,7 @@
|
||||
#include <stdexcept>
|
||||
#include <vector>
|
||||
#include "controller_protocol.h"
|
||||
#include "file_transfer_session.h"
|
||||
#include "android_codec_support.h"
|
||||
#include "latest_frame_worker.h"
|
||||
#include "video_picture.h"
|
||||
@@ -76,6 +77,14 @@ struct Session {
|
||||
float pointer_x = .5f, pointer_y = .5f;
|
||||
android_controller::LatestFrameWorker<VideoPicture> renderer{
|
||||
[this](VideoPicture picture) { RenderVideo(this, std::move(picture)); }};
|
||||
android_controller::FileTransferSession files{
|
||||
[this](const char* stream, const char* data, size_t size) {
|
||||
return controller && !stopping
|
||||
? SendReliableDataFrame(controller, data, size, stream) : -1;
|
||||
},
|
||||
[this](const std::string& name, double progress, bool sending, const std::string& path) {
|
||||
Event(10, {{"name", name}, {"progress", progress}, {"sending", sending}, {"path", path}});
|
||||
}};
|
||||
|
||||
void Event(int type, const char* data, size_t size) {
|
||||
if (stopping || size > 1024 * 1024) return;
|
||||
@@ -89,7 +98,7 @@ struct Session {
|
||||
if (scope.env->ExceptionCheck()) scope.env->ExceptionClear();
|
||||
}
|
||||
void Event(int type, const json& value) {
|
||||
const std::string data = value.dump(); Event(type, data.data(), data.size());
|
||||
const std::string data = value.dump(-1, ' ', false, json::error_handler_t::replace); Event(type, data.data(), data.size());
|
||||
}
|
||||
void Send(const char* stream, const std::string& data) {
|
||||
if (controller && !stopping && !data.empty())
|
||||
@@ -180,6 +189,8 @@ void Data(const char* data, size_t size, const char*, size_t,
|
||||
auto* ctx = static_cast<CallbackContext*>(user);
|
||||
if (!ctx->controller || !data || !stream || size > kMaxClipboardBytes) return;
|
||||
const std::string name(stream, stream_size);
|
||||
if (name == kFileStream) { ctx->owner->files.Receive(data, size); return; }
|
||||
if (name == kFileFeedbackStream) { ctx->owner->files.Ack(data, size); return; }
|
||||
if (name == kControlStream) {
|
||||
RemoteAction action{};
|
||||
if (action.from_json(std::string(data, size))) {
|
||||
@@ -352,7 +363,7 @@ extern "C" JNIEXPORT jboolean JNICALL JNI_METHOD(nConnect)(
|
||||
AddAudioStream(s->controller, kAudioStream);
|
||||
AddDataStream(s->controller, kDataStream, false);
|
||||
for (const char* name : {kMouseStream, kKeyboardStream, kControlStream,
|
||||
kClipboardStream}) AddDataStream(s->controller, name, true);
|
||||
kClipboardStream, kFileStream, kFileFeedbackStream}) AddDataStream(s->controller, name, true);
|
||||
if (Init(s->controller) != 0) return false;
|
||||
const auto join = s->remote + "@" + String(env, password);
|
||||
return JoinConnection(s->controller, join.c_str()) == 0;
|
||||
@@ -401,6 +412,30 @@ extern "C" JNIEXPORT jboolean JNICALL JNI_METHOD(nVideoSettings)(
|
||||
if (data.empty()) return false;
|
||||
return SendReliableDataFrame(s->controller, data.data(), data.size(), kControlStream) == 0;
|
||||
}
|
||||
extern "C" JNIEXPORT jboolean JNICALL JNI_METHOD(nSecureAttention)(JNIEnv*, jclass, jlong handle) {
|
||||
auto* s = From(handle);
|
||||
if (!s || !s->controller || s->stopping) return false;
|
||||
const auto data = android_controller::SecureAttentionMessage();
|
||||
return SendReliableDataFrame(s->controller, data.data(), data.size(), kControlStream) == 0;
|
||||
}
|
||||
extern "C" JNIEXPORT void JNICALL JNI_METHOD(nFileDirectory)(JNIEnv* env, jclass, jlong handle, jstring directory) {
|
||||
auto* s = From(handle); if (s) s->files.SetDirectory(String(env, directory));
|
||||
}
|
||||
extern "C" JNIEXPORT jboolean JNICALL JNI_METHOD(nSendFile)(JNIEnv* env, jclass, jlong handle, jstring path, jbyteArray name) {
|
||||
auto* s = From(handle);
|
||||
if (!s || !s->controller || s->stopping || !name) return false;
|
||||
const auto size = env->GetArrayLength(name);
|
||||
if (!size || size > UINT16_MAX) return false;
|
||||
std::string label(size, '\0');
|
||||
env->GetByteArrayRegion(name, 0, size, reinterpret_cast<jbyte*>(label.data()));
|
||||
return !env->ExceptionCheck() && s->files.Start(String(env, path), label);
|
||||
}
|
||||
extern "C" JNIEXPORT jboolean JNICALL JNI_METHOD(nPumpFiles)(JNIEnv*, jclass, jlong handle) {
|
||||
auto* s = From(handle);
|
||||
if (!s || s->stopping) return false;
|
||||
s->files.Pump();
|
||||
return s->files.Sending();
|
||||
}
|
||||
extern "C" JNIEXPORT jboolean JNICALL JNI_METHOD(nAnnouncementRequest)(
|
||||
JNIEnv* env, jclass, jlong handle, jbyteArray data) {
|
||||
auto* s = From(handle);
|
||||
|
||||
@@ -10,7 +10,7 @@ target("crossdesk_android")
|
||||
set_kind("shared")
|
||||
add_deps("minirtc")
|
||||
add_packages("nlohmann_json", "libyuv", "spdlog")
|
||||
add_files("rtc_bridge.cpp", "../../../libs/wire/src/remote_action.cpp", "../../../libs/wire/src/app_version.cpp")
|
||||
add_files("rtc_bridge.cpp", "../../../libs/wire/src/remote_action.cpp", "../../../libs/wire/src/app_version.cpp", "../../../libs/wire/src/file_transfer.cpp")
|
||||
add_includedirs("../../../deps/submodules/minirtc/src/api", "../../../libs/wire/include")
|
||||
-- This source-level JNI adapter is the only application layer using MiniRTC internals.
|
||||
add_includedirs("../../../deps/submodules/minirtc/src/media/mediacodec")
|
||||
|
||||
Reference in New Issue
Block a user