ddidderr/lanspread
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

[client][server] file transfer working, ui not ready for it

Browse Source
This commit is contained in:
ddidderr 2024-11-14 23:26:31 +01:00
parent 942eb8003e
commit 2b64d1e4ba
Signed by: ddidderr
GPG Key ID: 3841F1C27E6F0E14
9 changed files with 334 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -2613,6 +2613,7 @@ dependencies = [
"tokio",
"tracing",
"tracing-subscriber",
"walkdir",
]
[[package]]

1
Cargo.toml
View File

@ -23,6 +23,7 @@ serde_json = "1.0"
tokio = { version = "1.41", features = ["full"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
walkdir = "2.5"
[profile.release]
debug = true

155
crates/lanspread-client/src/lib.rs
View File

@ -1,26 +1,36 @@
#![allow(clippy::missing_errors_doc)]
use std::{net::SocketAddr, time::Duration};
use std::{fs::File, io::Write as _, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
use bytes::{Bytes, BytesMut};
use lanspread_db::db::Game;
use lanspread_db::db::{Game, GameFileDescription};
use lanspread_proto::{Message as _, Request, Response};
use lanspread_utils::maybe_addr;
use s2n_quic::{client::Connect, provider::limits::Limits, Client as QuicClient, Connection};
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::{
io::AsyncWriteExt,
stream,
sync::{
mpsc::{UnboundedReceiver, UnboundedSender},
Mutex,
},
};
static CERT_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../cert.pem"));
#[derive(Debug)]
pub enum ClientEvent {
ListGames(Vec<Game>),
GotGameFiles(Vec<GameFileDescription>),
}
#[derive(Debug)]
pub enum ClientCommand {
ListGames,
GetGame(String),
DownloadGameFiles(Vec<GameFileDescription>),
ServerAddr(SocketAddr),
SetGameDir(String),
}
async fn initial_server_alive_check(conn: &mut Connection) -> bool {
@ -55,6 +65,73 @@ async fn initial_server_alive_check(conn: &mut Connection) -> bool {
false
}
async fn download_game_files(
game_file_descs: Vec<GameFileDescription>,
games_dir: String,
server_addr: SocketAddr,
) -> eyre::Result<()> {
let limits = Limits::default()
.with_max_handshake_duration(Duration::from_secs(3))?
.with_max_idle_timeout(Duration::from_secs(1))?;
let client = QuicClient::builder()
.with_tls(CERT_PEM)?
.with_io("0.0.0.0:0")?
.with_limits(limits)?
.start()?;
let conn = Connect::new(server_addr).with_server_name("localhost");
let mut conn = client.connect(conn).await?;
conn.keep_alive(true)?;
let game_file_descs = game_file_descs
.into_iter()
.filter(|desc| !desc.is_dir)
.collect::<Vec<_>>();
for file_desc in game_file_descs {
log::info!("downloading file: {}", file_desc.relative_path);
let stream = conn.open_bidirectional_stream().await?;
let (mut rx, mut tx) = stream.split();
let request = Request::GetGameFileData(file_desc.clone());
if let Ok(()) = tx.write_all(&request.encode()).await {
let path = PathBuf::from(&games_dir).join(&file_desc.relative_path);
let mut file = match File::create(&path) {
Ok(file) => file,
Err(e) => {
log::error!("failed to create file: {e}");
continue;
}
};
// if let Err(e) = tokio::io::copy(&mut rx, &mut file).await {
// log::error!("failed to download file: {e}");
// continue;
// }
while let Ok(Some(data)) = rx.receive().await {
if let Err(e) = file.write_all(&data) {
log::error!("failed to write to file: {e}");
break;
}
}
log::error!("file download complete: {}", path.display());
}
if let Err(e) = tx.close().await {
log::error!("failed to close stream: {e}");
}
}
Ok(())
}
struct Ctx {
game_dir: Arc<Mutex<Option<String>>>,
}
#[allow(clippy::too_many_lines)]
pub async fn run(
mut rx_control: UnboundedReceiver<ClientCommand>,
@ -69,6 +146,11 @@ pub async fn run(
}
};
// client context
let ctx = Ctx {
game_dir: Arc::new(Mutex::new(None)),
};
loop {
let limits = Limits::default()
.with_max_handshake_duration(Duration::from_secs(3))?
@ -106,11 +188,36 @@ pub async fn run(
while let Some(cmd) = rx_control.recv().await {
let request = match cmd {
ClientCommand::ListGames => Request::ListGames,
ClientCommand::GetGame(id) => Request::GetGame { id },
ClientCommand::ServerAddr(_) => Request::Invalid(
Bytes::new(),
"invalid control message (ServerAddr), should not happen".into(),
),
ClientCommand::GetGame(id) => {
log::debug!("requesting game from server: {id}");
Request::GetGame { id }
}
ClientCommand::ServerAddr(_) => {
log::warn!("unexpected ServerAddr command from UI client");
continue;
}
ClientCommand::SetGameDir(game_dir) => {
*ctx.game_dir.lock().await = Some(game_dir.clone());
continue;
}
ClientCommand::DownloadGameFiles(game_file_descs) => {
log::info!("got ClientCommand::DownloadGameFiles");
let games_dir = { ctx.game_dir.lock().await.clone() };
if let Some(games_dir) = games_dir {
tokio::task::spawn(async move {
if let Err(e) =
download_game_files(game_file_descs, games_dir, server_addr).await
{
log::error!("failed to download game files: {e}");
}
});
} else {
log::error!("Cannot handle game file descriptions: game_dir is not set");
}
continue;
}
};
let data = request.encode();
@ -152,7 +259,37 @@ pub async fn run(
log::info!("sent ClientEvent::ListGames to Tauri client");
}
}
Response::Game(game) => log::debug!("game received: {game:?}"),
Response::GetGame(game_file_descs) => {
log::info!(
"got {} game file descriptions from server",
game_file_descs.len()
);
let games_dir = { ctx.game_dir.lock().await.clone() };
match games_dir {
Some(games_dir) => {
game_file_descs.iter().filter(|f| f.is_dir).for_each(|dir| {
let path = PathBuf::from(&games_dir).join(&dir.relative_path);
if let Err(e) = std::fs::create_dir_all(path) {
log::error!("failed to create directory: {e}");
}
});
if let Err(e) =
tx_notify_ui.send(ClientEvent::GotGameFiles(game_file_descs))
{
log::error!(
"failed to send ClientEvent::GotGameFiles to client: {e}"
);
}
}
None => {
log::error!(
"Cannot handle game file descriptions: game_dir is not set"
);
}
}
}
Response::GameNotFound(id) => log::debug!("game not found {id}"),
Response::InvalidRequest(request_bytes, err) => log::error!(
"server says our request was invalid (error: {}): {}",

18
crates/lanspread-db/src/db.rs
View File

@ -133,3 +133,21 @@ impl Default for GameDB {
Self::empty()
}
}
#[derive(Clone, Serialize, Deserialize)]
pub struct GameFileDescription {
pub game_id: String,
pub relative_path: String,
pub is_dir: bool,
}
impl fmt::Debug for GameFileDescription {
#[allow(clippy::cast_precision_loss)]
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"game:{} path:{} dir:{}",
self.game_id, self.relative_path, self.is_dir
)
}
}

18
crates/lanspread-proto/src/lib.rs
View File

@ -1,5 +1,5 @@
use bytes::Bytes;
use lanspread_db::db::Game;
use lanspread_db::db::{Game, GameFileDescription};
use serde::{Deserialize, Serialize};
use tracing::error;
@ -8,6 +8,7 @@ pub enum Request {
Ping,
ListGames,
GetGame { id: String },
GetGameFileData(GameFileDescription),
Invalid(Bytes, String),
}
@ -15,7 +16,7 @@ pub enum Request {
pub enum Response {
Pong,
Games(Vec<Game>),
Game(Game),
GetGame(Vec<GameFileDescription>),
GameNotFound(String),
InvalidRequest(Bytes, String),
EncodingError(String),
@ -74,16 +75,3 @@ impl Message for Response {
}
}
}
// Helper methods for Response
impl Response {
#[must_use]
pub fn games(games: Vec<Game>) -> Self {
Response::Games(games)
}
#[must_use]
pub fn game(game: Game) -> Self {
Response::Game(game)
}
}

1
crates/lanspread-server/Cargo.toml
View File

@ -30,3 +30,4 @@ semver = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
walkdir = { workspace = true }

152
crates/lanspread-server/src/main.rs
View File

@ -1,15 +1,18 @@
#![allow(clippy::doc_markdown)]
use std::{
fs::File,
io::Read as _,
net::{IpAddr, SocketAddr},
path::PathBuf,
path::{Path, PathBuf},
sync::Arc,
};
use assets::Thumbnails;
use bytes::Bytes;
use clap::Parser;
use lanspread_compat::eti::{self, EtiGame};
use lanspread_db::db::{Game, GameDB};
use lanspread_db::db::{Game, GameDB, GameFileDescription};
use lanspread_mdns::{
DaemonEvent,
MdnsAdvertiser,
@ -21,6 +24,7 @@ use lanspread_utils::maybe_addr;
use s2n_quic::Server as QuicServer;
use tokio::{io::AsyncWriteExt, sync::Mutex};
use tracing_subscriber::EnvFilter;
use walkdir::WalkDir;
mod assets;
@ -30,6 +34,7 @@ static CERT_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../..
#[derive(Clone, Debug)]
struct ServerCtx {
handler: RequestHandler,
games_folder: PathBuf,
}
#[derive(Clone, Debug)]
@ -38,7 +43,7 @@ struct ConnectionCtx {
remote_addr: String,
}
async fn run(addr: SocketAddr, db: GameDB) -> eyre::Result<()> {
async fn run(addr: SocketAddr, db: GameDB, games_folder: PathBuf) -> eyre::Result<()> {
let mut server = QuicServer::builder()
.with_tls((CERT_PEM, KEY_PEM))?
.with_io(addr)?
@ -46,6 +51,7 @@ async fn run(addr: SocketAddr, db: GameDB) -> eyre::Result<()> {
let server_ctx = Arc::new(ServerCtx {
handler: RequestHandler::new(db),
games_folder,
});
while let Some(mut connection) = server.accept().await {
@ -75,7 +81,69 @@ async fn run(addr: SocketAddr, db: GameDB) -> eyre::Result<()> {
let request = Request::decode(data);
tracing::debug!("{} msg: {:?}", conn_ctx.remote_addr, request);
let response = conn_ctx.server_ctx.handler.handle_request(request).await;
if let Request::GetGameFileData(game_file_desc) = &request {
tracing::debug!(
"{} client requested game file data: {:?}",
conn_ctx.remote_addr,
game_file_desc
);
// deliver file data to client
let path = conn_ctx
.server_ctx
.games_folder
.join(&game_file_desc.relative_path);
if let Ok(mut file) = File::open(&path) {
let mut buf = vec![0; 64 * 1024];
while let Ok(n) = file.read(&mut buf) {
if n == 0 {
break;
}
if let Err(e) = tx.write_all(&buf[..n]).await {
tracing::error!(
"{} failed to send file data: {}",
conn_ctx.remote_addr,
e
);
}
}
// if let Err(e) = tokio::io::copy(&mut file, &mut tx).await {
// tracing::error!(
// "{} failed to send file data: {}",
// conn_ctx.remote_addr,
// e
// );
// }
} else {
tracing::error!(
"{} failed to open file: {}",
conn_ctx.remote_addr,
path.display()
);
}
if let Err(e) = tx.close().await {
tracing::error!(
"{} failed to close stream: {}",
conn_ctx.remote_addr,
e
);
}
continue;
}
let response = conn_ctx
.server_ctx
.handler
.handle_request(request, &conn_ctx)
.await;
tracing::trace!("{} server response: {:?}", conn_ctx.remote_addr, response);
let raw_response = response.encode();
tracing::trace!(
@ -102,6 +170,21 @@ async fn run(addr: SocketAddr, db: GameDB) -> eyre::Result<()> {
Ok(())
}
fn get_relative_path(base: &Path, deep_path: &Path) -> std::io::Result<PathBuf> {
let base_canonical = base.canonicalize()?;
let full_canonical = deep_path.canonicalize()?;
full_canonical
.strip_prefix(&base_canonical)
.map(std::path::Path::to_path_buf)
.map_err(|_| {
std::io::Error::new(
std::io::ErrorKind::Other,
"Path is not within base directory",
)
})
}
#[derive(Clone, Debug)]
struct RequestHandler {
db: Arc<Mutex<GameDB>>,
@ -114,7 +197,7 @@ impl RequestHandler {
}
}
async fn handle_request(&self, request: Request) -> Response {
async fn handle_request(&self, request: Request, conn_ctx: &ConnectionCtx) -> Response {
match request {
Request::Ping => Response::Pong,
Request::ListGames => {
@ -122,11 +205,52 @@ impl RequestHandler {
Response::Games(db.all_games().into_iter().cloned().collect())
}
Request::GetGame { id } => {
let db = self.db.lock().await;
match db.get_game_by_id(&id) {
Some(game) => Response::Game(game.clone()),
None => Response::GameNotFound(id),
if self.db.lock().await.get_game_by_id(&id).is_none() {
tracing::error!("Game not found in DB: {id}");
return Response::GameNotFound(id);
}
let games_folder = &conn_ctx.server_ctx.games_folder;
let game_dir = games_folder.join(&id);
if !game_dir.exists() {
tracing::error!("Game folder does not exist: {}", game_dir.display());
return Response::GameNotFound(id);
}
let mut game_files_descs: Vec<GameFileDescription> = vec![];
for entry in WalkDir::new(&game_dir)
.into_iter()
.filter_map(std::result::Result::ok)
{
match get_relative_path(games_folder, entry.path()) {
Ok(relative_path) => match relative_path.to_str() {
Some(relative_path) => {
let game_file_description = GameFileDescription {
game_id: id.clone(),
relative_path: relative_path.to_string(),
is_dir: entry.file_type().is_dir(),
};
tracing::debug!("Found game file: {:?}", game_file_description);
game_files_descs.push(game_file_description);
}
None => {
tracing::error!("Failed to get relative path: {relative_path:?}",);
}
},
Err(e) => {
tracing::error!("Failed to get relative path: {e}");
}
}
}
Response::GetGame(game_files_descs)
}
Request::GetGameFileData(_) => {
Response::InvalidRequest(Bytes::new(), "Not implemented".to_string())
}
Request::Invalid(data, err_msg) => {
tracing::error!(
@ -185,6 +309,14 @@ async fn main() -> eyre::Result<()> {
let cli = Cli::parse();
#[allow(clippy::manual_assert)]
if !cli.folder.exists() {
panic!(
"Games folder does not exist: {}",
cli.folder.to_str().expect("Invalid path")
);
}
let eti_games = eti::get_games(&cli.db).await?;
let mut games: Vec<Game> = eti_games.into_iter().map(eti_game_to_game).collect();
let thumbnails = Thumbnails::new(cli.thumbnails);
@ -218,5 +350,5 @@ async fn main() -> eyre::Result<()> {
tracing::info!("Server listening on {}:{}", cli.ip, cli.port);
run(SocketAddr::from((cli.ip, cli.port)), game_db).await
run(SocketAddr::from((cli.ip, cli.port)), game_db, cli.folder).await
}

23
crates/lanspread-tauri-deno-ts/src-tauri/src/lib.rs
View File

@ -28,7 +28,7 @@ fn request_games(state: tauri::State<LanSpreadState>) {
}
#[tauri::command]
fn run_game_backend(id: String, state: tauri::State<LanSpreadState>) -> String {
fn install_game(id: String, state: tauri::State<LanSpreadState>) -> String {
log::error!("Running game with id {id}");
// let result = Command::new(r#"C:\Users\ddidderr\scoop\apps\mpv\0.39.0\mpv.exe"#).spawn();
@ -63,6 +63,12 @@ fn set_game_install_state_from_path(game_db: &mut GameDB, path: &Path, installed
#[tauri::command]
fn update_game_directory(app_handle: tauri::AppHandle, path: String) {
app_handle
.state::<LanSpreadState>()
.client_ctrl
.send(ClientCommand::SetGameDir(path.clone()))
.unwrap();
let path = PathBuf::from(path);
if !path.exists() {
log::error!("game dir {path:?} does not exist");
@ -185,7 +191,7 @@ pub fn run() {
.plugin(tauri_logger_builder.build())
.plugin(tauri_plugin_shell::init())
.invoke_handler(tauri::generate_handler![
run_game_backend,
install_game,
request_games,
update_game_directory
])
@ -206,6 +212,19 @@ pub fn run() {
log::debug!("Received client event: ListGames");
update_game_db(games, app_handle.clone()).await;
}
ClientEvent::GotGameFiles(game_file_descs) => {
log::debug!("Received client event: GotGameFiles");
if let Err(e) = app_handle.emit("game-download-in-progress", Some(())) {
log::error!("Failed to emit game-files event: {e}");
}
app_handle
.state::<LanSpreadState>()
.inner()
.client_ctrl
.send(ClientCommand::DownloadGameFiles(game_file_descs))
.unwrap();
}
}
}
});

4
crates/lanspread-tauri-deno-ts/src/App.tsx
View File

@ -18,8 +18,6 @@ interface Game {
installed: boolean;
}
const App = () => {
const [gameItems, setGameItems] = useState<Game[]>([]);
const [searchTerm, setSearchTerm] = useState('');
@ -100,7 +98,7 @@ const App = () => {
const runGame = async (id: string) => {
console.log(`🎯 Running game with id=${id}`);
try {
const result = await invoke('run_game_backend', {id});
const result = await invoke('install_game', {id});
console.log(`✅ Game started, result=${result}`);
} catch (error) {
console.error('❌ Error running game:', error);