code cleanup + purge anyhow in library code (#157)

src/server/emulation_task.rs

@@ -1,12 +1,16 @@
-use anyhow::{anyhow, Result};
 use std::net::SocketAddr;
 
+use thiserror::Error;
 use tokio::{
     sync::mpsc::{Receiver, Sender},
     task::JoinHandle,
 };
 
-use crate::{client::ClientHandle, config::EmulationBackend, server::State};
+use crate::{
+    client::{ClientHandle, ClientManager},
+    config::EmulationBackend,
+    server::State,
+};
 use input_emulation::{
     self,
     error::{EmulationCreationError, EmulationError},
@@ -14,7 +18,7 @@ use input_emulation::{
 };
 use input_event::{Event, KeyboardEvent};
 
-use super::{CaptureEvent, Server};
+use super::{network_task::NetworkError, CaptureEvent, Server};
 
 #[derive(Clone, Debug)]
 pub enum EmulationEvent {
@@ -31,51 +35,73 @@ pub enum EmulationEvent {
 pub fn new(
     backend: Option<EmulationBackend>,
     server: Server,
-    mut udp_rx: Receiver<Result<(Event, SocketAddr)>>,
+    udp_rx: Receiver<Result<(Event, SocketAddr), NetworkError>>,
     sender_tx: Sender<(Event, SocketAddr)>,
     capture_tx: Sender<CaptureEvent>,
     timer_tx: Sender<()>,
-) -> Result<(JoinHandle<Result<()>>, Sender<EmulationEvent>), EmulationCreationError> {
-    let (tx, mut rx) = tokio::sync::mpsc::channel(32);
-    let emulate_task = tokio::task::spawn_local(async move {
-        let backend = backend.map(|b| b.into());
-        let mut emulate = input_emulation::create(backend).await?;
-        let mut last_ignored = None;
+) -> (
+    JoinHandle<Result<(), LanMouseEmulationError>>,
+    Sender<EmulationEvent>,
+) {
+    let (tx, rx) = tokio::sync::mpsc::channel(32);
+    let emulation_task =
+        emulation_task(backend, rx, server, udp_rx, sender_tx, capture_tx, timer_tx);
+    let emulate_task = tokio::task::spawn_local(emulation_task);
+    (emulate_task, tx)
+}
 
-        loop {
-            tokio::select! {
-                udp_event = udp_rx.recv() => {
-                    let udp_event = udp_event.ok_or(anyhow!("receiver closed"))??;
-                    handle_udp_rx(&server, &capture_tx, &mut emulate, &sender_tx, &mut last_ignored, udp_event, &timer_tx).await?;
-                }
-                emulate_event = rx.recv() => {
-                    match emulate_event {
-                        Some(e) => match e {
-                            EmulationEvent::Create(h) => emulate.create(h).await,
-                            EmulationEvent::Destroy(h) => emulate.destroy(h).await,
-                            EmulationEvent::ReleaseKeys(c) => release_keys(&server, &mut emulate, c).await?,
-                            EmulationEvent::Terminate => break,
-                        },
-                        None => break,
+#[derive(Debug, Error)]
+pub enum LanMouseEmulationError {
+    #[error("error creating input-emulation: `{0}`")]
+    Create(#[from] EmulationCreationError),
+    #[error("error emulating input: `{0}`")]
+    Emulate(#[from] EmulationError),
+}
+
+async fn emulation_task(
+    backend: Option<EmulationBackend>,
+    mut rx: Receiver<EmulationEvent>,
+    server: Server,
+    mut udp_rx: Receiver<Result<(Event, SocketAddr), NetworkError>>,
+    sender_tx: Sender<(Event, SocketAddr)>,
+    capture_tx: Sender<CaptureEvent>,
+    timer_tx: Sender<()>,
+) -> Result<(), LanMouseEmulationError> {
+    let backend = backend.map(|b| b.into());
+    let mut emulation = input_emulation::create(backend).await?;
+
+    let mut last_ignored = None;
+    loop {
+        tokio::select! {
+            udp_event = udp_rx.recv() => {
+                let udp_event = match udp_event {
+                    Some(Ok(e)) => e,
+                    Some(Err(e)) => {
+                        log::warn!("network error: {e}");
+                        continue;
                     }
+                    None => break,
+                };
+                handle_udp_rx(&server, &capture_tx, &mut emulation, &sender_tx, &mut last_ignored, udp_event, &timer_tx).await?;
+            }
+            emulate_event = rx.recv() => {
+                match emulate_event {
+                    Some(e) => match e {
+                        EmulationEvent::Create(h) => emulation.create(h).await,
+                        EmulationEvent::Destroy(h) => emulation.destroy(h).await,
+                        EmulationEvent::ReleaseKeys(c) => release_keys(&server, &mut emulation, c).await?,
+                        EmulationEvent::Terminate => break,
+                    },
+                    None => break,
                 }
             }
         }
+    }
 
-        // release potentially still pressed keys
-        let clients = server
-            .client_manager
-            .borrow()
-            .get_client_states()
-            .map(|(h, _)| h)
-            .collect::<Vec<_>>();
-        for client in clients {
-            release_keys(&server, &mut emulate, client).await?;
-        }
+    // release potentially still pressed keys
+    release_all_keys(&server, &mut emulation).await?;
 
-        anyhow::Ok(())
-    });
-    Ok((emulate_task, tx))
+    Ok(())
 }
 
 async fn handle_udp_rx(
@@ -89,38 +115,15 @@ async fn handle_udp_rx(
 ) -> Result<(), EmulationError> {
     let (event, addr) = event;
 
-    // get handle for addr
-    let handle = match server.client_manager.borrow().get_client(addr) {
-        Some(a) => a,
-        None => {
-            if last_ignored.is_none() || last_ignored.is_some() && last_ignored.unwrap() != addr {
-                log::warn!("ignoring events from client {addr}");
-                last_ignored.replace(addr);
-            }
-            return Ok(());
-        }
+    log::trace!("{:20} <-<-<-<------ {addr}", event.to_string());
+
+    // get client handle for addr
+    let Some(handle) =
+        activate_client_if_exists(&mut server.client_manager.borrow_mut(), addr, last_ignored)
+    else {
+        return Ok(());
     };
 
-    // next event can be logged as ignored again
-    last_ignored.take();
-
-    log::trace!("{:20} <-<-<-<------ {addr} ({handle})", event.to_string());
-    {
-        let mut client_manager = server.client_manager.borrow_mut();
-        let client_state = match client_manager.get_mut(handle) {
-            Some((_, s)) => s,
-            None => {
-                log::error!("unknown handle");
-                return Ok(());
-            }
-        };
-
-        // reset ttl for client and
-        client_state.alive = true;
-        // set addr as new default for this client
-        client_state.active_addr = Some(addr);
-    }
-
     match (event, addr) {
         (Event::Pong(), _) => { /* ignore pong events */ }
         (Event::Ping(), addr) => {
@@ -148,30 +151,22 @@ async fn handle_udp_rx(
                     }
                 }
                 State::Receiving => {
-                    let mut ignore_event = false;
-                    if let Event::Keyboard(KeyboardEvent::Key {
-                        time: _,
-                        key,
-                        state,
-                    }) = event
-                    {
-                        let mut client_manager = server.client_manager.borrow_mut();
-                        let client_state = if let Some((_, s)) = client_manager.get_mut(handle) {
-                            s
+                    let ignore_event =
+                        if let Event::Keyboard(KeyboardEvent::Key { key, state, .. }) = event {
+                            let (ignore_event, restart_timer) = update_client_keys(
+                                &mut server.client_manager.borrow_mut(),
+                                handle,
+                                key,
+                                state,
+                            );
+                            // restart timer if necessary
+                            if restart_timer {
+                                let _ = timer_tx.try_send(());
+                            }
+                            ignore_event
                         } else {
-                            log::error!("unknown handle");
-                            return Ok(());
+                            false
                         };
-                        if state == 0 {
-                            // ignore release event if key not pressed
-                            ignore_event = !client_state.pressed_keys.remove(&key);
-                        } else {
-                            // ignore press event if key not released
-                            ignore_event = !client_state.pressed_keys.insert(key);
-                            let _ = timer_tx.try_send(());
-                        }
-                    }
-                    // ignore double press / release events to
                     // workaround buggy rdp backend.
                     if !ignore_event {
                         // consume event
@@ -203,6 +198,22 @@ async fn handle_udp_rx(
     Ok(())
 }
 
+async fn release_all_keys(
+    server: &Server,
+    emulation: &mut Box<dyn InputEmulation>,
+) -> Result<(), EmulationError> {
+    let clients = server
+        .client_manager
+        .borrow()
+        .get_client_states()
+        .map(|(h, _)| h)
+        .collect::<Vec<_>>();
+    for client in clients {
+        release_keys(server, emulation, client).await?;
+    }
+    Ok(())
+}
+
 async fn release_keys(
     server: &Server,
     emulate: &mut Box<dyn InputEmulation>,
@@ -237,3 +248,50 @@ async fn release_keys(
     emulate.consume(event, client).await?;
     Ok(())
 }
+
+fn activate_client_if_exists(
+    client_manager: &mut ClientManager,
+    addr: SocketAddr,
+    last_ignored: &mut Option<SocketAddr>,
+) -> Option<ClientHandle> {
+    let Some(handle) = client_manager.get_client(addr) else {
+        // log ignored if it is the first event from the client in a series
+        if last_ignored.is_none() || last_ignored.is_some() && last_ignored.unwrap() != addr {
+            log::warn!("ignoring events from client {addr}");
+            last_ignored.replace(addr);
+        }
+        return None;
+    };
+    // next event can be logged as ignored again
+    last_ignored.take();
+
+    let (_, client_state) = client_manager.get_mut(handle)?;
+
+    // reset ttl for client
+    client_state.alive = true;
+    // set addr as new default for this client
+    client_state.active_addr = Some(addr);
+    Some(handle)
+}
+
+fn update_client_keys(
+    client_manager: &mut ClientManager,
+    handle: ClientHandle,
+    key: u32,
+    state: u8,
+) -> (bool, bool) {
+    let Some(client_state) = client_manager.get_mut(handle).map(|(_, s)| s) else {
+        return (true, false);
+    };
+
+    // ignore double press / release events
+    let ignore_event = if state == 0 {
+        // ignore release event if key not pressed
+        !client_state.pressed_keys.remove(&key)
+    } else {
+        // ignore press event if key not released
+        !client_state.pressed_keys.insert(key)
+    };
+    let restart_timer = !client_state.pressed_keys.is_empty();
+    (ignore_event, restart_timer)
+}

src/server/network_task.rs

@@ -1,6 +1,6 @@
-use std::net::SocketAddr;
+use std::{io, net::SocketAddr};
 
-use anyhow::Result;
+use thiserror::Error;
 use tokio::{
     net::UdpSocket,
     sync::mpsc::{Receiver, Sender},
@@ -8,66 +8,37 @@ use tokio::{
 };
 
 use crate::frontend::FrontendEvent;
-use input_event::Event;
+use input_event::{Event, ProtocolError};
 
 use super::Server;
 
 pub async fn new(
     server: Server,
     frontend_notify_tx: Sender<FrontendEvent>,
-) -> Result<(
+) -> io::Result<(
     JoinHandle<()>,
     Sender<(Event, SocketAddr)>,
-    Receiver<Result<(Event, SocketAddr)>>,
+    Receiver<Result<(Event, SocketAddr), NetworkError>>,
     Sender<u16>,
 )> {
     // bind the udp socket
     let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), server.port.get());
     let mut socket = UdpSocket::bind(listen_addr).await?;
     let (receiver_tx, receiver_rx) = tokio::sync::mpsc::channel(32);
-    let (sender_tx, mut sender_rx) = tokio::sync::mpsc::channel(32);
+    let (sender_tx, sender_rx) = tokio::sync::mpsc::channel(32);
     let (port_tx, mut port_rx) = tokio::sync::mpsc::channel(32);
 
     let udp_task = tokio::task::spawn_local(async move {
+        let mut sender_rx = sender_rx;
         loop {
+            let udp_receiver = udp_receiver(&socket, &receiver_tx);
+            let udp_sender = udp_sender(&socket, &mut sender_rx);
             tokio::select! {
-                event = receive_event(&socket) => {
-                    let _ = receiver_tx.send(event).await;
-                }
-                event = sender_rx.recv() => {
-                    let Some((event, addr)) = event else {
-                        break;
-                    };
-                    if let Err(e) = send_event(&socket, event, addr) {
-                        log::warn!("udp send failed: {e}");
-                    };
-                }
-                port = port_rx.recv() => {
-                    let Some(port) = port else {
-                        break;
-                    };
-
-                    if socket.local_addr().unwrap().port() == port {
-                        continue;
-                    }
-
-                    let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), port);
-                    match UdpSocket::bind(listen_addr).await {
-                        Ok(new_socket) => {
-                            socket = new_socket;
-                            server.port.replace(port);
-                            let _ = frontend_notify_tx.send(FrontendEvent::PortChanged(port, None)).await;
-                        }
-                        Err(e) => {
-                            log::warn!("could not change port: {e}");
-                            let port = socket.local_addr().unwrap().port();
-                            let _ = frontend_notify_tx.send(FrontendEvent::PortChanged(
-                                    port,
-                                    Some(format!("could not change port: {e}")),
-                                )).await;
-                        }
-                    }
-
+                _ = udp_receiver => break, /* channel closed */
+                _ = udp_sender => break, /* channel closed */
+                port = port_rx.recv() => match port {
+                    Some(port) => update_port(&server, &frontend_notify_tx, &mut socket, port).await,
+                    _ => break,
                 }
             }
         }
@@ -75,13 +46,73 @@ pub async fn new(
     Ok((udp_task, sender_tx, receiver_rx, port_tx))
 }
 
-async fn receive_event(socket: &UdpSocket) -> Result<(Event, SocketAddr)> {
+async fn update_port(
+    server: &Server,
+    frontend_chan: &Sender<FrontendEvent>,
+    socket: &mut UdpSocket,
+    port: u16,
+) {
+    // if port is the same, we dont need to change it
+    if socket.local_addr().unwrap().port() == port {
+        return;
+    }
+
+    // create new socket
+    let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), port);
+    let frontend_event = match UdpSocket::bind(listen_addr).await {
+        Ok(new_socket) => {
+            *socket = new_socket;
+            server.port.replace(port);
+            FrontendEvent::PortChanged(port, None)
+        }
+        Err(e) => {
+            log::warn!("could not change port: {e}");
+            let port = socket.local_addr().unwrap().port();
+            FrontendEvent::PortChanged(port, Some(format!("could not change port: {e}")))
+        }
+    };
+    let _ = frontend_chan.send(frontend_event).await;
+}
+
+async fn udp_receiver(
+    socket: &UdpSocket,
+    receiver_tx: &Sender<Result<(Event, SocketAddr), NetworkError>>,
+) {
+    loop {
+        let event = receive_event(socket).await;
+        if receiver_tx.send(event).await.is_err() {
+            break;
+        }
+    }
+}
+
+async fn udp_sender(socket: &UdpSocket, rx: &mut Receiver<(Event, SocketAddr)>) {
+    loop {
+        let (event, addr) = match rx.recv().await {
+            Some(e) => e,
+            None => return,
+        };
+        if let Err(e) = send_event(socket, event, addr) {
+            log::warn!("udp send failed: {e}");
+        };
+    }
+}
+
+#[derive(Debug, Error)]
+pub(crate) enum NetworkError {
+    #[error(transparent)]
+    Protocol(#[from] ProtocolError),
+    #[error("network error: `{0}`")]
+    Io(#[from] io::Error),
+}
+
+async fn receive_event(socket: &UdpSocket) -> Result<(Event, SocketAddr), NetworkError> {
     let mut buf = vec![0u8; 22];
     let (_amt, src) = socket.recv_from(&mut buf).await?;
     Ok((Event::try_from(buf)?, src))
 }
 
-fn send_event(sock: &UdpSocket, e: Event, addr: SocketAddr) -> Result<usize> {
+fn send_event(sock: &UdpSocket, e: Event, addr: SocketAddr) -> Result<usize, NetworkError> {
     log::trace!("{:20} ------>->->-> {addr}", e.to_string());
     let data: Vec<u8> = (&e).into();
     // When udp blocks, we dont want to block the event loop.