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

wip

Browse Source
This commit is contained in:
ddidderr
2025-11-11 21:30:26 +01:00
parent d831179783
commit 9c1b94fa6a
9 changed files with 191 additions and 450 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/lanspread-peer/Cargo.toml
-8
View File
@@ -13,7 +13,6 @@ unwrap_used = "warn"
[dependencies]
# local
lanspread-compat = { path = "../lanspread-compat" }
lanspread-db = { path = "../lanspread-db" }
lanspread-mdns = { path = "../lanspread-mdns" }
lanspread-proto = { path = "../lanspread-proto" }
@@ -21,18 +20,11 @@ lanspread-utils = { path = "../lanspread-utils" }
# external
bytes = { workspace = true }
chrono = { workspace = true }
clap = { workspace = true }
eyre = { workspace = true }
gethostname = { workspace = true }
itertools = { workspace = true }
log = { workspace = true }
mimalloc = { workspace = true }
s2n-quic = { workspace = true }
serde_json = { workspace = true }
semver = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
uuid = { workspace = true }
walkdir = { workspace = true }
crates/lanspread-peer/src/cli.rs
-23
View File
@@ -1,23 +0,0 @@
use std::{net::IpAddr, path::PathBuf};
use clap::Parser;
#[allow(clippy::doc_markdown)]
#[derive(Debug, Parser)]
pub struct Cli {
/// IP address to bind to.
#[clap(long)]
pub ip: IpAddr,
/// Listen port.
#[clap(long)]
pub port: u16,
/// Game database path (SQLite).
#[clap(long)]
pub db: PathBuf,
/// Games folder.
#[clap(long)]
pub game_dir: PathBuf,
/// Thumbnails folder.
#[clap(long)]
pub thumbs_dir: PathBuf,
}
crates/lanspread-peer/src/lib.rs
+24 -2
View File
@@ -34,6 +34,30 @@ use uuid::Uuid;
use crate::peer::{send_game_file_chunk, send_game_file_data};
/// Initialize and start the peer system
/// This function replaces the main.rs entry point and allows the peer to be started from other crates
pub fn start_peer(
game_dir: String,
tx_notify_ui: UnboundedSender<PeerEvent>,
) -> eyre::Result<UnboundedSender<PeerCommand>> {
log::info!("Starting peer system with game directory: {game_dir}");
let (tx_control, rx_control) = tokio::sync::mpsc::unbounded_channel();
// Start the peer in a background task
let tx_control_clone = tx_control.clone();
tokio::spawn(async move {
if let Err(e) = run_peer(rx_control, tx_notify_ui).await {
log::error!("Peer system failed: {e}");
}
});
// Set the game directory
tx_control.send(PeerCommand::SetGameDir(game_dir))?;
Ok(tx_control_clone)
}
static CERT_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../cert.pem"));
static KEY_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../key.pem"));
@@ -769,7 +793,6 @@ struct Ctx {
struct PeerCtx {
game_dir: Arc<RwLock<Option<String>>>,
local_game_db: Arc<RwLock<Option<GameDB>>>,
peer_game_db: Arc<RwLock<PeerGameDB>>,
}
pub async fn run_peer(
@@ -786,7 +809,6 @@ pub async fn run_peer(
let peer_ctx = PeerCtx {
game_dir: ctx.game_dir.clone(),
local_game_db: ctx.local_game_db.clone(),
peer_game_db: ctx.peer_game_db.clone(),
};
// Start server component
crates/lanspread-peer/src/main.rs
-178
View File
@@ -1,178 +0,0 @@
use mimalloc::MiMalloc;
#[global_allocator]
static GLOBAL: MiMalloc = MiMalloc;
mod cli;
mod peer;
use std::{net::SocketAddr, time::Duration};
use clap::Parser as _;
use cli::Cli;
use gethostname::gethostname;
use lanspread_compat::eti;
use lanspread_db::db::{Game, GameDB};
use lanspread_mdns::{DaemonEvent, LANSPREAD_SERVICE_TYPE, MdnsAdvertiser};
use lanspread_peer::{PeerCommand, PeerEvent, run_peer};
use tracing_subscriber::EnvFilter;
use uuid::Uuid;
fn spawn_mdns_task(server_addr: SocketAddr) -> eyre::Result<()> {
let peer_id = Uuid::now_v7().simple().to_string();
let hostname = gethostname();
let hostname_str = hostname.to_str().unwrap_or("");
// Calculate maximum hostname length that fits with UUID in 63 char limit
let max_hostname_len = 63usize.saturating_sub(peer_id.len() + 1); // +1 for the dash
let truncated_hostname = if hostname_str.len() > max_hostname_len {
hostname_str.get(..max_hostname_len).unwrap_or(hostname_str)
} else {
hostname_str
};
let combined_str = if truncated_hostname.is_empty() {
// If no hostname is available, use just the UUID
peer_id
} else {
format!("{truncated_hostname}-{peer_id}")
};
let mdns = MdnsAdvertiser::new(LANSPREAD_SERVICE_TYPE, &combined_str, server_addr)?;
tokio::spawn(async move {
while let Ok(event) = mdns.monitor.recv() {
tracing::trace!("mDNS: {:?}", &event);
if let DaemonEvent::Error(e) = event {
tracing::error!("mDNS: {e}");
tokio::time::sleep(Duration::from_secs(1)).await;
}
}
});
Ok(())
}
async fn prepare_game_db(cli: &Cli) -> eyre::Result<GameDB> {
// build games from ETI database
let mut games: Vec<Game> = eti::get_games(&cli.db)
.await?
.into_iter()
.map(Into::into)
.collect();
// filter out games that the peer does not have in game_dir
games.retain(|game| cli.game_dir.join(&game.id).is_dir());
// read version.ini files and update eti_game_version
for game in &mut games {
let game_dir = cli.game_dir.join(&game.id);
if let Ok(version) = lanspread_db::db::read_version_from_ini(&game_dir) {
game.eti_game_version = version;
if let Some(ref version) = game.eti_game_version {
tracing::debug!("Read version for game {}: {}", game.id, version);
}
} else {
tracing::warn!("Failed to read version.ini for game: {}", game.id);
}
}
let mut game_db = GameDB::from(games);
game_db.add_thumbnails(&cli.thumbs_dir);
game_db.all_games().iter().for_each(|game| {
tracing::debug!("Found game: {game}");
});
tracing::info!("Prepared game database with {} games", game_db.games.len());
Ok(game_db)
}
#[tokio::main]
async fn main() -> eyre::Result<()> {
tracing_subscriber::fmt()
.with_env_filter(EnvFilter::from_default_env())
.init();
let cli = Cli::parse();
assert!(
cli.game_dir.exists(),
"Games folder does not exist: {}",
cli.game_dir.to_str().expect("Invalid path")
);
let server_addr = SocketAddr::from((cli.ip, cli.port));
// spawn mDNS listener task
spawn_mdns_task(server_addr)?;
let _game_db = prepare_game_db(&cli).await?;
tracing::info!("Peer listening on {server_addr}");
let (tx_control, rx_control) = tokio::sync::mpsc::unbounded_channel();
let (tx_notify_ui, mut rx_notify_ui) = tokio::sync::mpsc::unbounded_channel();
// Start peer task
let peer_task = tokio::spawn(async move { run_peer(rx_control, tx_notify_ui).await });
// Handle events from peer
let event_handler = tokio::spawn(async move {
while let Some(event) = rx_notify_ui.recv().await {
match event {
PeerEvent::ListGames(games) => {
tracing::info!("Received list of {} games", games.len());
}
PeerEvent::GotGameFiles {
id,
file_descriptions,
} => {
tracing::info!(
"Got game files for {}: {} files",
id,
file_descriptions.len()
);
}
PeerEvent::DownloadGameFilesBegin { id } => {
tracing::info!("Download started for game: {}", id);
}
PeerEvent::DownloadGameFilesFinished { id } => {
tracing::info!("Download finished for game: {}", id);
}
PeerEvent::DownloadGameFilesFailed { id } => {
tracing::error!("Download failed for game: {}", id);
}
PeerEvent::PeerConnected(addr) => {
tracing::info!("Peer connected: {}", addr);
}
PeerEvent::PeerDisconnected(addr) => {
tracing::info!("Peer disconnected: {}", addr);
}
PeerEvent::PeerDiscovered(addr) => {
tracing::info!("Peer discovered: {}", addr);
}
PeerEvent::PeerLost(addr) => {
tracing::info!("Peer lost: {}", addr);
}
}
}
});
// Set the game directory from CLI args
if let Err(e) = tx_control.send(PeerCommand::SetGameDir(
cli.game_dir.to_string_lossy().to_string(),
)) {
tracing::error!("Failed to send SetGameDir command: {e}");
}
// TODO: Add additional CLI interaction or other peer discovery logic here
// Wait for tasks
let (peer_result, _) = tokio::join!(peer_task, event_handler);
peer_result??;
Ok(())
}
crates/lanspread-peer/src/peer.rs
+7 -149
View File
@@ -1,145 +1,13 @@
use std::{
convert::TryInto,
path::{Path, PathBuf},
sync::Arc,
};
use std::{convert::TryInto, path::Path};
use bytes::Bytes;
use lanspread_db::db::{GameDB, GameFileDescription};
use lanspread_proto::{Message as _, Request, Response};
use lanspread_db::db::GameFileDescription;
use lanspread_utils::maybe_addr;
use s2n_quic::stream::SendStream;
use tokio::{
io::{AsyncReadExt, AsyncSeekExt},
sync::RwLock,
time::Instant,
};
use walkdir::WalkDir;
#[derive(Clone, Debug)]
pub struct PeerRequestHandler {
db: Arc<RwLock<GameDB>>,
}
impl PeerRequestHandler {
pub fn new(games: GameDB) -> PeerRequestHandler {
PeerRequestHandler {
db: Arc::new(RwLock::new(games)),
}
}
pub async fn handle_request(
&self,
request: Request,
games_folder: &Path,
tx: &mut SendStream,
) -> eyre::Result<()> {
let remote_addr = maybe_addr!(tx.connection().remote_addr());
// process request and generate response
let response = self.process_request(request, games_folder).await;
tracing::trace!("{remote_addr} peer response: {response:?}");
// write response back to client
tx.send(response.encode()).await?;
// close the stream
tx.close().await?;
Ok(())
}
fn handle_ping() -> Response {
Response::Pong
}
async fn handle_list_games(&self) -> Response {
let db = self.db.read().await;
Response::ListGames(db.all_games().into_iter().cloned().collect())
}
async fn handle_get_game(&self, id: String, games_folder: &Path) -> Response {
if self.db.read().await.get_game_by_id(&id).is_none() {
tracing::error!("Game not found in DB: {id}");
return Response::GameNotFound(id);
}
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 is_dir = entry.file_type().is_dir();
let size = if is_dir {
None
} else {
match entry.metadata() {
Ok(metadata) => Some(metadata.len()),
Err(e) => {
tracing::error!(
"Failed to read metadata for {}: {e}",
relative_path
);
None
}
}
};
let game_file_description = GameFileDescription {
game_id: id.clone(),
relative_path: relative_path.to_string(),
is_dir,
size,
};
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 {
id,
file_descriptions: game_files_descs,
}
}
fn handle_get_game_file_data() -> Response {
Response::InvalidRequest(Bytes::new(), "Not implemented".to_string())
}
fn handle_invalid(data: Bytes, err_msg: String) -> Response {
Response::InvalidRequest(data, err_msg)
}
pub async fn process_request(&self, request: Request, games_folder: &Path) -> Response {
match request {
Request::Ping => PeerRequestHandler::handle_ping(),
Request::ListGames => self.handle_list_games().await,
Request::GetGame { id } => self.handle_get_game(id, games_folder).await,
Request::GetGameFileData(_) => PeerRequestHandler::handle_get_game_file_data(),
Request::GetGameFileChunk { .. } => PeerRequestHandler::handle_get_game_file_data(),
Request::Invalid(data, err_msg) => PeerRequestHandler::handle_invalid(data, err_msg),
}
}
}
async fn stream_file_bytes(
tx: &mut SendStream,
@@ -150,7 +18,7 @@ async fn stream_file_bytes(
) -> eyre::Result<()> {
let remote_addr = maybe_addr!(tx.connection().remote_addr());
let game_file = base_dir.join(relative_path);
tracing::debug!(
log::debug!(
"{remote_addr} streaming file bytes for peer: {:?}, offset: {offset}, length: {length:?}",
game_file
);
@@ -189,7 +57,7 @@ async fn stream_file_bytes(
if elapsed.as_secs_f64() >= 1.0 {
#[allow(clippy::cast_precision_loss)]
let mb_per_s = (diff_bytes as f64) / (elapsed.as_secs_f64() * 1_000_000.0);
tracing::debug!(
log::debug!(
"{remote_addr} sending file data: {:?}, MB/s: {mb_per_s:.2}",
game_file
);
@@ -199,7 +67,7 @@ async fn stream_file_bytes(
}
}
tracing::debug!(
log::debug!(
"{remote_addr} finished streaming file bytes: {:?}, total_bytes: {total_bytes}",
game_file
);
@@ -215,7 +83,7 @@ pub async fn send_game_file_data(
) {
if let Err(e) = stream_file_bytes(tx, game_dir, &game_file_desc.relative_path, 0, None).await {
let remote_addr = maybe_addr!(tx.connection().remote_addr());
tracing::error!(
log::error!(
"{remote_addr} failed to stream file {}: {e}",
game_file_desc.relative_path
);
@@ -232,18 +100,8 @@ pub async fn send_game_file_chunk(
) {
if let Err(e) = stream_file_bytes(tx, game_dir, relative_path, offset, Some(length)).await {
let remote_addr = maybe_addr!(tx.connection().remote_addr());
tracing::error!(
log::error!(
"{remote_addr} failed to stream chunk {game_id}/{relative_path} offset {offset} length {length}: {e}"
);
}
}
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::other("Path is not within base directory"))
}