use anyhow::{Context, Result}; use navka_common::navka::{relay_client::RelayClient, HidReport}; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tonic::Request; use crate::input::keyboard::KeyboardAggregator; pub struct NavkaClientApp { server_addr: String, } impl NavkaClientApp { pub fn new() -> Result { let addr = std::env::args() .nth(1) .or_else(|| std::env::var("NAVKA_SERVER_ADDR").ok()) .unwrap_or_else(|| "http://127.0.0.1:50051".to_owned()); Ok(Self { server_addr: addr }) } pub async fn run(&mut self) -> Result<()> { // 1) Connect to navka-server let mut client = RelayClient::connect(self.server_addr.clone()) .await .with_context(|| format!("failed to connect to {}", self.server_addr))?; // 2) Create a bidirectional streaming stub let (tx, rx) = mpsc::channel::(32); let outbound = ReceiverStream::new(rx); let response = client.stream(Request::new(outbound)).await?; let mut inbound = response.into_inner(); // 3) Start reading from all keyboards in a background task let mut aggregator = KeyboardAggregator::new(tx.clone()); aggregator.init_devices()?; // discover & grab let input_task = tokio::spawn(async move { if let Err(e) = aggregator.run().await { tracing::error!("KeyboardAggregator failed: {e}"); } }); // 4) Inbound loop: we do something with reports from the server, e.g. logging: while let Some(report) = inbound.message().await? { tracing::info!(?report.data, "echo from server"); } // 5) If inbound stream ends, stop the input task input_task.abort(); Ok(()) } }