jan/rusty-telemetry
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: multi-port UDP listener with data dump

Browse source 
Listen on three AC telemetry ports simultaneously (9996, 10101, 5005).
Dump received data to per-feed files under ./data/<session>/.
Notify on console for each received packet.

- Port 9996: AC built-in (ksudp), binary with 4-byte length framing
- Port 10101: Telemetry Tool plugin, binary with 4-byte length framing
- Port 5005: AC Lua plugin, JSON with newline-separated records
- Configurable bind address via RUSTY_TELEMETRY_BIND (default 127.0.0.1)
- Buffered file I/O via tokio BufWriter
- Graceful shutdown with flush on Ctrl+C
- Startup banner showing feed status
- Bumped version to 0.2.0, added futures dependency
This commit is contained in:
Jan-Peter von Hunnius 2026-06-03 11:50:06 +02:00
parent 30543fd0ba
commit 1731495197
4 changed files with 302 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -1,2 +1,4 @@
/target/
.idea/
AGENTS.md
/data/

67
Cargo.lock generated
View file

@ -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",

3
Cargo.toml
View file

@ -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"

263
src/main.rs
View file

@ -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<tokio::sync::Notify>,
) {
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)) => {
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
);
}
}
};
let frame: TelemetryFrame = match serde_json::from_str(text) {
Ok(f) => f,
Err(e) => {
eprintln!("[WARN] JSON parse error from {}: {}", src, e);
continue;
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 _ = writeln!(handle, "{}", frame);
let _ = writeln!(handle, " [{}] {}", src, stats);
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::<TelemetryFrame>(text)
{
let _ = writeln!(handle, "{frame}");
} else {
let _ = writeln!(handle, " [WARN] JSON parse failed");
}
}
}
let _ = writeln!(handle, " {}", stats);
let _ = writeln!(handle);
}
}
Err(e) => {
eprintln!("[ERROR] Receive error: {}", e);
eprintln!(
"[ERROR] Receive error on :{} ({}): {}",
feed.port, feed.name, 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");
}
}
}