bstein/lesavka
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

server updates

Browse Source
This commit is contained in:
Brad Stein 2025-06-16 19:19:14 -05:00
parent de2c99731f
commit 59f02adcf8
2 changed files with 62 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client/src/app.rs
View File

@ -26,7 +26,7 @@ impl NavkaClientApp {
.or_else(|| std::env::var("NAVKA_SERVER_ADDR").ok()) .or_else(|| std::env::var("NAVKA_SERVER_ADDR").ok())
.unwrap_or_else(|| "http://127.0.0.1:50051".to_owned()); .unwrap_or_else(|| "http://127.0.0.1:50051".to_owned());
let (tx, _) = broadcast::channel::<HidReport>(128); let (tx, _) = broadcast::channel::<HidReport>(2048); // 🖱️ + ⌨️ burst‑proof
let mut aggregator = InputAggregator::new(dev_mode, tx.clone()); let mut aggregator = InputAggregator::new(dev_mode, tx.clone());
aggregator.init()?; // discover & grab aggregator.init()?; // discover & grab

104
server/src/main.rs
View File

@ -36,55 +36,73 @@ impl Relay for Handler {
let (tx, rx) = tokio::sync::mpsc::channel(32); let (tx, rx) = tokio::sync::mpsc::channel(32);
tokio::spawn(async move { tokio::spawn(async move {
loop { // catch panics so that they are logged instead of killing the task silently
match in_stream.next().await { let task = std::panic::AssertUnwindSafe(async move {
/* ──────────────── message received ──────────────── */ // perpetually read client → server messages
Some(Ok(msg)) => { while let Some(res) = in_stream.next().await {
// 1. write to the right gadget --------------------------------- match res {
let io_res = match &msg.kind { /* ──────────────── message received ──────────────── */
Some(hid_report::Kind::KeyboardReport(v)) if v.len() == 8 => { Ok(msg) => {
kb.lock().await.write_all(v).await.map(|_| "⌨️ → /dev/hidg0 (8 B)") debug!("📥 recv {:?}", &msg.kind); // <‑‑ always log
// 1. write to the right gadget ---------------------------------
let io_res = match &msg.kind {
Some(hid_report::Kind::KeyboardReport(v)) if v.len() == 8 => {
kb.lock().await.write_all(v).await.map(|_| "⌨️ → /dev/hidg0 (8 B)")
}
Some(hid_report::Kind::MouseReport(v)) if v.len() == 4 => {
ms.lock().await.write_all(v).await.map(|_| "🖱️ → /dev/hidg1 (4 B)")
}
_ => {
error!(?msg.kind, "⚠️ malformed packet");
continue; // skip echo
}
};
// 2. I/O result -------------------------------------------------
match io_res {
Ok(msg_txt) => info!("{msg_txt}"),
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
trace!("🐛 gadget busy, dropped packet");
continue; // skip echo
}
Err(e) => {
error!("write error: {e}");
continue; // skip echo
}
} }
Some(hid_report::Kind::MouseReport(v)) if v.len() == 4 => {
ms.lock().await.write_all(v).await.map(|_| "🖱️ → /dev/hidg1 (4 B)") // 3. echo back (best‑effort) -----------------------------------
} if tx.try_send(Ok(msg)).is_err() {
_ => { trace!("↩️ echo buffer full – dropped");
error!(?msg.kind, "⚠️ malformed packet");
continue; // skip echo
}
};
// 2. I/O result -------------------------------------------------
match io_res {
Ok(msg_txt) => info!("{msg_txt}"),
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
trace!("🐛 gadget busy, dropped packet");
continue; // skip echo
}
Err(e) => {
error!("write error: {e}");
continue; // skip echo
} }
} }
// 3. echo back (best‑effort) ----------------------------------- /* ──────────────── benign back‑pressure error ──────────────── */
let _ = tx.try_send(Ok(msg)); Err(status) => {
// Tonic delivers back‑pressure as UNKNOWN / INTERNAL.
// They are *not* fatal for us – log & continue.
warn!("🐛 gRPC back‑pressure: {status}");
continue; // keep the stream alive
}
} }
}
/* ──────────────── benign back‑pressure error ──────────────── */ info!("🔚 client closed the upstream");
Some(Err(status)) => { Ok::<(), Status>(())
trace!("grpc recv error (ignored): {status}"); })
continue; // keep the stream alive .catch_unwind()
} .await;
/* ──────────────── client closed the stream ──────────────── */ if let Err(panic) = task {
None => break, // print the panic payload – this is what killed the stream earlier
if let Some(s) = panic.downcast_ref::<&str>() {
error!("‼️ stream task panicked: {s}");
} else if let Some(s) = panic.downcast_ref::<String>() {
error!("‼️ stream task panicked: {s}");
} else {
error!("‼️ stream task panicked with unknown payload");
} }
} }
info!("🔚 client stream closed");
// dropping `tx` here terminates the server→client stream gracefully
Ok::<(), Status>(())
}); });
Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) Ok(Response::new(Box::pin(ReceiverStream::new(rx))))