diff --git a/.gitignore b/.gitignore index 431845e..9a39a2d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,4 @@ /target/ .idea/ +AGENTS.md +/data/ diff --git a/Cargo.lock b/Cargo.lock index 362bdab..ad3a8ef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -86,12 +86,71 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +dependencies = [ + "futures-core", + "futures-sink", +] + [[package]] name = "futures-core" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-executor" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-sink" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" + [[package]] name = "futures-task" version = "0.3.32" @@ -104,8 +163,13 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -269,9 +333,10 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" [[package]] name = "rusty-telemetry" -version = "0.1.0" +version = "0.2.0" dependencies = [ "chrono", + "futures", "serde", "serde_json", "tokio", diff --git a/Cargo.toml b/Cargo.toml index a9d58d1..c005d9f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rusty-telemetry" -version = "0.1.0" +version = "0.2.0" edition = "2021" description = "Sim racing telemetry receiver and analyzer" @@ -9,3 +9,4 @@ tokio = { version = "1", features = ["full"] } serde = { version = "1", features = ["derive"] } serde_json = "1" chrono = "0.4" +futures = "0.3" diff --git a/src/main.rs b/src/main.rs index 7176009..f7710e6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,74 +5,263 @@ use stats::ReceiveStats; use telemetry::TelemetryFrame; use std::io::Write; +use std::sync::Arc; +use tokio::io::{AsyncWriteExt, BufWriter}; -const DEFAULT_BIND_ADDR: &str = "127.0.0.1:5005"; +const VERSION: &str = env!("CARGO_PKG_VERSION"); const BUFFER_SIZE: usize = 65535; -fn print_banner(bind_addr: &str) { +fn bind_addr() -> String { + std::env::var("RUSTY_TELEMETRY_BIND").unwrap_or_else(|_| "127.0.0.1".to_string()) +} + +#[derive(Clone)] +enum FeedFormat { + Binary, + Json, +} + +#[derive(Clone)] +struct FeedConfig { + port: u16, + name: &'static str, + label: &'static str, + format: FeedFormat, +} + +static FEEDS: &[FeedConfig] = &[ + FeedConfig { + port: 9996, + name: "ksudp", + label: "KSUDP", + format: FeedFormat::Binary, + }, + FeedConfig { + port: 10101, + name: "telemetry_tool", + label: "Telemetry Tool", + format: FeedFormat::Binary, + }, + FeedConfig { + port: 5005, + name: "ac_plugin", + label: "AC Lua Plugin", + format: FeedFormat::Json, + }, +]; + +fn print_banner(results: &[(&FeedConfig, bool)], session_dir: &str, bind: &str) { println!("========================================"); - println!(" rusty-telemetry listener"); - println!(" Phase 1 - UDP JSON receiver"); + println!(" rusty-telemetry v{VERSION} — Multi-feed telemetry listener"); println!("========================================"); - println!(" Bind: {}", bind_addr); + for (feed, ok) in results { + let status = if *ok { "✅ Listening" } else { "❌ Failed" }; + println!( + " Feed :{} [{}] {}", + feed.port, feed.label, status + ); + } + println!("========================================"); + println!(" Bind: {bind}"); + println!(" Data dir: {session_dir}/"); println!("========================================"); println!(); } -#[tokio::main] -async fn main() { - let bind_addr = - std::env::var("RUSTY_TELEMETRY_BIND").unwrap_or_else(|_| DEFAULT_BIND_ADDR.to_string()); - - print_banner(&bind_addr); - - let socket = match tokio::net::UdpSocket::bind(&bind_addr).await { - Ok(s) => { - let local_addr = s.local_addr().unwrap(); - println!("[INFO] Listening on {}", local_addr); - s - } +async fn run_feed( + feed: FeedConfig, + socket: tokio::net::UdpSocket, + session_dir: String, + shutdown: Arc, +) { + let ext = match feed.format { + FeedFormat::Binary => "bin", + FeedFormat::Json => "json", + }; + let file_path = format!( + "{session_dir}/feed_{}_{}.{ext}", + feed.port, feed.name + ); + let file = match tokio::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&file_path) + .await + { + Ok(f) => f, Err(e) => { - eprintln!("[ERROR] Failed to bind UDP socket: {}", e); - std::process::exit(1); + eprintln!( + "[ERROR] Cannot open dump file for :{} ({}): {}", + feed.port, feed.name, e + ); + return; } }; + let mut file = BufWriter::new(file); let mut buf = vec![0u8; BUFFER_SIZE]; let mut stats = ReceiveStats::new(); let stdout = std::io::stdout(); loop { - match socket.recv_from(&mut buf).await { - Ok((len, src)) => { - stats.record(len); + tokio::select! { + result = socket.recv_from(&mut buf) => { + match result { + Ok((len, _src)) => { + stats.record(len); - let text = match std::str::from_utf8(&buf[..len]) { - Ok(s) => s, - Err(e) => { - eprintln!("[WARN] Invalid UTF-8 from {}: {}", src, e); - continue; + if let FeedFormat::Binary = feed.format { + let header = (len as u32).to_be_bytes(); + if let Err(e) = file.write_all(&header).await { + eprintln!( + "[ERROR] Write error on :{} ({}): {}", + feed.port, feed.name, e + ); + } + } + + if let Err(e) = file.write_all(&buf[..len]).await { + eprintln!( + "[ERROR] Write error on :{} ({}): {}", + feed.port, feed.name, e + ); + } + if let FeedFormat::Json = feed.format { + let _ = file.write_all(b"\n").await; + } + + { + let mut handle = stdout.lock(); + let fmt_tag = match feed.format { + FeedFormat::Binary => "Binary", + FeedFormat::Json => "JSON", + }; + let _ = writeln!( + handle, + "[ :{} | {} ] {fmt_tag} packet: {} bytes (#{})", + feed.port, + feed.label, + len, + stats.total_packets, + ); + if let FeedFormat::Json = feed.format { + if let Ok(text) = std::str::from_utf8(&buf[..len]) { + if let Ok(frame) = + serde_json::from_str::(text) + { + let _ = writeln!(handle, "{frame}"); + } else { + let _ = writeln!(handle, " [WARN] JSON parse failed"); + } + } + } + let _ = writeln!(handle, " {}", stats); + let _ = writeln!(handle); + } } - }; - - let frame: TelemetryFrame = match serde_json::from_str(text) { - Ok(f) => f, Err(e) => { - eprintln!("[WARN] JSON parse error from {}: {}", src, e); - continue; + eprintln!( + "[ERROR] Receive error on :{} ({}): {}", + feed.port, feed.name, e + ); } - }; - - { - let mut handle = stdout.lock(); - let _ = writeln!(handle, "{}", frame); - let _ = writeln!(handle, " [{}] {}", src, stats); - let _ = writeln!(handle); } } - Err(e) => { - eprintln!("[ERROR] Receive error: {}", e); + _ = shutdown.notified() => { + break; } } } + + if let Err(e) = file.flush().await { + eprintln!( + "[ERROR] Flush error on :{} ({}): {}", + feed.port, feed.name, e + ); + } + if let Err(e) = file.get_ref().sync_all().await { + eprintln!( + "[ERROR] Sync error on :{} ({}): {}", + feed.port, feed.name, e + ); + } + eprintln!( + "[INFO] Feed :{} ({}) flushed and closed", + feed.port, feed.name + ); +} + +#[tokio::main] +async fn main() { + let bind = bind_addr(); + + let session_id = chrono::Local::now().format("%Y%m%d_%H%M%S").to_string(); + let session_dir = format!("./data/{session_id}"); + + if let Err(e) = tokio::fs::create_dir_all(&session_dir).await { + eprintln!("[ERROR] Cannot create data dir {}: {}", session_dir, e); + std::process::exit(1); + } + + let mut banner_data: Vec<(&FeedConfig, bool)> = Vec::new(); + let mut bound_feeds: Vec<(FeedConfig, tokio::net::UdpSocket)> = Vec::new(); + + for feed in FEEDS { + let addr = format!("{bind}:{}", feed.port); + match tokio::net::UdpSocket::bind(&addr).await { + Ok(socket) => { + banner_data.push((feed, true)); + bound_feeds.push((feed.clone(), socket)); + } + Err(e) => { + eprintln!( + "[WARN] Failed to bind :{} ({}): {} — skipping", + feed.port, feed.label, e + ); + banner_data.push((feed, false)); + } + } + } + + print_banner(&banner_data, &session_dir, &bind); + + if bound_feeds.is_empty() { + eprintln!("[ERROR] No feeds could bind. Exiting."); + std::process::exit(1); + } + + let shutdown = Arc::new(tokio::sync::Notify::new()); + + let mut tasks = Vec::new(); + for (feed, socket) in bound_feeds { + let dir = session_dir.clone(); + let shutdown = shutdown.clone(); + tasks.push(tokio::spawn(async move { + run_feed(feed, socket, dir, shutdown).await; + })); + } + + tokio::signal::ctrl_c().await.ok(); + println!(); + println!("[INFO] Ctrl+C received, shutting down..."); + + shutdown.notify_waiters(); + + match tokio::time::timeout( + std::time::Duration::from_secs(3), + futures::future::join_all(tasks), + ) + .await + { + Ok(results) => { + for r in results { + if let Err(e) = r { + eprintln!("[WARN] Task join error: {e}"); + } + } + } + Err(_) => { + eprintln!("[WARN] Timeout waiting for feeds to flush; forcing exit"); + } + } }