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

removed legacy crates

Browse Source
This commit is contained in:
ddidderr
2025-11-12 22:20:20 +01:00
parent c419bc0c89
commit b43327626d
8 changed files with 0 additions and 1005 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
-3
View File
@@ -1,5 +1,2 @@
- always check with `cargo clippy` and fix the issues - always check with `cargo clippy` and fix the issues
- always do a final `cargo +nightly fmt` after you're done with all changes - always do a final `cargo +nightly fmt` after you're done with all changes
# Legacy crates
The `lanspread-client` and `lanspread-server` crates are legacy. Not used anymore. Never touch them.
crates/lanspread-client/Cargo.toml
-29
View File
@@ -1,29 +0,0 @@
[package]
name = "lanspread-client"
version = "0.1.0"
edition = "2024"
[lints.rust]
unsafe_code = "forbid"
[lints.clippy]
pedantic = { level = "warn", priority = -1 }
todo = "warn"
unwrap_used = "warn"
[dependencies]
# local
lanspread-db = { path = "../lanspread-db" }
lanspread-proto = { path = "../lanspread-proto" }
lanspread-utils = { path = "../lanspread-utils" }
# external
bytes = { workspace = true }
clap = { workspace = true }
eyre = { workspace = true }
log = { workspace = true }
s2n-quic = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
crates/lanspread-client/src/lib.rs
-402
View File
@@ -1,402 +0,0 @@
#![allow(clippy::missing_errors_doc)]
use std::{fs::File, io::Write as _, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
use bytes::BytesMut;
use lanspread_db::db::{Game, GameFileDescription};
use lanspread_proto::{Message as _, Request, Response};
use lanspread_utils::maybe_addr;
use s2n_quic::{Client as QuicClient, Connection, client::Connect, provider::limits::Limits};
use tokio::{
io::AsyncWriteExt,
sync::{
RwLock,
mpsc::{UnboundedReceiver, UnboundedSender},
},
};
static CERT_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../cert.pem"));
#[derive(Debug)]
pub enum ClientEvent {
ListGames(Vec<Game>),
GotGameFiles {
id: String,
file_descriptions: Vec<GameFileDescription>,
},
DownloadGameFilesBegin {
id: String,
},
DownloadGameFilesFinished {
id: String,
},
DownloadGameFilesFailed {
id: String,
},
}
#[derive(Debug)]
pub enum ClientCommand {
ListGames,
GetGame(String),
DownloadGameFiles {
id: String,
file_descriptions: Vec<GameFileDescription>,
},
ServerAddr(SocketAddr),
SetGameDir(String),
}
async fn initial_server_alive_check(conn: &mut Connection) -> bool {
let stream = match conn.open_bidirectional_stream().await {
Ok(stream) => stream,
Err(e) => {
log::error!("failed to open stream: {e}");
return false;
}
};
let (mut rx, mut tx) = stream.split();
// send ping
if let Err(e) = tx.send(Request::Ping.encode()).await {
log::error!("failed to send ping to server: {e}");
return false;
}
let _ = tx.close().await;
// receive pong
if let Ok(Some(response)) = rx.receive().await {
let response = Response::decode(response);
if let Response::Pong = response {
log::info!("server is alive");
return true;
}
log::error!("server sent invalid response to ping: {response:?}");
}
false
}
async fn receive_game_file(
conn: &mut Connection,
desc: &GameFileDescription,
games_folder: &str,
) -> eyre::Result<()> {
log::info!("downloading: {desc:?}");
let stream = conn.open_bidirectional_stream().await?;
let (mut rx, mut tx) = stream.split();
let request = Request::GetGameFileData(desc.clone());
// request file
tx.write_all(&request.encode()).await?;
// create file
let path = PathBuf::from(&games_folder).join(&desc.relative_path);
let mut file = File::create(&path)?;
// receive file contents
while let Some(data) = rx.receive().await? {
file.write_all(&data)?;
}
log::debug!("file download complete: {}", path.display());
tx.close().await?;
Ok(())
}
async fn download_game_files(
game_id: &str,
game_file_descs: Vec<GameFileDescription>,
games_folder: String,
server_addr: SocketAddr,
tx_notify_ui: UnboundedSender<ClientEvent>,
) -> eyre::Result<()> {
let limits = Limits::default().with_max_handshake_duration(Duration::from_secs(3))?;
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_files = game_file_descs
.iter()
.filter(|desc| !desc.is_dir)
.filter(|desc| !desc.is_version_ini())
.collect::<Vec<_>>();
if game_files.is_empty() {
eyre::bail!("game_file_descs empty: no game files to download");
}
tx_notify_ui.send(ClientEvent::DownloadGameFilesBegin {
id: game_id.to_string(),
})?;
// receive all game files
for file_desc in game_files {
receive_game_file(&mut conn, file_desc, &games_folder).await?;
}
let version_file_desc = game_file_descs
.iter()
.find(|desc| desc.is_version_ini())
.ok_or_else(|| eyre::eyre!("version.ini not found"))?;
// receive version.ini
receive_game_file(&mut conn, version_file_desc, &games_folder).await?;
log::info!("all files downloaded for game: {game_id}");
tx_notify_ui.send(ClientEvent::DownloadGameFilesFinished {
id: game_id.to_string(),
})?;
Ok(())
}
struct Ctx {
game_dir: Arc<RwLock<Option<String>>>,
}
#[allow(clippy::too_many_lines)]
pub async fn run(
mut rx_control: UnboundedReceiver<ClientCommand>,
tx_notify_ui: UnboundedSender<ClientEvent>,
) -> eyre::Result<()> {
// blocking wait for remote address
log::debug!("waiting for server address");
let server_addr = loop {
if let Some(ClientCommand::ServerAddr(addr)) = rx_control.recv().await {
log::info!("got server address: {addr}");
break addr;
}
};
// client context
let ctx = Ctx {
game_dir: Arc::new(RwLock::new(None)),
};
loop {
let limits = Limits::default()
.with_max_handshake_duration(Duration::from_secs(3))?
.with_max_idle_timeout(Duration::from_secs(3))?;
let client = QuicClient::builder()
.with_tls(CERT_PEM)?
.with_io("0.0.0.0:0")?
.with_limits(limits)?
.start()?;
let connection = Connect::new(server_addr).with_server_name("localhost");
let mut conn = match client.connect(connection.clone()).await {
Ok(conn) => conn,
Err(e) => {
log::error!("failed to connect to server: {e}");
tokio::time::sleep(Duration::from_secs(3)).await;
continue;
}
};
conn.keep_alive(true)?;
if !initial_server_alive_check(&mut conn).await {
continue;
}
log::info!(
"connected: (server: {}) (client: {})",
maybe_addr!(conn.remote_addr()),
maybe_addr!(conn.local_addr())
);
// tx
while let Some(cmd) = rx_control.recv().await {
let request = match cmd {
ClientCommand::ListGames => Request::ListGames,
ClientCommand::GetGame(id) => {
log::info!("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.write().await = Some(game_dir.clone());
continue;
}
ClientCommand::DownloadGameFiles {
id,
file_descriptions,
} => {
log::info!("got ClientCommand::DownloadGameFiles");
let games_folder = { ctx.game_dir.read().await.clone() };
if let Some(games_folder) = games_folder {
let tx_notify_ui = tx_notify_ui.clone();
tokio::task::spawn(async move {
if let Err(e) = download_game_files(
&id,
file_descriptions,
games_folder,
server_addr,
tx_notify_ui.clone(),
)
.await
{
log::error!("failed to download game files: {e}");
if let Err(e) =
tx_notify_ui.send(ClientEvent::DownloadGameFilesFailed { id })
{
log::error!(
"failed to send DownloadGameFilesFailed event: {e}"
);
}
}
});
} else {
log::error!(
"Cannot handle game file descriptions: games_folder is not set"
);
}
continue;
}
};
// we got a command from the UI client
// but it is possible that we lost the connection to the server
// so we check and reconnect if needed
let mut retries = 0;
loop {
if initial_server_alive_check(&mut conn).await {
log::info!("server is back alive! 😊");
break;
}
if retries == 0 {
log::warn!("server connection lost, reconnecting...");
}
retries += 1;
conn = match client.connect(connection.clone()).await {
Ok(conn) => conn,
Err(e) => {
log::warn!("failed to connect to server: {e}");
log::warn!("retrying in 3 seconds...");
tokio::time::sleep(Duration::from_secs(3)).await;
continue;
}
};
}
let data = request.encode();
log::trace!("encoded data: {}", String::from_utf8_lossy(&data));
let stream = match conn.open_bidirectional_stream().await {
Ok(stream) => stream,
Err(e) => {
log::error!("failed to open stream: {e}");
break;
}
};
let (mut rx, mut tx) = stream.split();
if let Err(e) = tx.send(data).await {
log::error!("failed to send request to server {e:?}");
}
let mut data = BytesMut::new();
while let Ok(Some(bytes)) = rx.receive().await {
data.extend_from_slice(&bytes);
}
log::debug!("{} bytes received from server", data.len());
log::trace!("msg: (raw): {}", String::from_utf8_lossy(&data));
let response = Response::decode(data.freeze());
log::trace!("msg: {response:?}");
match response {
Response::ListGames(games) => {
for game in &games {
log::trace!("{game}");
}
if let Err(e) = tx_notify_ui.send(ClientEvent::ListGames(games)) {
log::error!("failed to send ClientEvent::ListGames to client {e:?}");
}
}
Response::GetGame {
id,
file_descriptions,
} => {
log::info!(
"got {} game file descriptions from server",
file_descriptions.len()
);
let games_folder = { ctx.game_dir.read().await.clone() };
match games_folder {
Some(games_folder) => {
// create all directories before receiving the actual files
file_descriptions
.iter()
.filter(|f| f.is_dir)
.for_each(|dir| {
let path =
PathBuf::from(&games_folder).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 {
id,
file_descriptions,
}) {
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: {}): {}",
err,
String::from_utf8_lossy(&request_bytes)
),
Response::EncodingError(err) => {
log::error!("server encoding error: {err}");
}
Response::DecodingError(data, err) => {
log::error!(
"response decoding error: {} (data: {})",
err,
String::from_utf8_lossy(&data)
);
}
Response::Pong => (), // ignore (should never happen)
}
if let Err(err) = tx.close().await {
log::error!("failed to close stream: {err}");
}
}
}
}
crates/lanspread-server/Cargo.toml
-37
View File
@@ -1,37 +0,0 @@
[package]
name = "lanspread-server"
version = "0.1.0"
edition = "2024"
[lints.rust]
unsafe_code = "forbid"
[lints.clippy]
pedantic = { level = "warn", priority = -1 }
todo = "warn"
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" }
lanspread-utils = { path = "../lanspread-utils" }
# external
bytes = { workspace = true }
chrono = { workspace = true }
clap = { workspace = true }
eyre = { workspace = true }
gethostname = { workspace = true }
itertools = { 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-server/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(crate) struct Cli {
/// IP address to bind to.
#[clap(long)]
pub(crate) ip: IpAddr,
/// Listen port.
#[clap(long)]
pub(crate) port: u16,
/// Game database path (SQLite).
#[clap(long)]
pub(crate) db: PathBuf,
/// Games folder.
#[clap(long)]
pub(crate) game_dir: PathBuf,
/// Thumbnails folder.
#[clap(long)]
pub(crate) thumbs_dir: PathBuf,
}
crates/lanspread-server/src/main.rs
-116
View File
@@ -1,116 +0,0 @@
use mimalloc::MiMalloc;
#[global_allocator]
static GLOBAL: MiMalloc = MiMalloc;
mod cli;
mod quic;
mod req;
use std::{convert::Into, 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 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 server 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!("Server listening on {server_addr}");
crate::quic::run_server(server_addr, game_db, cli.game_dir).await
}
crates/lanspread-server/src/quic.rs
-139
View File
@@ -1,139 +0,0 @@
use std::{net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
use lanspread_db::db::GameDB;
use lanspread_proto::{Message as _, Request};
use lanspread_utils::maybe_addr;
use s2n_quic::{Connection, Server, provider::limits::Limits, stream::BidirectionalStream};
use crate::req::{RequestHandler, send_game_file_chunk, send_game_file_data};
static KEY_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../key.pem"));
static CERT_PEM: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../cert.pem"));
#[derive(Clone, Debug)]
struct ServerCtx {
handler: RequestHandler,
games_folder: PathBuf,
}
async fn handle_bidi_stream(stream: BidirectionalStream, ctx: Arc<ServerCtx>) -> eyre::Result<()> {
let (mut rx, mut tx) = stream.split();
let remote_addr = maybe_addr!(rx.connection().remote_addr());
tracing::trace!("{remote_addr} stream opened");
// handle streams
loop {
match rx.receive().await {
Ok(Some(data)) => {
tracing::trace!(
"{remote_addr} msg: (raw): {}",
String::from_utf8_lossy(&data)
);
let request = Request::decode(data);
tracing::debug!("{remote_addr} msg: {request:?}");
// special case for now (send game file data to client)
match &request {
Request::GetGameFileData(game_file_desc) => {
send_game_file_data(game_file_desc, &mut tx, &ctx.games_folder).await;
continue;
}
Request::GetGameFileChunk {
game_id,
relative_path,
offset,
length,
} => {
send_game_file_chunk(
game_id,
relative_path,
*offset,
*length,
&mut tx,
&ctx.games_folder,
)
.await;
continue;
}
_ => {}
}
// normal case (handle request)
if let Err(e) = ctx
.handler
.handle_request(request, &ctx.games_folder, &mut tx)
.await
{
tracing::error!(?e, "{remote_addr} error handling request");
}
}
Ok(None) => {
tracing::trace!("{remote_addr} stream closed");
break;
}
Err(e) => {
tracing::error!("{remote_addr} stream error: {e}");
break;
}
}
}
Ok(())
}
async fn handle_connection(mut connection: Connection, ctx: Arc<ServerCtx>) -> eyre::Result<()> {
let remote_addr = maybe_addr!(connection.remote_addr());
tracing::info!("{remote_addr} connected");
// handle streams
while let Ok(Some(stream)) = connection.accept_bidirectional_stream().await {
let ctx = ctx.clone();
let remote_addr = remote_addr.clone();
// spawn a new task for the stream
tokio::spawn(async move {
if let Err(e) = handle_bidi_stream(stream, ctx).await {
tracing::error!("{remote_addr} stream error: {e}");
}
});
}
Ok(())
}
pub(crate) async fn run_server(
addr: SocketAddr,
db: GameDB,
games_folder: PathBuf,
) -> eyre::Result<()> {
let limits = Limits::default()
.with_max_handshake_duration(Duration::from_secs(3))?
.with_max_idle_timeout(Duration::from_secs(3))?;
let mut server = Server::builder()
.with_tls((CERT_PEM, KEY_PEM))?
.with_io(addr)?
.with_limits(limits)?
.start()?;
let ctx = Arc::new(ServerCtx {
handler: RequestHandler::new(db),
games_folder,
});
while let Some(connection) = server.accept().await {
let ctx = ctx.clone();
// spawn a new task for the connection
tokio::spawn(async move {
if let Err(e) = handle_connection(connection, ctx).await {
tracing::error!("Connection error: {}", e);
}
});
}
tracing::info!("Server shutting down");
Ok(())
}
crates/lanspread-server/src/req.rs
-256
View File
@@ -1,256 +0,0 @@
use std::{
path::{Path, PathBuf},
sync::Arc,
};
use bytes::Bytes;
use lanspread_db::db::{GameDB, GameFileDescription};
use lanspread_proto::{Message as _, Request, Response};
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(crate) struct RequestHandler {
db: Arc<RwLock<GameDB>>,
}
impl RequestHandler {
pub(crate) fn new(games: GameDB) -> RequestHandler {
RequestHandler {
db: Arc::new(RwLock::new(games)),
}
}
pub(crate) 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} server 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 {
0
} else {
match entry.metadata() {
Ok(metadata) => metadata.len(),
Err(e) => {
tracing::error!(
"Failed to read metadata for {}: {e}",
relative_path
);
// Return early since we can't proceed without file size
return Response::InvalidRequest(
format!("Failed to read metadata for {relative_path}")
.as_bytes()
.to_vec()
.into(),
"File size unavailable".to_string(),
);
}
}
};
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(crate) async fn process_request(&self, request: Request, games_folder: &Path) -> Response {
match request {
Request::Ping => RequestHandler::handle_ping(),
Request::ListGames => self.handle_list_games().await,
Request::GetGame { id } => self.handle_get_game(id, games_folder).await,
Request::GetGameFileData(_) | Request::GetGameFileChunk { .. } => {
RequestHandler::handle_get_game_file_data()
}
Request::Invalid(data, err_msg) => RequestHandler::handle_invalid(data, err_msg),
}
}
}
async fn stream_file_bytes(
tx: &mut SendStream,
base_dir: &Path,
relative_path: &str,
offset: u64,
length: Option<u64>,
) -> eyre::Result<()> {
let remote_addr = maybe_addr!(tx.connection().remote_addr());
let game_file = base_dir.join(relative_path);
tracing::debug!(
"{remote_addr} streaming file bytes: {:?}, offset: {offset}, length: {length:?}",
game_file
);
let mut file = tokio::fs::File::open(&game_file).await?;
if offset > 0 {
file.seek(std::io::SeekFrom::Start(offset)).await?;
}
let mut remaining = length.unwrap_or(u64::MAX);
let mut total_bytes = 0u64;
let mut last_total_bytes = 0u64;
let mut timestamp = Instant::now();
let mut buf = vec![0u8; 64 * 1024];
while remaining > 0 {
let read_len = std::cmp::min(remaining, buf.len() as u64);
let read_len = usize::try_from(read_len).unwrap_or(buf.len());
if read_len == 0 {
break;
}
let bytes_read = file.read(&mut buf[..read_len]).await?;
if bytes_read == 0 {
break;
}
tx.send(Bytes::copy_from_slice(&buf[..bytes_read])).await?;
remaining = remaining.saturating_sub(bytes_read as u64);
total_bytes += bytes_read as u64;
if last_total_bytes + 10_000_000 < total_bytes {
let elapsed = timestamp.elapsed();
let diff_bytes = total_bytes - last_total_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!(
"{remote_addr} sending file data: {:?}, MB/s: {mb_per_s:.2}",
game_file
);
last_total_bytes = total_bytes;
timestamp = Instant::now();
}
}
}
tracing::debug!(
"{remote_addr} finished streaming file bytes: {:?}, total_bytes: {total_bytes}",
game_file
);
tx.close().await?;
Ok(())
}
pub(crate) async fn send_game_file_data(
game_file_desc: &GameFileDescription,
tx: &mut SendStream,
game_dir: &Path,
) {
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!(
"{remote_addr} failed to stream file {}: {e}",
game_file_desc.relative_path
);
}
}
pub(crate) async fn send_game_file_chunk(
game_id: &str,
relative_path: &str,
offset: u64,
length: u64,
tx: &mut SendStream,
game_dir: &Path,
) {
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!(
"{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"))
}