wire frontend

src/server/capture_task.rs

@@ -1,22 +1,40 @@
-use anyhow::{anyhow, Result};
 use futures::StreamExt;
 use std::{collections::HashSet, net::SocketAddr, sync::Arc};
+use thiserror::Error;
 use tokio_util::sync::CancellationToken;
 
 use tokio::{
     process::Command,
-    sync::{mpsc::Sender, Notify},
+    sync::{
+        mpsc::{Receiver, Sender},
+        Notify,
+    },
     task::JoinHandle,
 };
 
-use input_capture::{self, error::CaptureCreationError, CaptureHandle, InputCapture, Position};
+use input_capture::{
+    self, error::CaptureCreationError, CaptureError, CaptureHandle, InputCapture, Position,
+};
 
 use input_event::{scancode, Event, KeyboardEvent};
 
-use crate::{client::ClientHandle, config::CaptureBackend, server::State};
+use crate::{
+    client::ClientHandle,
+    config::CaptureBackend,
+    frontend::{FrontendEvent, Status},
+    server::State,
+};
 
 use super::Server;
 
+#[derive(Debug, Error)]
+pub enum LanMouseCaptureError {
+    #[error("error creating input-capture: `{0}`")]
+    Create(#[from] CaptureCreationError),
+    #[error("error while capturing input: `{0}`")]
+    Capture(#[from] CaptureError),
+}
+
 #[derive(Clone, Copy, Debug)]
 pub enum CaptureEvent {
     /// capture must release the mouse
@@ -25,60 +43,125 @@ pub enum CaptureEvent {
     Create(CaptureHandle, Position),
     /// destory a capture client
     Destroy(CaptureHandle),
-    /// restart input capture
-    Restart,
 }
 
 pub fn new(
     backend: Option<CaptureBackend>,
     server: Server,
     sender_tx: Sender<(Event, SocketAddr)>,
+    frontend_tx: Sender<FrontendEvent>,
     timer_notify: Arc<Notify>,
     release_bind: Vec<scancode::Linux>,
     cancellation_token: CancellationToken,
-) -> Result<(JoinHandle<Result<()>>, Sender<CaptureEvent>), CaptureCreationError> {
-    let (tx, mut rx) = tokio::sync::mpsc::channel(32);
+    notify_capture: Arc<Notify>,
+) -> (JoinHandle<()>, Sender<CaptureEvent>) {
+    let (tx, rx) = tokio::sync::mpsc::channel(32);
     let backend = backend.map(|b| b.into());
-    let task = tokio::task::spawn_local(async move {
-        let mut capture = input_capture::create(backend).await?;
-        let mut pressed_keys = HashSet::new();
-        loop {
-            tokio::select! {
-                event = capture.next() => {
-                    match event {
-                        Some(Ok(event)) => handle_capture_event(&server, &mut capture, &sender_tx, &timer_notify, event, &mut pressed_keys, &release_bind).await?,
-                        Some(Err(e)) => return Err(anyhow!("input capture: {e:?}")),
-                        None => return Err(anyhow!("input capture terminated")),
-                    }
-                }
-                e = rx.recv() => {
-                    log::debug!("input capture notify rx: {e:?}");
-                    match e {
-                        Some(e) => match e {
-                            CaptureEvent::Release => {
-                                capture.release().await?;
-                                server.state.replace(State::Receiving);
-                            }
-                            CaptureEvent::Create(h, p) => capture.create(h, p).await?,
-                            CaptureEvent::Destroy(h) => capture.destroy(h).await?,
-                            CaptureEvent::Restart => {
-                                let clients = server.client_manager.borrow().get_client_states().map(|(h, (c,_))| (h, c.pos)).collect::<Vec<_>>();
-                                capture.terminate().await?;
-                                capture = input_capture::create(backend).await?;
-                                for (handle, pos) in clients {
-                                    capture.create(handle, pos.into()).await?;
-                                }
-                            }
-                        },
-                        None => break,
-                    }
-                }
-                _ = cancellation_token.cancelled() => break,
-            }
+    let task = tokio::task::spawn_local(capture_task(
+        backend,
+        server,
+        sender_tx,
+        rx,
+        frontend_tx,
+        timer_notify,
+        release_bind,
+        cancellation_token,
+        notify_capture,
+    ));
+    (task, tx)
+}
+
+async fn capture_task(
+    backend: Option<input_capture::Backend>,
+    server: Server,
+    sender_tx: Sender<(Event, SocketAddr)>,
+    mut notify_rx: Receiver<CaptureEvent>,
+    frontend_tx: Sender<FrontendEvent>,
+    timer_notify: Arc<Notify>,
+    release_bind: Vec<scancode::Linux>,
+    cancellation_token: CancellationToken,
+    notify_capture: Arc<Notify>,
+) {
+    loop {
+        if let Err(e) = do_capture(
+            backend,
+            &server,
+            &sender_tx,
+            &mut notify_rx,
+            &frontend_tx,
+            &timer_notify,
+            &release_bind,
+            &cancellation_token,
+        )
+        .await
+        {
+            log::warn!("input emulation exited: {e}");
         }
-        anyhow::Ok(())
-    });
-    Ok((task, tx))
+        let _ = frontend_tx
+            .send(FrontendEvent::CaptureStatus(Status::Disabled))
+            .await;
+        if cancellation_token.is_cancelled() {
+            break;
+        }
+        notify_capture.notified().await;
+    }
+}
+
+async fn do_capture(
+    backend: Option<input_capture::Backend>,
+    server: &Server,
+    sender_tx: &Sender<(Event, SocketAddr)>,
+    notify_rx: &mut Receiver<CaptureEvent>,
+    frontend_tx: &Sender<FrontendEvent>,
+    timer_notify: &Notify,
+    release_bind: &[scancode::Linux],
+    cancellation_token: &CancellationToken,
+) -> Result<(), LanMouseCaptureError> {
+    let mut capture = input_capture::create(backend).await?;
+    let _ = frontend_tx
+        .send(FrontendEvent::CaptureStatus(Status::Enabled))
+        .await;
+
+    // FIXME DUPLICATES
+    let clients = server
+        .client_manager
+        .borrow()
+        .get_client_states()
+        .map(|(h, (c, _))| (h, c.pos))
+        .collect::<Vec<_>>();
+    for (handle, pos) in clients {
+        capture.create(handle, pos.into()).await?;
+    }
+
+    let mut pressed_keys = HashSet::new();
+    loop {
+        tokio::select! {
+            event = capture.next() => {
+                match event {
+                    Some(Ok(event)) => handle_capture_event(server, &mut capture, sender_tx, timer_notify, event, &mut pressed_keys, release_bind).await?,
+                    Some(Err(e)) => return Err(e.into()),
+                    None => return Ok(()),
+                }
+            }
+            e = notify_rx.recv() => {
+                log::debug!("input capture notify rx: {e:?}");
+                match e {
+                    Some(e) => match e {
+                        CaptureEvent::Release => {
+                            capture.release().await?;
+                            server.state.replace(State::Receiving);
+                        }
+                        CaptureEvent::Create(h, p) => capture.create(h, p).await?,
+                        CaptureEvent::Destroy(h) => capture.destroy(h).await?,
+                    },
+                    None => break,
+                }
+            }
+            _ = cancellation_token.cancelled() => break,
+        }
+    }
+    capture.terminate().await?;
+    Ok(())
 }
 
 fn update_pressed_keys(pressed_keys: &mut HashSet<scancode::Linux>, key: u32, state: u8) {
@@ -99,7 +182,7 @@ async fn handle_capture_event(
     event: (CaptureHandle, Event),
     pressed_keys: &mut HashSet<scancode::Linux>,
     release_bind: &[scancode::Linux],
-) -> Result<()> {
+) -> Result<(), CaptureError> {
     let (handle, mut e) = event;
     log::trace!("({handle}) {e:?}");
 

src/server/emulation_task.rs

@@ -13,6 +13,7 @@ use tokio_util::sync::CancellationToken;
 use crate::{
     client::{ClientHandle, ClientManager},
     config::EmulationBackend,
+    frontend::{FrontendEvent, Status},
     server::State,
 };
 use input_emulation::{
@@ -32,8 +33,6 @@ pub enum EmulationEvent {
     Destroy(EmulationHandle),
     /// input emulation must release keys for client
     ReleaseKeys(ClientHandle),
-    /// restart input emulation
-    Restart,
 }
 
 pub fn new(
@@ -42,12 +41,11 @@ pub fn new(
     udp_rx: Receiver<Result<(Event, SocketAddr), NetworkError>>,
     sender_tx: Sender<(Event, SocketAddr)>,
     capture_tx: Sender<CaptureEvent>,
+    frontend_tx: Sender<FrontendEvent>,
     timer_notify: Arc<Notify>,
     cancellation_token: CancellationToken,
-) -> (
-    JoinHandle<Result<(), LanMouseEmulationError>>,
-    Sender<EmulationEvent>,
-) {
+    notify_emulation: Arc<Notify>,
+) -> (JoinHandle<()>, Sender<EmulationEvent>) {
     let (tx, rx) = tokio::sync::mpsc::channel(32);
     let emulation_task = emulation_task(
         backend,
@@ -56,8 +54,10 @@ pub fn new(
         udp_rx,
         sender_tx,
         capture_tx,
+        frontend_tx,
         timer_notify,
         cancellation_token,
+        notify_emulation,
     );
     let emulate_task = tokio::task::spawn_local(emulation_task);
     (emulate_task, tx)
@@ -78,11 +78,68 @@ async fn emulation_task(
     mut udp_rx: Receiver<Result<(Event, SocketAddr), NetworkError>>,
     sender_tx: Sender<(Event, SocketAddr)>,
     capture_tx: Sender<CaptureEvent>,
+    frontend_tx: Sender<FrontendEvent>,
     timer_notify: Arc<Notify>,
     cancellation_token: CancellationToken,
+    notify_emulation: Arc<Notify>,
+) {
+    loop {
+        match do_emulation(
+            backend,
+            &mut rx,
+            &server,
+            &mut udp_rx,
+            &sender_tx,
+            &capture_tx,
+            &frontend_tx,
+            &timer_notify,
+            &cancellation_token,
+        )
+        .await
+        {
+            Ok(()) => {}
+            Err(e) => {
+                log::warn!("input emulation exited: {e}");
+            }
+        }
+        let _ = frontend_tx
+            .send(FrontendEvent::EmulationStatus(Status::Disabled))
+            .await;
+        if cancellation_token.is_cancelled() {
+            break;
+        }
+        notify_emulation.notified().await;
+    }
+}
+
+async fn do_emulation(
+    backend: Option<EmulationBackend>,
+    rx: &mut Receiver<EmulationEvent>,
+    server: &Server,
+    udp_rx: &mut Receiver<Result<(Event, SocketAddr), NetworkError>>,
+    sender_tx: &Sender<(Event, SocketAddr)>,
+    capture_tx: &Sender<CaptureEvent>,
+    frontend_tx: &Sender<FrontendEvent>,
+    timer_notify: &Notify,
+    cancellation_token: &CancellationToken,
 ) -> Result<(), LanMouseEmulationError> {
     let backend = backend.map(|b| b.into());
     let mut emulation = input_emulation::create(backend).await?;
+    let _ = frontend_tx
+        .send(FrontendEvent::EmulationStatus(Status::Enabled))
+        .await;
+
+    // FIMXE DUPLICATES
+    // add clients
+    let clients = server
+        .client_manager
+        .borrow()
+        .get_client_states()
+        .map(|(h, _)| h)
+        .collect::<Vec<_>>();
+    for handle in clients {
+        emulation.create(handle).await;
+    }
 
     let mut last_ignored = None;
     loop {
@@ -104,13 +161,6 @@ async fn emulation_task(
                         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::Restart => {
-                            let clients = server.client_manager.borrow().get_client_states().map(|(h, _)| h).collect::<Vec<_>>();
-                            emulation = input_emulation::create(backend).await?;
-                            for handle in clients {
-                                emulation.create(handle).await;
-                            }
-                        },
                     },
                     None => break,
                 }

src/server/frontend_task.rs

@@ -2,6 +2,7 @@ use std::{
     collections::HashSet,
     io::ErrorKind,
     net::{IpAddr, SocketAddr},
+    sync::Arc,
 };
 #[cfg(unix)]
 use tokio::net::UnixStream;
@@ -12,7 +13,10 @@ use tokio::net::TcpStream;
 use anyhow::{anyhow, Result};
 use tokio::{
     io::ReadHalf,
-    sync::mpsc::{Receiver, Sender},
+    sync::{
+        mpsc::{Receiver, Sender},
+        Notify,
+    },
     task::JoinHandle,
 };
 use tokio_util::sync::CancellationToken;
@@ -30,6 +34,8 @@ pub(crate) fn new(
     mut frontend: FrontendListener,
     mut notify_rx: Receiver<FrontendEvent>,
     server: Server,
+    notify_capture: Arc<Notify>,
+    notify_emulation: Arc<Notify>,
     capture: Sender<CaptureEvent>,
     emulate: Sender<EmulationEvent>,
     resolve_ch: Sender<DnsRequest>,
@@ -54,7 +60,7 @@ pub(crate) fn new(
                 }
                 event = event_rx.recv() => {
                     let frontend_event = event.ok_or(anyhow!("frontend channel closed"))?;
-                    if handle_frontend_event(&server, &capture, &emulate, &resolve_ch, &mut frontend, &port_tx, frontend_event).await {
+                    if handle_frontend_event(&server, &notify_capture, &notify_emulation, &capture, &emulate, &resolve_ch, &mut frontend, &port_tx, frontend_event).await {
                         break;
                     }
                 }
@@ -115,6 +121,8 @@ async fn listen_frontend(
 
 async fn handle_frontend_event(
     server: &Server,
+    notify_capture: &Notify,
+    notify_emulation: &Notify,
     capture: &Sender<CaptureEvent>,
     emulate: &Sender<EmulationEvent>,
     resolve_tx: &Sender<DnsRequest>,
@@ -125,10 +133,10 @@ async fn handle_frontend_event(
     log::debug!("frontend: {event:?}");
     match event {
         FrontendRequest::EnableCapture => {
-            let _ = capture.send(CaptureEvent::Restart).await;
+            notify_capture.notify_waiters();
         }
         FrontendRequest::EnableEmulation => {
-            let _ = emulate.send(EmulationEvent::Restart).await;
+            notify_emulation.notify_waiters();
         }
         FrontendRequest::Create => {
             let handle = add_client(server, frontend).await;