Async (#30)

- manual eventloop now replaced by asycn-await using the tokio runtime
- dns no longer blocks the event loop
- simplifies logic
- makes xdg-desktop-portal easier to integrate

Cargo.lock

@@ -685,11 +685,10 @@ dependencies = [
  "libc",
  "log",
  "memmap",
- "mio",
- "mio-signals",
  "serde",
  "serde_json",
  "tempfile",
+ "tokio",
  "toml",
  "trust-dns-resolver",
  "wayland-client",
@@ -842,22 +841,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "927a765cd3fc26206e66b296465fa9d3e5ab003e651c1b3c060e7956d96b19d2"
 dependencies = [
  "libc",
- "log",
  "wasi",
  "windows-sys",
 ]
 
-[[package]]
-name = "mio-signals"
-version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "21e9524e26c8749824640a1282b68a695b21d55862efa6d465b5f71107c93368"
-dependencies = [
- "libc",
- "log",
- "mio",
-]
-
 [[package]]
 name = "nix"
 version = "0.26.4"
@@ -1346,9 +1333,21 @@ dependencies = [
  "num_cpus",
  "pin-project-lite",
  "socket2",
+ "tokio-macros",
  "windows-sys",
 ]
 
+[[package]]
+name = "tokio-macros"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.37",
+]
+
 [[package]]
 name = "toml"
 version = "0.7.8"

Cargo.toml

@@ -21,9 +21,9 @@ serde = { version = "1.0", features = ["derive"] }
 anyhow = "1.0.71"
 log = "0.4.20"
 env_logger = "0.10.0"
-mio = { version = "0.8", features = ["os-ext"] }
 libc = "0.2.148"
 serde_json = "1.0.107"
+tokio = {version = "1.32.0", features = ["io-util", "macros", "net", "rt", "sync" ] }
 
 [target.'cfg(unix)'.dependencies]
 wayland-client = { version="0.30.2", optional = true }
@@ -31,7 +31,6 @@ wayland-protocols = { version="0.30.0", features=["client", "staging", "unstable
 wayland-protocols-wlr = { version="0.1.0", features=["client"], optional = true }
 wayland-protocols-misc = { version="0.1.0", features=["client"], optional = true }
 wayland-protocols-plasma = { version="0.1.0", features=["client"], optional = true }
-mio-signals = "0.2.0"
 x11 = { version = "2.21.0", features = ["xlib", "xtest"], optional = true }
 gtk = { package = "gtk4", version = "0.7.2", features = ["v4_6"], optional = true }
 adw = { package = "libadwaita", version = "0.5.2", features = ["v1_1"], optional = true }

src/backend/producer/wayland.rs

@@ -1,9 +1,9 @@
 use crate::{client::{ClientHandle, Position, ClientEvent}, producer::EventProducer};
-use mio::{event::Source, unix::SourceFd};
 
-use std::{os::fd::RawFd, vec::Drain, io::ErrorKind, env};
+use std::{os::fd::RawFd, vec::Drain, io::{ErrorKind, self}, env};
 use memmap::MmapOptions;
 use anyhow::{anyhow, Result};
+use tokio::io::unix::AsyncFd;
 
 use std::{
     fs::File,
@@ -421,29 +421,12 @@ impl State {
     }
 }
 
-impl Source for WaylandEventProducer {
+impl AsRawFd for WaylandEventProducer {
-    fn register(
+    fn as_raw_fd(&self) -> RawFd {
-        &mut self,
+        self.state.wayland_fd
-        registry: &mio::Registry,
-        token: mio::Token,
-        interests: mio::Interest,
-    ) -> std::io::Result<()> {
-        SourceFd(&self.state.wayland_fd).register(registry, token, interests)
-    }
-
-    fn reregister(
-        &mut self,
-        registry: &mio::Registry,
-        token: mio::Token,
-        interests: mio::Interest,
-    ) -> std::io::Result<()> {
-        SourceFd(&self.state.wayland_fd).reregister(registry, token, interests)
-    }
-
-    fn deregister(&mut self, registry: &mio::Registry) -> std::io::Result<()> {
-        SourceFd(&self.state.wayland_fd).deregister(registry)
     }
 }
+
 impl WaylandEventProducer {
     fn read(&mut self) -> bool {
         match self.state.read_guard.take().unwrap().read() {
@@ -508,6 +491,10 @@ impl WaylandEventProducer {
 
 impl EventProducer for WaylandEventProducer {
 
+    fn get_async_fd(&self) -> io::Result<AsyncFd<RawFd>> {
+        AsyncFd::new(self.as_raw_fd())
+    }
+
     fn read_events(&mut self) -> Drain<(ClientHandle, Event)> {
         // read events
         while self.read() {

src/backend/producer/windows.rs

@@ -1,8 +1,4 @@
-use std::vec::Drain;
+use tokio::sync::mpsc::{self, Receiver, Sender};
-
-use mio::{Token, Registry};
-use mio::event::Source;
-use std::io::Result;
 
 use crate::{
     client::{ClientHandle, ClientEvent},
@@ -11,48 +7,24 @@ use crate::{
 };
 
 pub struct WindowsProducer {
-    pending_events: Vec<(ClientHandle, Event)>,
+    _tx: Sender<(ClientHandle, Event)>,
+    rx: Option<Receiver<(ClientHandle, Event)>>,
 }
 
-impl Source for WindowsProducer {
-    fn register(
-        &mut self,
-        _registry: &Registry,
-        _token: Token,
-        _interests: mio::Interest,
-    ) -> Result<()> {
-        Ok(())
-    }
-
-    fn reregister(
-        &mut self,
-        _registry: &Registry,
-        _token: Token,
-        _interests: mio::Interest,
-    ) -> Result<()> {
-        Ok(())
-    }
-
-    fn deregister(&mut self, _registry: &Registry) -> Result<()> {
-        Ok(())
-    }
-}
-
-
 impl EventProducer for WindowsProducer {
     fn notify(&mut self, _: ClientEvent) { }
 
-    fn read_events(&mut self) -> Drain<(ClientHandle, Event)> {
-        self.pending_events.drain(..)
-    }
-
     fn release(&mut self) { }
+    
+    fn get_wait_channel(&mut self) -> Option<mpsc::Receiver<(ClientHandle, Event)>> {
+        self.rx.take()
+    }
 }
 
 impl WindowsProducer {
     pub(crate) fn new() -> Self {
-        Self {
+        let (_tx, rx) = mpsc::channel(1);
-            pending_events: vec![],
+        let rx = Some(rx);
-        }
+        Self { _tx, rx }
     }
 }

src/backend/producer/x11.rs

@@ -1,12 +1,13 @@
+use std::io::Result;
+use std::os::fd::{AsRawFd, self};
 use std::vec::Drain;
 
-use mio::{Token, Registry};
+use tokio::io::unix::AsyncFd;
-use mio::event::Source;
-use std::io::Result;
 
+use crate::event::Event;
 use crate::producer::EventProducer;
 
-use crate::{client::{ClientHandle, ClientEvent}, event::Event};
+use crate::client::{ClientEvent, ClientHandle};
 
 pub struct X11Producer {
     pending_events: Vec<(ClientHandle, Event)>,
@@ -20,27 +21,9 @@ impl X11Producer {
     }
 }
 
-impl Source for X11Producer {
+impl AsRawFd for X11Producer {
-    fn register(
+    fn as_raw_fd(&self) -> fd::RawFd {
-        &mut self,
+        todo!()
-        _registry: &Registry,
-        _token: Token,
-        _interests: mio::Interest,
-    ) -> Result<()> {
-        Ok(())
-    }
-
-    fn reregister(
-        &mut self,
-        _registry: &Registry,
-        _token: Token,
-        _interests: mio::Interest,
-    ) -> Result<()> {
-        Ok(())
-    }
-
-    fn deregister(&mut self, _registry: &Registry) -> Result<()> {
-        Ok(())
     }
 }
 
@@ -52,4 +35,8 @@ impl EventProducer for X11Producer {
     }
 
     fn release(&mut self) {}
+
+    fn get_async_fd(&self) -> Result<AsyncFd<fd::RawFd>> {
+        todo!()
+    }
 }

src/dns.rs

@@ -1,20 +1,23 @@
 use anyhow::Result;
 use std::{error::Error, net::IpAddr};
 
-use trust_dns_resolver::Resolver;
+use trust_dns_resolver::TokioAsyncResolver;
 
 pub(crate) struct DnsResolver {
-    resolver: Resolver,
+    resolver: TokioAsyncResolver,
 }
 impl DnsResolver {
-    pub(crate) fn new() -> Result<Self> {
+    pub(crate) async fn new() -> Result<Self> {
-        let resolver = Resolver::from_system_conf()?;
+        let resolver = TokioAsyncResolver::tokio_from_system_conf()?;
         Ok(Self { resolver })
     }
 
-    pub(crate) fn resolve(&self, host: &str) -> Result<Vec<IpAddr>, Box<dyn Error>> {
+    pub(crate) async fn resolve(&self, host: &str) -> Result<Vec<IpAddr>, Box<dyn Error>> {
         log::info!("resolving {host} ...");
-        let response = self.resolver.lookup_ip(host)?;
+        let response = self.resolver.lookup_ip(host).await?;
+        for ip in response.iter() {
+            log::info!("{host}: adding ip {ip}");
+        }
         Ok(response.iter().collect())
     }
 }

src/event/server.rs

@@ -1,12 +1,16 @@
 use std::{error::Error, io::Result, collections::HashSet, time::{Duration, Instant}, net::IpAddr};
 use log;
-use mio::{Events, Poll, Interest, Token, net::UdpSocket, event::Source};
+use tokio::{net::UdpSocket, io::ReadHalf, sync::mpsc::{Sender, Receiver}};
-#[cfg(not(windows))]
+
-use mio_signals::{Signals, Signal, SignalSet};
+#[cfg(unix)]
+use tokio::net::UnixStream;
+
+#[cfg(windows)]
+use tokio::net::TcpStream;
 
 use std::{net::SocketAddr, io::ErrorKind};
 
-use crate::{client::{ClientEvent, ClientManager, Position, ClientHandle}, consumer::EventConsumer, producer::EventProducer, frontend::{FrontendEvent, FrontendListener, FrontendNotify}, dns::{self, DnsResolver}};
+use crate::{client::{ClientEvent, ClientManager, Position, ClientHandle}, consumer::EventConsumer, producer::EventProducer, frontend::{FrontendEvent, FrontendListener, FrontendNotify, self}, dns::{self, DnsResolver}};
 use super::Event;
 
 /// keeps track of state to prevent a feedback loop
@@ -18,92 +22,134 @@ enum State {
 }
 
 pub struct Server {
-    poll: Poll,
-    socket: UdpSocket,
-    producer: Box<dyn EventProducer>,
-    consumer: Box<dyn EventConsumer>,
     resolver: DnsResolver,
-    #[cfg(not(windows))]
-    signals: Signals,
-    frontend: FrontendListener,
     client_manager: ClientManager,
     state: State,
-    next_token: usize,
+    frontend: FrontendListener,
+    consumer: Box<dyn EventConsumer>,
+    producer: Box<dyn EventProducer>,
+    socket: UdpSocket,
+    frontend_rx: Receiver<FrontendEvent>,
+    frontend_tx: Sender<FrontendEvent>,
 }
 
-const UDP_RX: Token = Token(0);
-const FRONTEND_RX: Token = Token(1);
-const PRODUCER_RX: Token = Token(2);
-#[cfg(not(windows))]
-const SIGNAL: Token = Token(3);
-
-const MAX_TOKEN: usize = 4;
-
 impl Server {
-    pub fn new(
+    pub async fn new(
         port: u16,
-        mut producer: Box<dyn EventProducer>,
+        frontend: FrontendListener,
         consumer: Box<dyn EventConsumer>,
-        mut frontend: FrontendListener,
+        producer: Box<dyn EventProducer>,
     ) -> anyhow::Result<Self> {
-        // bind the udp socket
-        let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), port);
-        let mut socket = UdpSocket::bind(listen_addr)?;
 
         // create dns resolver
-        let resolver = dns::DnsResolver::new()?;
+        let resolver = dns::DnsResolver::new().await?;
 
-        // register event sources
+        // bind the udp socket
-        let poll = Poll::new()?;
+        let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), port);
-
+        let socket = UdpSocket::bind(listen_addr).await?;
-        // hand signal handling over to the event loop
+        let (frontend_tx, frontend_rx) = tokio::sync::mpsc::channel(1);
-        #[cfg(not(windows))]
-        let mut signals = Signals::new(SignalSet::all())?;
-
-        #[cfg(not(windows))]
-        poll.registry().register(&mut signals, SIGNAL, Interest::READABLE)?;
-        poll.registry().register(&mut socket, UDP_RX, Interest::READABLE)?;
-        poll.registry().register(&mut producer, PRODUCER_RX, Interest::READABLE)?;
-        poll.registry().register(&mut frontend, FRONTEND_RX, Interest::READABLE)?;
 
         // create client manager
         let client_manager = ClientManager::new();
         Ok(Server {
-            poll, socket, consumer, producer,
+            frontend,
+            consumer,
+            producer,
             resolver,
-            #[cfg(not(windows))]
+            socket,
-            signals, frontend,
             client_manager,
             state: State::Receiving,
-            next_token: MAX_TOKEN,
+            frontend_rx,
+            frontend_tx,
         })
     }
 
-    pub fn run(&mut self) -> Result<()> {
+    pub async fn run(&mut self) -> Result<()> {
-        let mut events = Events::with_capacity(10);
+
+        #[cfg(unix)]
+        let producer_fd = self.producer.get_async_fd()?;
+
+        #[cfg(unix)]
         loop {
-            match self.poll.poll(&mut events, None) {
+            tokio::select! {
-                Ok(()) => (),
+                udp_event = receive_event(&self.socket) => {
-                Err(e) if e.kind() == ErrorKind::Interrupted => continue,
+                    match udp_event {
-                Err(e) => return Err(e),
+                        Ok(e) => self.handle_udp_rx(e).await,
+                        Err(e) => log::error!("error reading event: {e}"),
+                    }
+                }
+                read_guard = producer_fd.readable() => {
+                    let mut guard = match read_guard {
+                        Ok(g) => g,
+                        Err(e) => {
+                            log::error!("wayland_fd read_guard: {e}");
+                            continue
+                        }
+                    };
+                    self.handle_producer_rx().await;
+                    guard.clear_ready_matching(tokio::io::Ready::READABLE);
+                }
+                stream = self.frontend.accept() => {
+                    match stream {
+                        Ok(s) => self.handle_frontend_stream(s).await,
+                        Err(e) => log::error!("error connecting to frontend: {e}"),
+                    }
+                }
+                frontend_event = self.frontend_rx.recv() => {
+                    if let Some(event) = frontend_event {
+                        if self.handle_frontend_event(event).await {
+                            break;
                         }
-            for event in &events {
-                if !event.is_readable() { continue }
-                match event.token() {
-                    UDP_RX => self.handle_udp_rx(),
-                    PRODUCER_RX => self.handle_producer_rx(),
-                    FRONTEND_RX => self.handle_frontend_incoming(),
-                    #[cfg(not(windows))]
-                    SIGNAL => if self.handle_signal() { return Ok(()) },
-                    _ => if self.handle_frontend_event(event.token()) { return Ok(()) },
                     }
                 }
             }
         }
 
-    pub fn add_client(&mut self, hostname: Option<String>, mut addr: HashSet<IpAddr>, port: u16, pos: Position) -> ClientHandle {
+        #[cfg(windows)]
+        let mut channel = self.producer.get_wait_channel().unwrap();
+
+        #[cfg(windows)]
+        loop {
+            tokio::select! {
+                udp_event = receive_event(&self.socket) => {
+                    match udp_event {
+                        Ok(e) => self.handle_udp_rx(e).await,
+                        Err(e) => log::error!("error reading event: {e}"),
+                    }
+                }
+                event = channel.recv() => {
+                    if let Some((c,e)) = event {
+                        self.handle_producer_event(c,e).await;
+                    }
+                }
+                stream = self.frontend.accept() => {
+                    match stream {
+                        Ok(s) => self.handle_frontend_stream(s).await,
+                        Err(e) => log::error!("error connecting to frontend: {e}"),
+                    }
+                }
+                frontend_event = self.frontend_rx.recv() => {
+                    if let Some(event) = frontend_event {
+                        if self.handle_frontend_event(event).await {
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    pub async fn add_client(&mut self, hostname: Option<String>, mut addr: HashSet<IpAddr>, port: u16, pos: Position) -> ClientHandle {
         let ips = if let Some(hostname) = hostname.as_ref() {
-            HashSet::from_iter(self.resolver.resolve(hostname.as_str()).ok().iter().flatten().cloned())
+            match self.resolver.resolve(hostname.as_str()).await {
+                Ok(ips) => HashSet::from_iter(ips.iter().cloned()),
+                Err(e) => {
+                    log::warn!("could not resolve host: {e}");
+                    HashSet::new()
+                }
+            }
         } else {
             HashSet::new()
         };
@@ -112,7 +158,7 @@ impl Server {
         let client = self.client_manager.add_client(hostname.clone(), addr, port, pos);
         log::debug!("add_client {client}");
         let notify = FrontendNotify::NotifyClientCreate(client, hostname, port, pos);
-        if let Err(e) = self.frontend.notify_all(notify) {
+        if let Err(e) = self.frontend.notify_all(notify).await {
             log::error!("{e}");
         };
         client
@@ -131,13 +177,13 @@ impl Server {
         }
     }
 
-    pub fn remove_client(&mut self, client: ClientHandle) -> Option<ClientHandle> {
+    pub async fn remove_client(&mut self, client: ClientHandle) -> Option<ClientHandle> {
         self.producer.notify(ClientEvent::Destroy(client));
         self.consumer.notify(ClientEvent::Destroy(client));
         if let Some(client) = self.client_manager.remove_client(client).map(|s| s.client.handle) {
             let notify = FrontendNotify::NotifyClientDelete(client);
             log::debug!("{notify:?}");
-            if let Err(e) = self.frontend.notify_all(notify) {
+            if let Err(e) = self.frontend.notify_all(notify).await {
                 log::error!("{e}");
             }
             Some(client)
@@ -146,7 +192,7 @@ impl Server {
         }
     }
 
-    pub fn update_client(
+    pub async fn update_client(
         &mut self,
         client: ClientHandle,
         hostname: Option<String>,
@@ -184,46 +230,38 @@ impl Server {
             state.client.active_addr = None;
             state.client.hostname = hostname;
             if let Some(hostname) = state.client.hostname.as_ref() {
-                if let Ok(ips) = self.resolver.resolve(hostname.as_str()) {
+                match self.resolver.resolve(hostname.as_str()).await {
+                    Ok(ips) => {
                         let addrs = ips.iter().map(|i| SocketAddr::new(*i, port));
                         state.client.addrs = HashSet::from_iter(addrs);
                     }
+                    Err(e) => {
+                        log::warn!("could not resolve host: {e}");
+                    }
+                }
             }
         }
         log::debug!("client updated: {:?}", state);
     }
 
-    fn handle_udp_rx(&mut self) {
+    async fn handle_udp_rx(&mut self, event: (Event, SocketAddr)) {
-        loop {
+        let (event, addr) = event;
-            let (event, addr) = match self.receive_event() {
-                Ok(e) => e,
-                Err(e) => {
-                    if e.is::<std::io::Error>() {
-                        if let ErrorKind::WouldBlock = e.downcast_ref::<std::io::Error>()
-                            .unwrap()
-                            .kind() {
-                            return
-                        }
-                    }
-                    log::error!("{}", e);
-                    continue
-                }
-            };
 
         // get handle for addr
         let handle = match self.client_manager.get_client(addr) {
             Some(a) => a,
             None => {
                 log::warn!("ignoring event from client {addr:?}");
-                    continue
+                return;
             }
         };
+
         log::trace!("{:20} <-<-<-<------ {addr} ({handle})", event.to_string());
         let state = match self.client_manager.get_mut(handle) {
             Some(s) => s,
             None => {
                 log::error!("unknown handle");
-                    continue
+                return;
             }
         };
 
@@ -234,7 +272,7 @@ impl Server {
         match (event, addr) {
             (Event::Pong(), _) => {},
             (Event::Ping(), addr) => {
-                    if let Err(e) = Self::send_event(&self.socket, Event::Pong(), addr) {
+                if let Err(e) = send_event(&self.socket, Event::Pong(), addr).await {
                     log::error!("udp send: {}", e);
                 }
                 // we release the mouse here,
@@ -264,7 +302,7 @@ impl Server {
                     || last_replied.is_some()
                     && last_replied.unwrap().elapsed() > Duration::from_secs(1) {
                         state.last_replied = Some(Instant::now());
-                            if let Err(e) = Self::send_event(&self.socket, Event::Pong(), addr) {
+                        if let Err(e) = send_event(&self.socket, Event::Pong(), addr).await {
                             log::error!("udp send: {}", e);
                         }
                     }
@@ -272,12 +310,17 @@ impl Server {
             }
         }
     }
+
+    #[cfg(unix)]
+    async fn handle_producer_rx(&mut self) {
+        let events: Vec<(ClientHandle, Event)> = self.producer.read_events().collect();
+        for (c,e) in events.into_iter() {
+            self.handle_producer_event(c,e).await;
+        }
     }
 
-    fn handle_producer_rx(&mut self) {
+    async fn handle_producer_event(&mut self, c: ClientHandle, e: Event) {
-        let events = self.producer.read_events();
         let mut should_release = false;
-        for (c, e) in events.into_iter() {
         // in receiving state, only release events
         // must be transmitted
         if let Event::Release() = e {
@@ -289,13 +332,13 @@ impl Server {
             Some(state) => state,
             None => {
                 log::warn!("unknown client!");
-                    continue
+                return
             }
         };
         // otherwise we should have an address to send to
         // transmit events to the corrensponding client
         if let Some(addr) = state.client.active_addr {
-                if let Err(e) = Self::send_event(&self.socket, e, addr) {
+            if let Err(e) = send_event(&self.socket, e, addr).await {
                 log::error!("udp send: {}", e);
             }
         }
@@ -305,13 +348,13 @@ impl Server {
         // send a ping
         if state.last_seen.is_some()
         && state.last_seen.unwrap().elapsed() < Duration::from_secs(2) {
-                continue
+            return
         }
 
         // client last seen > 500ms ago
         if state.last_ping.is_some()
         && state.last_ping.unwrap().elapsed() < Duration::from_millis(500) {
-                continue
+            return
         }
 
         // release mouse if client didnt respond to the first ping
@@ -324,142 +367,94 @@ impl Server {
         state.last_ping = Some(Instant::now());
         for addr in state.client.addrs.iter() {
             log::debug!("pinging {addr}");
-                if let Err(e) = Self::send_event(&self.socket, Event::Ping(), *addr) {
+            if let Err(e) = send_event(&self.socket, Event::Ping(), *addr).await {
                 if e.kind() != ErrorKind::WouldBlock {
                     log::error!("udp send: {}", e);
                 }
             }
             // send additional release event, in case client is still in sending mode
-                if let Err(e) = Self::send_event(&self.socket, Event::Release(), *addr) {
+            if let Err(e) = send_event(&self.socket, Event::Release(), *addr).await {
                 if e.kind() != ErrorKind::WouldBlock {
                     log::error!("udp send: {}", e);
                 }
             }
         }
-        }
 
         if should_release && self.state != State::Receiving {
             log::info!("client not responding - releasing pointer");
             self.producer.release();
             self.state = State::Receiving;
         }
-
     }
 
-    fn handle_frontend_incoming(&mut self) {
+    #[cfg(unix)]
+    async fn handle_frontend_stream(&mut self, mut stream: ReadHalf<UnixStream>) {
+        let tx = self.frontend_tx.clone();
+        tokio::task::spawn_local(async move {
             loop {
-            let token = self.fresh_token();
+                let event = frontend::read_event(&mut stream).await;
-            let poll = &mut self.poll;
+                match event {
-            match self.frontend.handle_incoming(|s, i| {
+                    Ok(event) => tx.send(event).await.unwrap(),
-                poll.registry().register(s, token, i)?;
+                    Err(e) => log::error!("error reading frontend event: {e}"),
-                Ok(token)
-            }) {
-                Err(e) if e.kind() == ErrorKind::WouldBlock => break,
-                Err(e) if e.kind() == ErrorKind::Interrupted => continue,
-                Err(e) =>  {
-                    log::error!("{e}");
-                    break
-                }
-                _ => continue,
                 }
             }
-        // notify new frontend connections of current clients
+        });
-        self.enumerate();
+        self.enumerate().await;
     }
 
-    fn handle_frontend_event(&mut self, token: Token) -> bool {
+    #[cfg(windows)]
+    async fn handle_frontend_stream(&mut self, mut stream: ReadHalf<TcpStream>) {
+        let tx = self.frontend_tx.clone();
+        tokio::task::spawn_local(async move {
             loop {
-            let event = match self.frontend.read_event(token) {
+                let event = frontend::read_event(&mut stream).await;
-                Ok(event) => event,
+                match event {
-                Err(e) if e.kind() == ErrorKind::WouldBlock => return false,
+                    Ok(event) => tx.send(event).await.unwrap(),
-                Err(e) if e.kind() == ErrorKind::Interrupted => continue,
+                    Err(e) => log::error!("error reading frontend event: {e}"),
-                Err(e) => {
-                    log::error!("{e}");
-                    return false;
                 }
-            };
+            }
-            if let Some(event) = event {
+        });
+        self.enumerate().await;
+    }
+
+    async fn handle_frontend_event(&mut self, event: FrontendEvent) -> bool {
         log::debug!("frontend: {event:?}");
         match event {
-                    FrontendEvent::AddClient(hostname, port, pos) => {
+            FrontendEvent::AddClient(hostname, port, pos) => { self.add_client(hostname, HashSet::new(), port, pos).await; },
-                        self.add_client(hostname, HashSet::new(), port, pos);
+            FrontendEvent::ActivateClient(client, active) => self.activate_client(client, active),
-                    }
+            FrontendEvent::DelClient(client) => { self.remove_client(client).await; },
-                    FrontendEvent::ActivateClient(client, active) => {
+            FrontendEvent::UpdateClient(client, hostname, port, pos) => self.update_client(client, hostname, port, pos).await,
-                        self.activate_client(client, active);
+            FrontendEvent::Enumerate() => self.enumerate().await,
-                    }
-                    FrontendEvent::DelClient(client) => {
-                        self.remove_client(client);
-                    }
-                    FrontendEvent::UpdateClient(client, hostname, port, pos) => {
-                        self.update_client(client, hostname, port, pos);
-                    }
-                    FrontendEvent::Enumerate() => self.enumerate(),
             FrontendEvent::Shutdown() => {
                 log::info!("terminating gracefully...");
                 return true;
             },
         }
-            }
+        false
-        }
     }
 
-    fn enumerate(&mut self) {
+    async fn enumerate(&mut self) {
         let clients = self.client_manager.enumerate();
-        if let Err(e) = self.frontend.notify_all(FrontendNotify::Enumerate(clients)) {
+        if let Err(e) = self.frontend.notify_all(FrontendNotify::Enumerate(clients)).await {
             log::error!("{e}");
         }
     }
+}
 
-    #[cfg(not(windows))]
+async fn receive_event(socket: &UdpSocket) -> std::result::Result<(Event, SocketAddr), Box<dyn Error>> {
-    fn handle_signal(&mut self) -> bool {
+    let mut buf = vec![0u8; 22];
-        #[cfg(windows)]
+    match socket.recv_from(&mut buf).await {
-        return false;
+        Ok((_amt, src)) => Ok((Event::try_from(buf)?, src)),
-        #[cfg(not(windows))]
+        Err(e) => Err(Box::new(e)),
-        loop {
-            match self.signals.receive() {
-                Err(e) if e.kind() == ErrorKind::WouldBlock => return false,
-                Err(e) => {
-                    log::error!("error reading signal: {e}");
-                    return false;
-                }
-                Ok(Some(Signal::Interrupt) | Some(Signal::Terminate)) => {
-                    // terminate on SIG_INT or SIG_TERM
-                    log::info!("terminating gracefully...");
-                    return true;
-                },
-                Ok(Some(signal)) => {
-                    log::info!("ignoring signal {signal:?}");
-                },
-                Ok(None) => return false,
-            }
-        }
     }
+}
 
-    fn send_event(sock: &UdpSocket, e: Event, addr: SocketAddr) -> Result<usize> {
+
+async fn send_event(sock: &UdpSocket, e: Event, addr: SocketAddr) -> Result<usize> {
     log::trace!("{:20} ------>->->-> {addr}", e.to_string());
     let data: Vec<u8> = (&e).into();
     // We are currently abusing a blocking send to get the lowest possible latency.
     // It may be better to set the socket to non-blocking and only send when ready.
-        sock.send_to(&data[..], addr)
+    sock.send_to(&data[..], addr).await
-    }
-
-    fn receive_event(&self) -> std::result::Result<(Event, SocketAddr), Box<dyn Error>> {
-        let mut buf = vec![0u8; 22];
-        match self.socket.recv_from(&mut buf) {
-            Ok((_amt, src)) => Ok((Event::try_from(buf)?, src)),
-            Err(e) => Err(Box::new(e)),
-        }
-    }
-
-    fn fresh_token(&mut self) ->  Token {
-        let token = self.next_token as usize;
-        self.next_token += 1;
-        Token(token)
-    }
-
-    pub fn register_frontend(&mut self, source: &mut dyn Source, interests: Interest) -> Result<Token> {
-        let token = self.fresh_token();
-        self.poll.registry().register(source, token, interests)?;
-        Ok(token)
-    }
 }
+

src/frontend.rs

@@ -1,21 +1,20 @@
-use std::collections::HashMap;
+use std::io::Result;
-use std::io::{Read, Result, Write};
 use std::str;
 
 #[cfg(unix)]
 use std::{env, path::{Path, PathBuf}};
 
-use mio::Interest;
+use tokio::io::{AsyncReadExt, WriteHalf, AsyncWriteExt};
-use mio::{Registry, Token, event::Source};
+use tokio::io::ReadHalf;
 
 #[cfg(unix)]
-use mio::net::UnixStream;
+use tokio::net::UnixStream;
 #[cfg(unix)]
-use mio::net::UnixListener;
+use tokio::net::UnixListener;
 #[cfg(windows)]
-use mio::net::TcpStream;
+use tokio::net::TcpStream;
 #[cfg(windows)]
-use mio::net::TcpListener;
+use tokio::net::TcpListener;
 
 use serde::{Serialize, Deserialize};
 
@@ -60,11 +59,14 @@ pub struct FrontendListener {
     listener: UnixListener,
     #[cfg(unix)]
     socket_path: PathBuf,
-    frontend_connections: HashMap<Token, FrontendConnection>,
+    #[cfg(unix)]
+    tx_streams: Vec<WriteHalf<UnixStream>>,
+    #[cfg(windows)]
+    tx_streams: Vec<WriteHalf<TcpStream>>,
 }
 
 impl FrontendListener {
-    pub fn new() -> std::result::Result<Self, Box<dyn std::error::Error>> {
+    pub async fn new() -> std::result::Result<Self, Box<dyn std::error::Error>> {
         #[cfg(unix)]
         let socket_path = Path::new(env::var("XDG_RUNTIME_DIR")?.as_str()).join("lan-mouse-socket.sock");
         #[cfg(unix)]
@@ -77,86 +79,53 @@ impl FrontendListener {
         let listener = UnixListener::bind(&socket_path)?;
 
         #[cfg(windows)]
-        let listener = TcpListener::bind("127.0.0.1:5252".parse().unwrap())?; // abuse tcp
+        let listener = TcpListener::bind("127.0.0.1:5252").await?; // abuse tcp
 
         let adapter = Self {
             listener,
             #[cfg(unix)]
             socket_path,
-            frontend_connections: HashMap::new(),
+            tx_streams: vec![],
         };
 
         Ok(adapter)
     }
 
     #[cfg(unix)]
-    pub fn handle_incoming<F>(&mut self, register_frontend: F) -> Result<()>
+    pub async fn accept(&mut self) -> Result<ReadHalf<UnixStream>> {
-    where F: Fn(&mut UnixStream, Interest) -> Result<Token> {
+
-        let (mut stream, _) = self.listener.accept()?;
+        let stream = self.listener.accept().await?.0;
-        let token = register_frontend(&mut stream, Interest::READABLE)?;
+        let (rx, tx) = tokio::io::split(stream);
-        let con = FrontendConnection::new(stream);
+        self.tx_streams.push(tx);
-        self.frontend_connections.insert(token, con);
+        Ok(rx)
-        Ok(())
     }
 
     #[cfg(windows)]
-    pub fn handle_incoming<F>(&mut self, register_frontend: F) -> Result<()>
+    pub async fn accept(&mut self) -> Result<ReadHalf<TcpStream>> {
-    where F: Fn(&mut TcpStream, Interest) -> Result<Token> {
+        let stream = self.listener.accept().await?.0;
-        let (mut stream, _) = self.listener.accept()?;
+        let (rx, tx) = tokio::io::split(stream);
-        let token = register_frontend(&mut stream, Interest::READABLE)?;
+        self.tx_streams.push(tx);
-        let con = FrontendConnection::new(stream);
+        Ok(rx)
-        self.frontend_connections.insert(token, con);
-        Ok(())
     }
 
-    pub fn read_event(&mut self, token: Token) -> Result<Option<FrontendEvent>> {
-        if let Some(con) = self.frontend_connections.get_mut(&token) {
-            con.handle_event()
-        } else {
-            panic!("unknown token");
-        }
-    }
 
-    pub(crate) fn notify_all(&mut self, notify: FrontendNotify) -> Result<()> {
+    pub(crate) async fn notify_all(&mut self, notify: FrontendNotify) -> Result<()> {
         // encode event
         let json = serde_json::to_string(&notify).unwrap();
         let payload = json.as_bytes();
-        let len = payload.len().to_ne_bytes();
+        let len = payload.len().to_be_bytes();
         log::debug!("json: {json}, len: {}", payload.len());
 
-        for con in self.frontend_connections.values_mut() {
+        // TODO do simultaneously
+        for tx in self.tx_streams.iter_mut() {
             // write len + payload
-            con.stream.write(&len)?;
+            tx.write(&len).await?;
-            con.stream.write(payload)?;
+            tx.write(payload).await?;
         }
         Ok(())
     }
 }
 
-impl Source for FrontendListener {
-    fn register(
-        &mut self,
-        registry: &Registry,
-        token: Token,
-        interests: mio::Interest,
-    ) -> Result<()> {
-        self.listener.register(registry, token, interests)
-    }
-
-    fn reregister(
-        &mut self,
-        registry: &Registry,
-        token: Token,
-        interests: mio::Interest,
-    ) -> Result<()> {
-        self.listener.reregister(registry, token, interests)
-    }
-
-    fn deregister(&mut self, registry: &Registry) -> Result<()> {
-        self.listener.deregister(registry)
-    }
-}
-
 #[cfg(unix)]
 impl Drop for FrontendListener {
     fn drop(&mut self) {
@@ -165,72 +134,20 @@ impl Drop for FrontendListener {
     }
 }
 
-enum ReceiveState {
+#[cfg(unix)]
-    Len, Data,
+pub async fn read_event(stream: &mut ReadHalf<UnixStream>) -> Result<FrontendEvent> {
+    let len = stream.read_u64().await?;
+    assert!(len <= 256);
+    let mut buf = [0u8; 256];
+    stream.read_exact(&mut buf[..len as usize]).await?;
+    Ok(serde_json::from_slice(&buf[..len as usize])?)
 }
 
-pub struct FrontendConnection {
+#[cfg(windows)]
-    #[cfg(unix)]
+pub async fn read_event(stream: &mut ReadHalf<TcpStream>) -> Result<FrontendEvent> {
-    stream: UnixStream,
+    let len = stream.read_u64().await?;
-    #[cfg(windows)]
+    let mut buf = [0u8; 256];
-    stream: TcpStream,
+    stream.read_exact(&mut buf[..len as usize]).await?;
-    state: ReceiveState,
+    Ok(serde_json::from_slice(&buf[..len as usize])?)
-    len: usize,
-    len_buf: [u8; std::mem::size_of::<usize>()],
-    recieve_buf: [u8; 256], // FIXME
-    pos: usize,
 }
 
-impl FrontendConnection {
-    #[cfg(unix)]
-    pub fn new(stream: UnixStream) -> Self {
-        Self {
-            stream,
-            state: ReceiveState::Len,
-            len: 0,
-            len_buf: [0u8; std::mem::size_of::<usize>()],
-            recieve_buf: [0u8; 256],
-            pos: 0,
-        }
-    }
-
-    #[cfg(windows)]
-    pub fn new(stream: TcpStream) -> Self {
-        Self {
-            stream,
-            state: ReceiveState::Len,
-            len: 0,
-            len_buf: [0u8; std::mem::size_of::<usize>()],
-            recieve_buf: [0u8; 256],
-            pos: 0,
-        }
-    }
-
-    pub fn handle_event(&mut self) -> Result<Option<FrontendEvent>> {
-        match self.state {
-            ReceiveState::Len => {
-                // we receive sizeof(usize) Bytes
-                let n = self.stream.read(&mut self.len_buf)?;
-                self.pos += n;
-                if self.pos == self.len_buf.len() {
-                    self.state = ReceiveState::Data;
-                    self.len = usize::from_ne_bytes(self.len_buf);
-                    self.pos = 0;
-                }
-                Ok(None)
-            },
-            ReceiveState::Data => {
-                // read at most as many bytes as the length of the next event
-                let n = self.stream.read(&mut self.recieve_buf[..self.len])?;
-                self.pos += n;
-                if n == self.len {
-                    self.state = ReceiveState::Len;
-                    self.pos = 0;
-                    Ok(Some(serde_json::from_slice(&self.recieve_buf[..self.len])?))
-                } else {
-                    Ok(None)
-                }
-            }
-        }
-    }
-}

src/frontend/cli.rs

@@ -42,7 +42,7 @@ pub fn start() -> Result<(JoinHandle<()>, JoinHandle<()>)> {
                         for event in events.iter() {
                             let json = serde_json::to_string(&event).unwrap();
                             let bytes = json.as_bytes();
-                            let len = bytes.len().to_ne_bytes();
+                            let len = bytes.len().to_be_bytes();
                             if let Err(e) = tx.write(&len) {
                                 log::error!("error sending message: {e}");
                             };
@@ -77,7 +77,7 @@ pub fn start() -> Result<(JoinHandle<()>, JoinHandle<()>)> {
                     Err(e) if e.kind() == ErrorKind::UnexpectedEof => break,
                     Err(e) => break log::error!("{e}"),
                 };
-                let len = usize::from_ne_bytes(len);
+                let len = usize::from_be_bytes(len);
 
                 // read payload
                 let mut buf: Vec<u8> = vec![0u8; len];

src/frontend/gtk.rs

@@ -15,6 +15,7 @@ use self::client_object::ClientObject;
 use super::FrontendNotify;
 
 pub fn start() -> Result<JoinHandle<glib::ExitCode>> {
+    log::debug!("starting gtk frontend");
     thread::Builder::new()
         .name("gtk-thread".into())
         .spawn(gtk_main)
@@ -58,6 +59,7 @@ fn build_ui(app: &Application) {
             process::exit(1);
         }
     };
+    log::debug!("connecting to lan-mouse-socket ... ");
     let socket_path = Path::new(xdg_runtime_dir.as_str())
         .join("lan-mouse-socket.sock");
     let Ok(mut rx) = UnixStream::connect(&socket_path) else {
@@ -71,6 +73,7 @@ fn build_ui(app: &Application) {
             process::exit(1);
         }
     };
+    log::debug!("connected to lan-mouse-socket");
     
     let (sender, receiver) = MainContext::channel::<FrontendNotify>(Priority::default());
 
@@ -83,7 +86,7 @@ fn build_ui(app: &Application) {
                 Err(e) if e.kind() == ErrorKind::UnexpectedEof => break Ok(()),
                 Err(e) => break Err(e),
             };
-            let len = usize::from_ne_bytes(len);
+            let len = usize::from_be_bytes(len);
 
             // read payload
             let mut buf = vec![0u8; len];

src/frontend/gtk/window.rs

@@ -141,7 +141,7 @@ impl Window {
         let mut stream = self.imp().stream.borrow_mut();
         let stream = stream.as_mut().unwrap();
         let bytes = json.as_bytes();
-        let len = bytes.len().to_ne_bytes();
+        let len = bytes.len().to_be_bytes();
         if let Err(e) = stream.write(&len) {
             log::error!("error sending message: {e}");
         };

src/main.rs

@@ -9,6 +9,7 @@ use lan_mouse::{
 
 #[cfg(all(unix, feature = "gtk"))]
 use lan_mouse::frontend::gtk;
+use tokio::task::LocalSet;
 
 pub fn main() {
 
@@ -30,13 +31,17 @@ pub fn run() -> Result<(), Box<dyn Error>> {
     let producer = producer::create()?;
     let consumer = consumer::create()?;
 
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_io()
+        .enable_time()
+        .build()?;
+
+    // run async event loop
+    runtime.block_on(LocalSet::new().run_until(async {
         // create frontend communication adapter
-    let frontend_adapter = FrontendListener::new()?;
+        let frontend_adapter = FrontendListener::new().await?;
 
-    // start sending and receiving events
+        // start frontend
-    let mut event_server = Server::new(config.port, producer, consumer, frontend_adapter)?;
-
-    // any threads need to be started after event_server sets up signal handling
         match config.frontend {
             #[cfg(all(unix, feature = "gtk"))]
             Gtk => { gtk::start()?; }
@@ -45,14 +50,20 @@ pub fn run() -> Result<(), Box<dyn Error>> {
             Cli => { cli::start()?; }
         };
 
+        // start sending and receiving events
+        let mut event_server = Server::new(config.port, frontend_adapter, consumer, producer).await?;
+
         // add clients from config
-    config.get_clients().into_iter().for_each(|(c, h, port, p)| {
+        for (c,h,port,p) in config.get_clients().into_iter() {
-        event_server.add_client(h, c, port, p);
+            event_server.add_client(h, c, port, p).await;
-    });
+        }
 
         log::info!("Press Ctrl+Alt+Shift+Super to release the mouse");
         // run event loop
-    event_server.run()?;
+        event_server.run().await?;
+        Result::<_, Box<dyn Error>>::Ok(())
+    }))?;
+    log::debug!("exiting main");
 
     Ok(())
 }

src/producer.rs

@@ -1,5 +1,11 @@
-use mio::event::Source;
+use std::error::Error;
-use std::{error::Error, vec::Drain};
+
+#[cfg(unix)]
+use std::{io, os::fd::RawFd, vec::Drain};
+
+#[cfg(unix)]
+use tokio::io::unix::AsyncFd;
+
 use crate::{client::{ClientHandle, ClientEvent}, event::Event};
 use crate::backend::producer;
 
@@ -49,15 +55,24 @@ pub fn create() -> Result<Box<dyn EventProducer>, Box<dyn Error>> {
     }
 }
 
-pub trait EventProducer: Source {
+pub trait EventProducer {
+
     /// notify event producer of configuration changes
     fn notify(&mut self, event: ClientEvent);
 
+    /// release mouse
+    fn release(&mut self);
+
+    /// unix only
+    #[cfg(unix)]
+    fn get_async_fd(&self) -> io::Result<AsyncFd<RawFd>>;
+
     /// read an event
     /// this function must be invoked to retrieve an Event after
     /// the eventfd indicates a pending Event
+    #[cfg(unix)]
     fn read_events(&mut self) -> Drain<(ClientHandle, Event)>;
 
-    /// release mouse
+    #[cfg(not(unix))]
-    fn release(&mut self);
+    fn get_wait_channel(&mut self) -> Option<tokio::sync::mpsc::Receiver<(ClientHandle, Event)>>;
 }