log pressed keys

Co-authored-by: Jacob Barber <jacob.barber@disney.com>

closes #83

Cargo.lock

@@ -1196,10 +1196,15 @@ version = "0.2.0"
 dependencies = [
  "ashpd",
  "async-trait",
+ "bitflags 2.6.0",
+ "core-foundation",
+ "core-foundation-sys",
  "core-graphics",
  "futures",
  "futures-core",
  "input-event",
+ "keycode",
+ "libc",
  "log",
  "memmap",
  "once_cell",
@@ -1308,22 +1313,19 @@ dependencies = [
 name = "lan-mouse"
 version = "0.9.1"
 dependencies = [
- "anyhow",
- "async-channel",
  "clap",
- "endi",
  "env_logger",
  "futures",
- "glib-build-tools",
- "gtk4",
  "hickory-resolver",
- "hostname 0.4.0",
  "input-capture",
  "input-emulation",
  "input-event",
+ "lan-mouse-cli",
+ "lan-mouse-gtk",
+ "lan-mouse-ipc",
  "lan-mouse-proto",
- "libadwaita",
  "libc",
+ "local-channel",
  "log",
  "serde",
  "serde_json",
@@ -1334,6 +1336,41 @@ dependencies = [
  "toml",
 ]
 
+[[package]]
+name = "lan-mouse-cli"
+version = "0.1.0"
+dependencies = [
+ "futures",
+ "lan-mouse-ipc",
+ "tokio",
+]
+
+[[package]]
+name = "lan-mouse-gtk"
+version = "0.1.0"
+dependencies = [
+ "async-channel",
+ "glib-build-tools",
+ "gtk4",
+ "hostname 0.4.0",
+ "lan-mouse-ipc",
+ "libadwaita",
+ "log",
+]
+
+[[package]]
+name = "lan-mouse-ipc"
+version = "0.1.0"
+dependencies = [
+ "futures",
+ "log",
+ "serde",
+ "serde_json",
+ "thiserror",
+ "tokio",
+ "tokio-stream",
+]
+
 [[package]]
 name = "lan-mouse-proto"
 version = "0.1.0"
@@ -1403,6 +1440,23 @@ version = "0.4.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "78b3ae25bc7c8c38cec158d1f2757ee79e9b3740fbc7ccf0e59e4b08d793fa89"
 
+[[package]]
+name = "local-channel"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6cbc85e69b8df4b8bb8b89ec634e7189099cea8927a276b7384ce5488e53ec8"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "local-waker",
+]
+
+[[package]]
+name = "local-waker"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4d873d7c67ce09b42110d801813efbc9364414e356be9935700d368351657487"
+
 [[package]]
 name = "lock_api"
 version = "0.4.12"
@@ -2067,6 +2121,17 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "tokio-stream"
+version = "0.1.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "267ac89e0bec6e691e5813911606935d77c476ff49024f98abcea3e7b15e37af"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.11"

Cargo.toml

@@ -1,5 +1,13 @@
 [workspace]
-members = ["input-capture", "input-emulation", "input-event", "lan-mouse-proto"]
+members = [
+    "input-capture",
+    "input-emulation",
+    "input-event",
+    "lan-mouse-ipc",
+    "lan-mouse-cli",
+    "lan-mouse-gtk",
+    "lan-mouse-proto",
+]
 
 [package]
 name = "lan-mouse"
@@ -17,12 +25,14 @@ lto = "fat"
 input-event = { path = "input-event", version = "0.2.1" }
 input-emulation = { path = "input-emulation", version = "0.2.1", default-features = false }
 input-capture = { path = "input-capture", version = "0.2.0", default-features = false }
+lan-mouse-cli = { path = "lan-mouse-cli", version = "0.1.0" }
+lan-mouse-gtk = { path = "lan-mouse-gtk", version = "0.1.0", optional = true }
+lan-mouse-ipc = { path = "lan-mouse-ipc", version = "0.1.0" }
 lan-mouse-proto = { path = "lan-mouse-proto", version = "0.1.0" }
 
 hickory-resolver = "0.24.1"
 toml = "0.8"
 serde = { version = "1.0", features = ["derive"] }
-anyhow = "1.0.71"
 log = "0.4.20"
 env_logger = "0.11.3"
 serde_json = "1.0.107"
@@ -38,29 +48,18 @@ tokio = { version = "1.32.0", features = [
 ] }
 futures = "0.3.28"
 clap = { version = "4.4.11", features = ["derive"] }
-gtk = { package = "gtk4", version = "0.9.0", features = [
-    "v4_2",
-], optional = true }
-adw = { package = "libadwaita", version = "0.7.0", features = [
-    "v1_1",
-], optional = true }
-async-channel = { version = "2.1.1", optional = true }
-hostname = "0.4.0"
 slab = "0.4.9"
-endi = "1.1.0"
 thiserror = "1.0.61"
 tokio-util = "0.7.11"
+local-channel = "0.1.5"
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2.148"
 
-[build-dependencies]
-glib-build-tools = { version = "0.20.0", optional = true }
-
 [features]
 default = ["wayland", "x11", "xdg_desktop_portal", "libei", "gtk"]
 wayland = ["input-capture/wayland", "input-emulation/wayland"]
 x11 = ["input-capture/x11", "input-emulation/x11"]
 xdg_desktop_portal = ["input-emulation/xdg_desktop_portal"]
 libei = ["input-event/libei", "input-capture/libei", "input-emulation/libei"]
-gtk = ["dep:gtk", "dep:adw", "dep:async-channel", "dep:glib-build-tools"]
+gtk = ["dep:lan-mouse-gtk"]

README.md

@@ -20,7 +20,7 @@ Focus lies on performance and a clean, manageable implementation that can easily
 
 ***blazingly fast™*** because it's written in rust.
 
-For an alternative (with slightly different goals) you may check out [Input Leap](https://github.com/input-leap).
+For an alternative (with slightly different goals) you may check out [Synergy 1 Community Edition](https://github.com/symless/synergy) or [Input Leap](https://github.com/input-leap) (Synergy fork).
 
 
 > [!WARNING]
@@ -116,7 +116,7 @@ sudo cp target/release/lan-mouse /usr/local/bin/
 
 # install app icon
 sudo mkdir -p /usr/local/share/icons/hicolor/scalable/apps
-sudo cp resources/de.feschber.LanMouse.svg /usr/local/share/icons/hicolor/scalable/apps
+sudo cp lan-mouse-gtk/resources/de.feschber.LanMouse.svg /usr/local/share/icons/hicolor/scalable/apps
 
 # update icon cache
 gtk-update-icon-cache /usr/local/share/icons/hicolor/

build.rs

@@ -1,9 +1,15 @@
+use std::process::Command;
+
 fn main() {
-    // composite_templates
-    #[cfg(feature = "gtk")]
-    glib_build_tools::compile_resources(
-        &["resources"],
-        "resources/resources.gresource.xml",
-        "lan-mouse.gresource",
-    );
+    // commit hash
+    let git_describe = Command::new("git")
+        .arg("describe")
+        .arg("--always")
+        .arg("--dirty")
+        .arg("--tags")
+        .output()
+        .unwrap();
+
+    let git_describe = String::from_utf8(git_describe.stdout).unwrap();
+    println!("cargo::rustc-env=GIT_DESCRIBE={git_describe}");
 }

flake.nix

@@ -53,9 +53,11 @@
           libadwaita
           librsvg
           xorg.libXtst
-        ] ++ lib.optionals stdenv.isDarwin [
-          darwin.apple_sdk_11_0.frameworks.CoreGraphics
-        ];
+        ] ++ lib.optionals stdenv.isDarwin
+        (with darwin.apple_sdk_11_0.frameworks; [
+          CoreGraphics
+          ApplicationServices
+        ]);
 
         RUST_SRC_PATH = "${rust}/lib/rustlib/src/rust/library";
       };

input-capture/Cargo.toml

@@ -47,6 +47,11 @@ reis = { version = "0.2", features = ["tokio"], optional = true }
 
 [target.'cfg(target_os="macos")'.dependencies]
 core-graphics = { version = "0.23", features = ["highsierra"] }
+core-foundation = "0.9.4"
+core-foundation-sys = "0.8.6"
+libc = "0.2.155"
+keycode = "0.4.0"
+bitflags = "2.5.0"
 
 [target.'cfg(windows)'.dependencies]
 windows = { version = "0.58.0", features = [

input-capture/src/dummy.rs

@@ -1,16 +1,28 @@
+use std::f64::consts::PI;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
+use std::time::Duration;
 
 use async_trait::async_trait;
 use futures_core::Stream;
+use input_event::PointerEvent;
+use tokio::time::{self, Instant, Interval};
 
 use super::{Capture, CaptureError, CaptureEvent, CaptureHandle, Position};
 
-pub struct DummyInputCapture {}
+pub struct DummyInputCapture {
+    start: Option<Instant>,
+    interval: Interval,
+    offset: (i32, i32),
+}
 
 impl DummyInputCapture {
     pub fn new() -> Self {
-        Self {}
+        Self {
+            start: None,
+            interval: time::interval(Duration::from_millis(1)),
+            offset: (0, 0),
+        }
     }
 }
 
@@ -39,10 +51,36 @@ impl Capture for DummyInputCapture {
     }
 }
 
+const FREQUENCY_HZ: f64 = 1.0;
+const RADIUS: f64 = 100.0;
+
 impl Stream for DummyInputCapture {
     type Item = Result<(CaptureHandle, CaptureEvent), CaptureError>;
 
-    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        Poll::Pending
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let current = ready!(self.interval.poll_tick(cx));
+        let event = match self.start {
+            None => {
+                self.start.replace(current);
+                CaptureEvent::Begin
+            }
+            Some(start) => {
+                let elapsed = start.elapsed();
+                let elapsed_sec_f64 = elapsed.as_secs_f64();
+                let second_fraction = elapsed_sec_f64 - elapsed_sec_f64 as u64 as f64;
+                let radians = second_fraction * 2. * PI * FREQUENCY_HZ;
+                let offset = (radians.cos() * RADIUS * 2., (radians * 2.).sin() * RADIUS);
+                let offset = (offset.0 as i32, offset.1 as i32);
+                let relative_motion = (offset.0 - self.offset.0, offset.1 - self.offset.1);
+                self.offset = offset;
+                let (dx, dy) = (relative_motion.0 as f64, relative_motion.1 as f64);
+                CaptureEvent::Input(input_event::Event::Pointer(PointerEvent::Motion {
+                    time: 0,
+                    dx,
+                    dy,
+                }))
+            }
+        };
+        Poll::Ready(Some(Ok((0, event))))
     }
 }

input-capture/src/error.rs

@@ -22,6 +22,9 @@ use ashpd::desktop::ResponseError;
 #[cfg(all(unix, feature = "libei", not(target_os = "macos")))]
 use reis::tokio::{EiConvertEventStreamError, HandshakeError};
 
+#[cfg(target_os = "macos")]
+use core_graphics::base::CGError;
+
 #[cfg(all(unix, feature = "libei", not(target_os = "macos")))]
 #[derive(Debug, Error)]
 #[error("error in libei stream: {inner:?}")]
@@ -56,6 +59,21 @@ pub enum CaptureError {
     #[cfg(all(unix, feature = "libei", not(target_os = "macos")))]
     #[error("libei disconnected - reason: `{0}`")]
     Disconnected(String),
+    #[cfg(target_os = "macos")]
+    #[error("failed to warp mouse cursor: `{0}`")]
+    WarpCursor(CGError),
+    #[cfg(target_os = "macos")]
+    #[error("reset_mouse_position called without a connected client")]
+    ResetMouseWithoutClient,
+    #[cfg(target_os = "macos")]
+    #[error("core-graphics error: {0}")]
+    CoreGraphics(CGError),
+    #[cfg(target_os = "macos")]
+    #[error("unable to map key event: {0}")]
+    KeyMapError(i64),
+    #[cfg(target_os = "macos")]
+    #[error("Event tap disabled")]
+    EventTapDisabled,
 }
 
 #[derive(Debug, Error)]
@@ -71,12 +89,12 @@ pub enum CaptureCreationError {
     #[cfg(all(unix, feature = "x11", not(target_os = "macos")))]
     #[error("error creating x11 capture backend: `{0}`")]
     X11(#[from] X11InputCaptureCreationError),
-    #[cfg(target_os = "macos")]
-    #[error("error creating macos capture backend: `{0}`")]
-    Macos(#[from] MacOSInputCaptureCreationError),
     #[cfg(windows)]
     #[error("error creating windows capture backend")]
     Windows,
+    #[cfg(target_os = "macos")]
+    #[error("error creating macos capture backend")]
+    MacOS(#[from] MacosCaptureCreationError),
 }
 
 impl CaptureCreationError {
@@ -144,7 +162,12 @@ pub enum X11InputCaptureCreationError {
 
 #[cfg(target_os = "macos")]
 #[derive(Debug, Error)]
-pub enum MacOSInputCaptureCreationError {
-    #[error("MacOS input capture is not yet implemented :(")]
-    NotImplemented,
+pub enum MacosCaptureCreationError {
+    #[error("event source creation failed!")]
+    EventSourceCreation,
+    #[error("failed to set CG Cursor property")]
+    CGCursorProperty,
+    #[cfg(target_os = "macos")]
+    #[error("failed to get display ids: {0}")]
+    ActiveDisplays(CGError),
 }

input-capture/src/lib.rs

@@ -160,6 +160,7 @@ impl InputCapture {
                 _ => self.pressed_keys.remove(&scancode),
             };
         }
+        log::info!("pressed keys: {:?}", self.pressed_keys);
     }
 }
 
@@ -217,7 +218,7 @@ async fn create_backend(
         #[cfg(windows)]
         Backend::Windows => Ok(Box::new(windows::WindowsInputCapture::new())),
         #[cfg(target_os = "macos")]
-        Backend::MacOs => Ok(Box::new(macos::MacOSInputCapture::new()?)),
+        Backend::MacOs => Ok(Box::new(macos::MacOSInputCapture::new().await?)),
         Backend::Dummy => Ok(Box::new(dummy::DummyInputCapture::new())),
     }
 }

input-capture/src/macos.rs

@@ -1,39 +1,522 @@
-use crate::{
-    error::MacOSInputCaptureCreationError, Capture, CaptureError, CaptureEvent, CaptureHandle,
-    Position,
+use super::{
+    error::MacosCaptureCreationError, Capture, CaptureError, CaptureEvent, CaptureHandle, Position,
 };
 use async_trait::async_trait;
+use bitflags::bitflags;
+use core_foundation::base::{kCFAllocatorDefault, CFRelease};
+use core_foundation::date::CFTimeInterval;
+use core_foundation::number::{kCFBooleanTrue, CFBooleanRef};
+use core_foundation::runloop::{kCFRunLoopCommonModes, CFRunLoop, CFRunLoopSource};
+use core_foundation::string::{kCFStringEncodingUTF8, CFStringCreateWithCString, CFStringRef};
+use core_graphics::base::{kCGErrorSuccess, CGError};
+use core_graphics::display::{CGDisplay, CGPoint};
+use core_graphics::event::{
+    CGEvent, CGEventFlags, CGEventTap, CGEventTapLocation, CGEventTapOptions, CGEventTapPlacement,
+    CGEventTapProxy, CGEventType, EventField,
+};
+use core_graphics::event_source::{CGEventSource, CGEventSourceStateID};
 use futures_core::Stream;
+use input_event::{Event, KeyboardEvent, PointerEvent, BTN_LEFT, BTN_MIDDLE, BTN_RIGHT};
+use keycode::{KeyMap, KeyMapping};
+use libc::c_void;
+use once_cell::unsync::Lazy;
+use std::collections::HashMap;
+use std::ffi::{c_char, CString};
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::sync::Arc;
+use std::task::{ready, Context, Poll};
+use std::thread::{self};
+use tokio::sync::mpsc::{Receiver, Sender};
+use tokio::sync::Mutex;
 
-pub struct MacOSInputCapture;
+#[derive(Debug, Default)]
+struct Bounds {
+    xmin: f64,
+    xmax: f64,
+    ymin: f64,
+    ymax: f64,
+}
 
-impl MacOSInputCapture {
-    pub fn new() -> std::result::Result<Self, MacOSInputCaptureCreationError> {
-        Err(MacOSInputCaptureCreationError::NotImplemented)
+#[derive(Debug)]
+struct InputCaptureState {
+    client_for_pos: Lazy<HashMap<Position, CaptureHandle>>,
+    current_client: Option<(CaptureHandle, Position)>,
+    bounds: Bounds,
+}
+
+#[derive(Debug)]
+enum ProducerEvent {
+    Release,
+    Create(CaptureHandle, Position),
+    Destroy(CaptureHandle),
+    Grab((CaptureHandle, Position)),
+    EventTapDisabled,
+}
+
+impl InputCaptureState {
+    fn new() -> Result<Self, MacosCaptureCreationError> {
+        let mut res = Self {
+            client_for_pos: Lazy::new(HashMap::new),
+            current_client: None,
+            bounds: Bounds::default(),
+        };
+        res.update_bounds()?;
+        Ok(res)
+    }
+
+    fn crossed(&mut self, event: &CGEvent) -> Option<(CaptureHandle, Position)> {
+        let location = event.location();
+        let relative_x = event.get_double_value_field(EventField::MOUSE_EVENT_DELTA_X);
+        let relative_y = event.get_double_value_field(EventField::MOUSE_EVENT_DELTA_Y);
+
+        for (position, client) in self.client_for_pos.iter() {
+            if (position == &Position::Left && (location.x + relative_x) <= self.bounds.xmin)
+                || (position == &Position::Right && (location.x + relative_x) >= self.bounds.xmax)
+                || (position == &Position::Top && (location.y + relative_y) <= self.bounds.ymin)
+                || (position == &Position::Bottom && (location.y + relative_y) >= self.bounds.ymax)
+            {
+                log::debug!("Crossed barrier into client: {client}, {position:?}");
+                return Some((*client, *position));
+            }
+        }
+        None
+    }
+
+    // Get the max bounds of all displays
+    fn update_bounds(&mut self) -> Result<(), MacosCaptureCreationError> {
+        let active_ids =
+            CGDisplay::active_displays().map_err(MacosCaptureCreationError::ActiveDisplays)?;
+        active_ids.iter().for_each(|d| {
+            let bounds = CGDisplay::new(*d).bounds();
+            self.bounds.xmin = self.bounds.xmin.min(bounds.origin.x);
+            self.bounds.xmax = self.bounds.xmax.max(bounds.origin.x + bounds.size.width);
+            self.bounds.ymin = self.bounds.ymin.min(bounds.origin.y);
+            self.bounds.ymax = self.bounds.ymax.max(bounds.origin.y + bounds.size.height);
+        });
+
+        log::debug!("Updated displays bounds: {0:?}", self.bounds);
+        Ok(())
+    }
+
+    // We can't disable mouse movement when in a client so we need to reset the cursor position
+    // to the edge of the screen, the cursor will be hidden but we dont want it to appear in a
+    // random location when we exit the client
+    fn reset_mouse_position(&self, event: &CGEvent) -> Result<(), CaptureError> {
+        if let Some((_, pos)) = self.current_client {
+            let location = event.location();
+            let edge_offset = 1.0;
+
+            // After the cursor is warped no event is produced but the next event
+            // will carry the delta from the warp so only half the delta is needed to move the cursor
+            let delta_y = event.get_double_value_field(EventField::MOUSE_EVENT_DELTA_Y) / 2.0;
+            let delta_x = event.get_double_value_field(EventField::MOUSE_EVENT_DELTA_X) / 2.0;
+
+            let mut new_x = location.x + delta_x;
+            let mut new_y = location.y + delta_y;
+
+            match pos {
+                Position::Left => {
+                    new_x = self.bounds.xmin + edge_offset;
+                }
+                Position::Right => {
+                    new_x = self.bounds.xmax - edge_offset;
+                }
+                Position::Top => {
+                    new_y = self.bounds.ymin + edge_offset;
+                }
+                Position::Bottom => {
+                    new_y = self.bounds.ymax - edge_offset;
+                }
+            }
+            let new_pos = CGPoint::new(new_x, new_y);
+
+            log::trace!("Resetting cursor position to: {new_x}, {new_y}");
+
+            return CGDisplay::warp_mouse_cursor_position(new_pos)
+                .map_err(CaptureError::WarpCursor);
+        }
+
+        Err(CaptureError::ResetMouseWithoutClient)
+    }
+
+    async fn handle_producer_event(
+        &mut self,
+        producer_event: ProducerEvent,
+    ) -> Result<(), CaptureError> {
+        log::debug!("handling event: {producer_event:?}");
+        match producer_event {
+            ProducerEvent::Release => {
+                if self.current_client.is_some() {
+                    CGDisplay::show_cursor(&CGDisplay::main())
+                        .map_err(CaptureError::CoreGraphics)?;
+                    self.current_client = None;
+                }
+            }
+            ProducerEvent::Grab(client) => {
+                if self.current_client.is_none() {
+                    CGDisplay::hide_cursor(&CGDisplay::main())
+                        .map_err(CaptureError::CoreGraphics)?;
+                    self.current_client = Some(client);
+                }
+            }
+            ProducerEvent::Create(c, p) => {
+                self.client_for_pos.insert(p, c);
+            }
+            ProducerEvent::Destroy(c) => {
+                for pos in [
+                    Position::Left,
+                    Position::Right,
+                    Position::Top,
+                    Position::Bottom,
+                ] {
+                    if let Some((current_c, _)) = self.current_client {
+                        if current_c == c {
+                            CGDisplay::show_cursor(&CGDisplay::main())
+                                .map_err(CaptureError::CoreGraphics)?;
+                            self.current_client = None;
+                        };
+                    }
+                    if self.client_for_pos.get(&pos).copied() == Some(c) {
+                        self.client_for_pos.remove(&pos);
+                    }
+                }
+            }
+            ProducerEvent::EventTapDisabled => return Err(CaptureError::EventTapDisabled),
+        };
+        Ok(())
     }
 }
 
-impl Stream for MacOSInputCapture {
-    type Item = Result<(CaptureHandle, CaptureEvent), CaptureError>;
+fn get_events(
+    ev_type: &CGEventType,
+    ev: &CGEvent,
+    result: &mut Vec<CaptureEvent>,
+) -> Result<(), CaptureError> {
+    fn map_pointer_event(ev: &CGEvent) -> PointerEvent {
+        PointerEvent::Motion {
+            time: 0,
+            dx: ev.get_double_value_field(EventField::MOUSE_EVENT_DELTA_X),
+            dy: ev.get_double_value_field(EventField::MOUSE_EVENT_DELTA_Y),
+        }
+    }
 
-    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        Poll::Pending
+    fn map_key(ev: &CGEvent) -> Result<u32, CaptureError> {
+        let code = ev.get_integer_value_field(EventField::KEYBOARD_EVENT_KEYCODE);
+        match KeyMap::from_key_mapping(KeyMapping::Mac(code as u16)) {
+            Ok(k) => Ok(k.evdev as u32),
+            Err(()) => Err(CaptureError::KeyMapError(code)),
+        }
+    }
+
+    match ev_type {
+        CGEventType::KeyDown => {
+            let k = map_key(ev)?;
+            result.push(CaptureEvent::Input(Event::Keyboard(KeyboardEvent::Key {
+                time: 0,
+                key: k,
+                state: 1,
+            })));
+        }
+        CGEventType::KeyUp => {
+            let k = map_key(ev)?;
+            result.push(CaptureEvent::Input(Event::Keyboard(KeyboardEvent::Key {
+                time: 0,
+                key: k,
+                state: 0,
+            })));
+        }
+        CGEventType::FlagsChanged => {
+            let mut mods = XMods::empty();
+            let mut mods_locked = XMods::empty();
+            let cg_flags = ev.get_flags();
+
+            if cg_flags.contains(CGEventFlags::CGEventFlagShift) {
+                mods |= XMods::ShiftMask;
+            }
+            if cg_flags.contains(CGEventFlags::CGEventFlagControl) {
+                mods |= XMods::ControlMask;
+            }
+            if cg_flags.contains(CGEventFlags::CGEventFlagAlternate) {
+                mods |= XMods::Mod1Mask;
+            }
+            if cg_flags.contains(CGEventFlags::CGEventFlagCommand) {
+                mods |= XMods::Mod4Mask;
+            }
+            if cg_flags.contains(CGEventFlags::CGEventFlagAlphaShift) {
+                mods |= XMods::LockMask;
+                mods_locked |= XMods::LockMask;
+            }
+
+            let modifier_event = KeyboardEvent::Modifiers {
+                depressed: mods.bits(),
+                latched: 0,
+                locked: mods_locked.bits(),
+                group: 0,
+            };
+
+            result.push(CaptureEvent::Input(Event::Keyboard(modifier_event)));
+        }
+        CGEventType::MouseMoved => {
+            result.push(CaptureEvent::Input(Event::Pointer(map_pointer_event(ev))))
+        }
+        CGEventType::LeftMouseDragged => {
+            result.push(CaptureEvent::Input(Event::Pointer(map_pointer_event(ev))))
+        }
+        CGEventType::RightMouseDragged => {
+            result.push(CaptureEvent::Input(Event::Pointer(map_pointer_event(ev))))
+        }
+        CGEventType::OtherMouseDragged => {
+            result.push(CaptureEvent::Input(Event::Pointer(map_pointer_event(ev))))
+        }
+        CGEventType::LeftMouseDown => {
+            result.push(CaptureEvent::Input(Event::Pointer(PointerEvent::Button {
+                time: 0,
+                button: BTN_LEFT,
+                state: 1,
+            })))
+        }
+        CGEventType::LeftMouseUp => {
+            result.push(CaptureEvent::Input(Event::Pointer(PointerEvent::Button {
+                time: 0,
+                button: BTN_LEFT,
+                state: 0,
+            })))
+        }
+        CGEventType::RightMouseDown => {
+            result.push(CaptureEvent::Input(Event::Pointer(PointerEvent::Button {
+                time: 0,
+                button: BTN_RIGHT,
+                state: 1,
+            })))
+        }
+        CGEventType::RightMouseUp => {
+            result.push(CaptureEvent::Input(Event::Pointer(PointerEvent::Button {
+                time: 0,
+                button: BTN_RIGHT,
+                state: 0,
+            })))
+        }
+        CGEventType::OtherMouseDown => {
+            result.push(CaptureEvent::Input(Event::Pointer(PointerEvent::Button {
+                time: 0,
+                button: BTN_MIDDLE,
+                state: 1,
+            })))
+        }
+        CGEventType::OtherMouseUp => {
+            result.push(CaptureEvent::Input(Event::Pointer(PointerEvent::Button {
+                time: 0,
+                button: BTN_MIDDLE,
+                state: 0,
+            })))
+        }
+        CGEventType::ScrollWheel => {
+            let v = ev.get_integer_value_field(EventField::SCROLL_WHEEL_EVENT_POINT_DELTA_AXIS_1);
+            let h = ev.get_integer_value_field(EventField::SCROLL_WHEEL_EVENT_POINT_DELTA_AXIS_2);
+            if v != 0 {
+                result.push(CaptureEvent::Input(Event::Pointer(PointerEvent::Axis {
+                    time: 0,
+                    axis: 0, // Vertical
+                    value: v as f64,
+                })));
+            }
+            if h != 0 {
+                result.push(CaptureEvent::Input(Event::Pointer(PointerEvent::Axis {
+                    time: 0,
+                    axis: 1, // Horizontal
+                    value: h as f64,
+                })));
+            }
+        }
+        _ => (),
+    }
+    Ok(())
+}
+
+fn event_tap_thread(
+    client_state: Arc<Mutex<InputCaptureState>>,
+    event_tx: Sender<(CaptureHandle, CaptureEvent)>,
+    notify_tx: Sender<ProducerEvent>,
+    exit: tokio::sync::oneshot::Sender<Result<(), &'static str>>,
+) {
+    let cg_events_of_interest: Vec<CGEventType> = vec![
+        CGEventType::LeftMouseDown,
+        CGEventType::LeftMouseUp,
+        CGEventType::RightMouseDown,
+        CGEventType::RightMouseUp,
+        CGEventType::OtherMouseDown,
+        CGEventType::OtherMouseUp,
+        CGEventType::MouseMoved,
+        CGEventType::LeftMouseDragged,
+        CGEventType::RightMouseDragged,
+        CGEventType::OtherMouseDragged,
+        CGEventType::ScrollWheel,
+        CGEventType::KeyDown,
+        CGEventType::KeyUp,
+        CGEventType::FlagsChanged,
+    ];
+
+    let tap = CGEventTap::new(
+        CGEventTapLocation::Session,
+        CGEventTapPlacement::HeadInsertEventTap,
+        CGEventTapOptions::Default,
+        cg_events_of_interest,
+        |_proxy: CGEventTapProxy, event_type: CGEventType, cg_ev: &CGEvent| {
+            log::trace!("Got event from tap: {event_type:?}");
+            let mut state = client_state.blocking_lock();
+            let mut client = None;
+            let mut res_events = vec![];
+
+            if matches!(
+                event_type,
+                CGEventType::TapDisabledByTimeout | CGEventType::TapDisabledByUserInput
+            ) {
+                log::error!("CGEventTap disabled");
+                notify_tx
+                    .blocking_send(ProducerEvent::EventTapDisabled)
+                    .unwrap_or_else(|e| {
+                        log::error!("Failed to send notification: {e}");
+                    });
+            }
+
+            // Are we in a client?
+            if let Some((current_client, _)) = state.current_client {
+                client = Some(current_client);
+                get_events(&event_type, cg_ev, &mut res_events).unwrap_or_else(|e| {
+                    log::error!("Failed to get events: {e}");
+                });
+
+                // Keep (hidden) cursor at the edge of the screen
+                if matches!(event_type, CGEventType::MouseMoved) {
+                    state.reset_mouse_position(cg_ev).unwrap_or_else(|e| {
+                        log::error!("Failed to reset mouse position: {e}");
+                    })
+                }
+            }
+            // Did we cross a barrier?
+            else if matches!(event_type, CGEventType::MouseMoved) {
+                if let Some((new_client, pos)) = state.crossed(cg_ev) {
+                    client = Some(new_client);
+                    res_events.push(CaptureEvent::Begin);
+                    notify_tx
+                        .blocking_send(ProducerEvent::Grab((new_client, pos)))
+                        .expect("Failed to send notification");
+                }
+            }
+
+            if let Some(client) = client {
+                res_events.iter().for_each(|e| {
+                    event_tx
+                        .blocking_send((client, *e))
+                        .expect("Failed to send event");
+                });
+                // Returning None should stop the event from being processed
+                // but core fundation still returns the event
+                cg_ev.set_type(CGEventType::Null);
+            }
+            Some(cg_ev.to_owned())
+        },
+    )
+    .expect("Failed creating tap");
+
+    let tap_source: CFRunLoopSource = tap
+        .mach_port
+        .create_runloop_source(0)
+        .expect("Failed creating loop source");
+
+    unsafe {
+        CFRunLoop::get_current().add_source(&tap_source, kCFRunLoopCommonModes);
+    }
+
+    CFRunLoop::run_current();
+
+    let _ = exit.send(Err("tap thread exited"));
+}
+
+pub struct MacOSInputCapture {
+    event_rx: Receiver<(CaptureHandle, CaptureEvent)>,
+    notify_tx: Sender<ProducerEvent>,
+}
+
+impl MacOSInputCapture {
+    pub async fn new() -> Result<Self, MacosCaptureCreationError> {
+        let state = Arc::new(Mutex::new(InputCaptureState::new()?));
+        let (event_tx, event_rx) = tokio::sync::mpsc::channel(32);
+        let (notify_tx, mut notify_rx) = tokio::sync::mpsc::channel(32);
+        let (tap_exit_tx, mut tap_exit_rx) = tokio::sync::oneshot::channel();
+
+        unsafe {
+            configure_cf_settings()?;
+        }
+
+        log::info!("Enabling CGEvent tap");
+        let event_tap_thread_state = state.clone();
+        let event_tap_notify = notify_tx.clone();
+        thread::spawn(move || {
+            event_tap_thread(
+                event_tap_thread_state,
+                event_tx,
+                event_tap_notify,
+                tap_exit_tx,
+            )
+        });
+
+        let _tap_task: tokio::task::JoinHandle<()> = tokio::task::spawn_local(async move {
+            loop {
+                tokio::select! {
+                    producer_event = notify_rx.recv() => {
+                        let producer_event = producer_event.expect("channel closed");
+                        let mut state = state.lock().await;
+                        state.handle_producer_event(producer_event).await.unwrap_or_else(|e| {
+                            log::error!("Failed to handle producer event: {e}");
+                        })
+                    }
+
+                    res = &mut tap_exit_rx => {
+                        if let Err(e) = res.expect("channel closed") {
+                            log::error!("Tap thread failed: {:?}", e);
+                            break;
+                        }
+                    }
+                }
+            }
+        });
+
+        Ok(Self {
+            event_rx,
+            notify_tx,
+        })
     }
 }
 
 #[async_trait]
 impl Capture for MacOSInputCapture {
-    async fn create(&mut self, _id: CaptureHandle, _pos: Position) -> Result<(), CaptureError> {
+    async fn create(&mut self, id: CaptureHandle, pos: Position) -> Result<(), CaptureError> {
+        let notify_tx = self.notify_tx.clone();
+        tokio::task::spawn_local(async move {
+            log::debug!("creating client {id}, {pos}");
+            let _ = notify_tx.send(ProducerEvent::Create(id, pos)).await;
+            log::debug!("done !");
+        });
         Ok(())
     }
 
-    async fn destroy(&mut self, _id: CaptureHandle) -> Result<(), CaptureError> {
+    async fn destroy(&mut self, id: CaptureHandle) -> Result<(), CaptureError> {
+        let notify_tx = self.notify_tx.clone();
+        tokio::task::spawn_local(async move {
+            log::debug!("destroying client {id}");
+            let _ = notify_tx.send(ProducerEvent::Destroy(id)).await;
+            log::debug!("done !");
+        });
         Ok(())
     }
 
     async fn release(&mut self) -> Result<(), CaptureError> {
+        let notify_tx = self.notify_tx.clone();
+        tokio::task::spawn_local(async move {
+            log::debug!("notifying Release");
+            let _ = notify_tx.send(ProducerEvent::Release).await;
+        });
         Ok(())
     }
 
@@ -41,3 +524,79 @@ impl Capture for MacOSInputCapture {
         Ok(())
     }
 }
+
+impl Stream for MacOSInputCapture {
+    type Item = Result<(CaptureHandle, CaptureEvent), CaptureError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        match ready!(self.event_rx.poll_recv(cx)) {
+            None => Poll::Ready(None),
+            Some(e) => Poll::Ready(Some(Ok(e))),
+        }
+    }
+}
+
+type CGSConnectionID = u32;
+
+#[link(name = "ApplicationServices", kind = "framework")]
+extern "C" {
+    fn CGSSetConnectionProperty(
+        cid: CGSConnectionID,
+        targetCID: CGSConnectionID,
+        key: CFStringRef,
+        value: CFBooleanRef,
+    ) -> CGError;
+    fn _CGSDefaultConnection() -> CGSConnectionID;
+}
+
+extern "C" {
+    fn CGEventSourceSetLocalEventsSuppressionInterval(
+        event_source: CGEventSource,
+        seconds: CFTimeInterval,
+    );
+}
+
+unsafe fn configure_cf_settings() -> Result<(), MacosCaptureCreationError> {
+    // When we warp the cursor using CGWarpMouseCursorPosition local events are suppressed for a short time
+    // this leeds to the cursor not flowing when crossing back from a clinet, set this to to 0 stops the warp
+    // from working, set a low value by trial and error, 0.05s seems good. 0.25s is the default
+    let event_source = CGEventSource::new(CGEventSourceStateID::CombinedSessionState)
+        .map_err(|_| MacosCaptureCreationError::EventSourceCreation)?;
+    CGEventSourceSetLocalEventsSuppressionInterval(event_source, 0.05);
+
+    // This is a private settings that allows the cursor to be hidden while in the background.
+    // It is used by Barrier and other apps.
+    let key = CString::new("SetsCursorInBackground").unwrap();
+    let cf_key = CFStringCreateWithCString(
+        kCFAllocatorDefault,
+        key.as_ptr() as *const c_char,
+        kCFStringEncodingUTF8,
+    );
+    if CGSSetConnectionProperty(
+        _CGSDefaultConnection(),
+        _CGSDefaultConnection(),
+        cf_key,
+        kCFBooleanTrue,
+    ) != kCGErrorSuccess
+    {
+        return Err(MacosCaptureCreationError::CGCursorProperty);
+    }
+    CFRelease(cf_key as *const c_void);
+    Ok(())
+}
+
+// From X11/X.h
+bitflags! {
+    #[repr(C)]
+    #[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
+    struct XMods: u32 {
+        const ShiftMask = (1<<0);
+        const LockMask = (1<<1);
+        const ControlMask = (1<<2);
+        const Mod1Mask = (1<<3);
+        const Mod2Mask = (1<<4);
+        const Mod3Mask = (1<<5);
+        const Mod4Mask = (1<<6);
+        const Mod5Mask = (1<<7);
+    }
+}

input-capture/src/wayland.rs

@@ -71,7 +71,7 @@ struct Globals {
     compositor: wl_compositor::WlCompositor,
     pointer_constraints: ZwpPointerConstraintsV1,
     relative_pointer_manager: ZwpRelativePointerManagerV1,
-    shortcut_inhibit_manager: ZwpKeyboardShortcutsInhibitManagerV1,
+    shortcut_inhibit_manager: Option<ZwpKeyboardShortcutsInhibitManagerV1>,
     seat: wl_seat::WlSeat,
     shm: wl_shm::WlShm,
     layer_shell: ZwlrLayerShellV1,
@@ -285,9 +285,18 @@ impl WaylandInputCapture {
         let relative_pointer_manager: ZwpRelativePointerManagerV1 = g
             .bind(&qh, 1..=1, ())
             .map_err(|e| WaylandBindError::new(e, "zwp_relative_pointer_manager_v1"))?;
-        let shortcut_inhibit_manager: ZwpKeyboardShortcutsInhibitManagerV1 = g
+        let shortcut_inhibit_manager: Result<
+            ZwpKeyboardShortcutsInhibitManagerV1,
+            WaylandBindError,
+        > = g
             .bind(&qh, 1..=1, ())
-            .map_err(|e| WaylandBindError::new(e, "zwp_keyboard_shortcuts_inhibit_manager_v1"))?;
+            .map_err(|e| WaylandBindError::new(e, "zwp_keyboard_shortcuts_inhibit_manager_v1"));
+        // layer-shell backend still works without this protocol so we make it an optional dependency
+        if let Err(e) = &shortcut_inhibit_manager {
+            log::warn!("shortcut_inhibit_manager not supported: {e}\nkeybinds handled by the compositor will not be passed
+                to the client");
+        }
+        let shortcut_inhibit_manager = shortcut_inhibit_manager.ok();
         let outputs = vec![];
 
         let g = Globals {
@@ -424,13 +433,11 @@ impl State {
         }
 
         // capture modifier keys
-        if self.shortcut_inhibitor.is_none() {
-            self.shortcut_inhibitor = Some(self.g.shortcut_inhibit_manager.inhibit_shortcuts(
-                surface,
-                &self.g.seat,
-                qh,
-                (),
-            ));
+        if let Some(shortcut_inhibit_manager) = &self.g.shortcut_inhibit_manager {
+            if self.shortcut_inhibitor.is_none() {
+                self.shortcut_inhibitor =
+                    Some(shortcut_inhibit_manager.inhibit_shortcuts(surface, &self.g.seat, qh, ()));
+            }
         }
     }
 

input-emulation/src/lib.rs

@@ -217,13 +217,15 @@ impl InputEmulation {
             return false;
         };
 
-        if state == 0 {
+        let ret = if state == 0 {
             // currently pressed => can release
             pressed_keys.remove(&key)
         } else {
             // currently not pressed => can press
             pressed_keys.insert(key)
-        }
+        };
+        log::info!("client {handle}: pressed keys: {:?}", pressed_keys);
+        ret
     }
 }
 

input-emulation/src/macos.rs

@@ -1,6 +1,9 @@
 use super::{error::EmulationError, Emulation, EmulationHandle};
 use async_trait::async_trait;
-use core_graphics::display::{CGDisplayBounds, CGMainDisplayID, CGPoint};
+use core_graphics::base::CGFloat;
+use core_graphics::display::{
+    CGDirectDisplayID, CGDisplayBounds, CGGetDisplaysWithRect, CGPoint, CGRect, CGSize,
+};
 use core_graphics::event::{
     CGEvent, CGEventTapLocation, CGEventType, CGKeyCode, CGMouseButton, EventField, ScrollEventUnit,
 };
@@ -105,6 +108,77 @@ fn key_event(event_source: CGEventSource, key: u16, state: u8) {
     event.post(CGEventTapLocation::HID);
 }
 
+fn get_display_at_point(x: CGFloat, y: CGFloat) -> Option<CGDirectDisplayID> {
+    let mut displays: [CGDirectDisplayID; 16] = [0; 16];
+    let mut display_count: u32 = 0;
+    let rect = CGRect::new(&CGPoint::new(x, y), &CGSize::new(0.0, 0.0));
+
+    let error = unsafe {
+        CGGetDisplaysWithRect(
+            rect,
+            1,
+            displays.as_mut_ptr(),
+            &mut display_count as *mut u32,
+        )
+    };
+
+    if error != 0 {
+        log::warn!("error getting displays at point ({}, {}): {}", x, y, error);
+        return Option::None;
+    }
+
+    if display_count == 0 {
+        log::debug!("no displays found at point ({}, {})", x, y);
+        return Option::None;
+    }
+
+    return displays.first().copied();
+}
+
+fn get_display_bounds(display: CGDirectDisplayID) -> (CGFloat, CGFloat, CGFloat, CGFloat) {
+    unsafe {
+        let bounds = CGDisplayBounds(display);
+        let min_x = bounds.origin.x;
+        let max_x = bounds.origin.x + bounds.size.width;
+        let min_y = bounds.origin.y;
+        let max_y = bounds.origin.y + bounds.size.height;
+        (min_x as f64, min_y as f64, max_x as f64, max_y as f64)
+    }
+}
+
+fn clamp_to_screen_space(
+    current_x: CGFloat,
+    current_y: CGFloat,
+    dx: CGFloat,
+    dy: CGFloat,
+) -> (CGFloat, CGFloat) {
+    // Check which display the mouse is currently on
+    // Determine what the location of the mouse would be after applying the move
+    // Get the display at the new location
+    // If the point is not on a display
+    //   Clamp the mouse to the current display
+    // Else If the point is on a display
+    //   Clamp the mouse to the new display
+    let current_display = match get_display_at_point(current_x, current_y) {
+        Some(display) => display,
+        None => {
+            log::warn!("could not get current display!");
+            return (current_x, current_y);
+        }
+    };
+
+    let new_x = current_x + dx;
+    let new_y = current_y + dy;
+
+    let final_display = get_display_at_point(new_x, new_y).unwrap_or(current_display);
+    let (min_x, min_y, max_x, max_y) = get_display_bounds(final_display);
+
+    (
+        new_x.clamp(min_x, max_x - 1.),
+        new_y.clamp(min_y, max_y - 1.),
+    )
+}
+
 #[async_trait]
 impl Emulation for MacOSEmulation {
     async fn consume(
@@ -115,16 +189,6 @@ impl Emulation for MacOSEmulation {
         match event {
             Event::Pointer(pointer_event) => match pointer_event {
                 PointerEvent::Motion { time: _, dx, dy } => {
-                    // FIXME secondary displays?
-                    let (min_x, min_y, max_x, max_y) = unsafe {
-                        let display = CGMainDisplayID();
-                        let bounds = CGDisplayBounds(display);
-                        let min_x = bounds.origin.x;
-                        let max_x = bounds.origin.x + bounds.size.width;
-                        let min_y = bounds.origin.y;
-                        let max_y = bounds.origin.y + bounds.size.height;
-                        (min_x as f64, min_y as f64, max_x as f64, max_y as f64)
-                    };
                     let mut mouse_location = match self.get_mouse_location() {
                         Some(l) => l,
                         None => {
@@ -133,8 +197,11 @@ impl Emulation for MacOSEmulation {
                         }
                     };
 
-                    mouse_location.x = (mouse_location.x + dx).clamp(min_x, max_x - 1.);
-                    mouse_location.y = (mouse_location.y + dy).clamp(min_y, max_y - 1.);
+                    let (new_mouse_x, new_mouse_y) =
+                        clamp_to_screen_space(mouse_location.x, mouse_location.y, dx, dy);
+
+                    mouse_location.x = new_mouse_x;
+                    mouse_location.y = new_mouse_y;
 
                     let mut event_type = CGEventType::MouseMoved;
                     if self.button_state.left {

lan-mouse-cli/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "lan-mouse-cli"
+description = "CLI Frontend for lan-mouse"
+version = "0.1.0"
+edition = "2021"
+license = "GPL-3.0-or-later"
+repository = "https://github.com/feschber/lan-mouse"
+
+[dependencies]
+futures = "0.3.30"
+lan-mouse-ipc = { path = "../lan-mouse-ipc", version = "0.1.0" }
+tokio = { version = "1.32.0", features = [
+    "io-util",
+    "io-std",
+    "macros",
+    "net",
+    "rt",
+] }

lan-mouse-cli/src/command.rs

@@ -3,7 +3,7 @@ use std::{
     str::{FromStr, SplitWhitespace},
 };
 
-use crate::client::{ClientHandle, Position};
+use lan_mouse_ipc::{ClientHandle, Position};
 
 pub(super) enum CommandType {
     NoCommand,

lan-mouse-cli/src/lib.rs

@@ -1,78 +1,65 @@
-use anyhow::{anyhow, Result};
+use futures::StreamExt;
 use tokio::{
-    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
+    io::{AsyncBufReadExt, BufReader},
     task::LocalSet,
 };
 
-#[cfg(windows)]
-use tokio::net::tcp::{ReadHalf, WriteHalf};
-#[cfg(unix)]
-use tokio::net::unix::{ReadHalf, WriteHalf};
-
 use std::io::{self, Write};
 
-use crate::{
-    client::{ClientConfig, ClientHandle, ClientState},
-    config::DEFAULT_PORT,
-};
-
 use self::command::{Command, CommandType};
 
-use super::{FrontendEvent, FrontendRequest};
+use lan_mouse_ipc::{
+    AsyncFrontendEventReader, AsyncFrontendRequestWriter, ClientConfig, ClientHandle, ClientState,
+    FrontendEvent, FrontendRequest, IpcError, DEFAULT_PORT,
+};
 
 mod command;
 
-pub fn run() -> Result<()> {
-    let Ok(stream) = super::wait_for_service() else {
-        return Err(anyhow!("Could not connect to lan-mouse-socket"));
-    };
-
+pub fn run() -> Result<(), IpcError> {
     let runtime = tokio::runtime::Builder::new_current_thread()
         .enable_io()
         .enable_time()
         .build()?;
     runtime.block_on(LocalSet::new().run_until(async move {
-        stream.set_nonblocking(true)?;
-        #[cfg(unix)]
-        let mut stream = tokio::net::UnixStream::from_std(stream)?;
-        #[cfg(windows)]
-        let mut stream = tokio::net::TcpStream::from_std(stream)?;
-        let (rx, tx) = stream.split();
-
+        let (rx, tx) = lan_mouse_ipc::connect_async().await?;
         let mut cli = Cli::new(rx, tx);
         cli.run().await
     }))?;
     Ok(())
 }
 
-struct Cli<'a> {
+struct Cli {
     clients: Vec<(ClientHandle, ClientConfig, ClientState)>,
-    rx: ReadHalf<'a>,
-    tx: WriteHalf<'a>,
+    changed: Option<ClientHandle>,
+    rx: AsyncFrontendEventReader,
+    tx: AsyncFrontendRequestWriter,
 }
 
-impl<'a> Cli<'a> {
-    fn new(rx: ReadHalf<'a>, tx: WriteHalf<'a>) -> Cli<'a> {
+impl Cli {
+    fn new(rx: AsyncFrontendEventReader, tx: AsyncFrontendRequestWriter) -> Cli {
         Self {
             clients: vec![],
+            changed: None,
             rx,
             tx,
         }
     }
 
-    async fn run(&mut self) -> Result<()> {
+    async fn run(&mut self) -> Result<(), IpcError> {
         let stdin = tokio::io::stdin();
         let stdin = BufReader::new(stdin);
         let mut stdin = stdin.lines();
 
         /* initial state sync */
-        let request = FrontendRequest::Enumerate();
-        self.send_request(request).await?;
-
         self.clients = loop {
-            let event = self.await_event().await?;
-            if let FrontendEvent::Enumerate(clients) = event {
-                break clients;
+            match self.rx.next().await {
+                Some(Ok(e)) => {
+                    if let FrontendEvent::Enumerate(clients) = e {
+                        break clients;
+                    }
+                }
+                Some(Err(e)) => return Err(e),
+                None => return Ok(()),
             }
         };
 
@@ -92,18 +79,23 @@ impl<'a> Cli<'a> {
                     };
                     self.execute(cmd).await?;
                 }
-                event = self.await_event() => {
-                    let event = event?;
-                    self.handle_event(event);
+                event = self.rx.next() => {
+                    if let Some(event) = event {
+                        self.handle_event(event?);
+                    } else {
+                        break Ok(());
+                    }
                 }
             }
+            if let Some(handle) = self.changed.take() {
+                self.update_client(handle).await?;
+            }
         }
     }
 
-    async fn update_client(&mut self, handle: ClientHandle) -> Result<()> {
-        self.send_request(FrontendRequest::GetState(handle)).await?;
-        loop {
-            let event = self.await_event().await?;
+    async fn update_client(&mut self, handle: ClientHandle) -> Result<(), IpcError> {
+        self.tx.request(FrontendRequest::GetState(handle)).await?;
+        while let Some(Ok(event)) = self.rx.next().await {
             self.handle_event(event.clone());
             if let FrontendEvent::State(_, _, _) | FrontendEvent::NoSuchClient(_) = event {
                 break;
@@ -112,22 +104,23 @@ impl<'a> Cli<'a> {
         Ok(())
     }
 
-    async fn execute(&mut self, cmd: Command) -> Result<()> {
+    async fn execute(&mut self, cmd: Command) -> Result<(), IpcError> {
         match cmd {
             Command::None => {}
             Command::Connect(pos, host, port) => {
                 let request = FrontendRequest::Create;
-                self.send_request(request).await?;
+                self.tx.request(request).await?;
                 let handle = loop {
-                    let event = self.await_event().await?;
-                    match event {
-                        FrontendEvent::Created(h, c, s) => {
-                            self.clients.push((h, c, s));
-                            break h;
-                        }
-                        _ => {
-                            self.handle_event(event);
-                            continue;
+                    if let Some(Ok(event)) = self.rx.next().await {
+                        match event {
+                            FrontendEvent::Created(h, c, s) => {
+                                self.clients.push((h, c, s));
+                                break h;
+                            }
+                            _ => {
+                                self.handle_event(event);
+                                continue;
+                            }
                         }
                     }
                 };
@@ -136,35 +129,36 @@ impl<'a> Cli<'a> {
                     FrontendRequest::UpdatePort(handle, port.unwrap_or(DEFAULT_PORT)),
                     FrontendRequest::UpdatePosition(handle, pos),
                 ] {
-                    self.send_request(request).await?;
+                    self.tx.request(request).await?;
                 }
                 self.update_client(handle).await?;
             }
             Command::Disconnect(id) => {
-                self.send_request(FrontendRequest::Delete(id)).await?;
+                self.tx.request(FrontendRequest::Delete(id)).await?;
                 loop {
-                    let event = self.await_event().await?;
-                    self.handle_event(event.clone());
-                    if let FrontendEvent::Deleted(_) = event {
-                        self.handle_event(event);
-                        break;
+                    if let Some(Ok(event)) = self.rx.next().await {
+                        self.handle_event(event.clone());
+                        if let FrontendEvent::Deleted(_) = event {
+                            self.handle_event(event);
+                            break;
+                        }
                     }
                 }
             }
             Command::Activate(id) => {
-                self.send_request(FrontendRequest::Activate(id, true))
-                    .await?;
+                self.tx.request(FrontendRequest::Activate(id, true)).await?;
                 self.update_client(id).await?;
             }
             Command::Deactivate(id) => {
-                self.send_request(FrontendRequest::Activate(id, false))
+                self.tx
+                    .request(FrontendRequest::Activate(id, false))
                     .await?;
                 self.update_client(id).await?;
             }
             Command::List => {
-                self.send_request(FrontendRequest::Enumerate()).await?;
-                loop {
-                    let event = self.await_event().await?;
+                self.tx.request(FrontendRequest::Enumerate()).await?;
+                while let Some(e) = self.rx.next().await {
+                    let event = e?;
                     self.handle_event(event.clone());
                     if let FrontendEvent::Enumerate(_) = event {
                         break;
@@ -173,12 +167,12 @@ impl<'a> Cli<'a> {
             }
             Command::SetHost(handle, host) => {
                 let request = FrontendRequest::UpdateHostname(handle, Some(host.clone()));
-                self.send_request(request).await?;
+                self.tx.request(request).await?;
                 self.update_client(handle).await?;
             }
             Command::SetPort(handle, port) => {
                 let request = FrontendRequest::UpdatePort(handle, port.unwrap_or(DEFAULT_PORT));
-                self.send_request(request).await?;
+                self.tx.request(request).await?;
                 self.update_client(handle).await?;
             }
             Command::Help => {
@@ -215,6 +209,7 @@ impl<'a> Cli<'a> {
 
     fn handle_event(&mut self, event: FrontendEvent) {
         match event {
+            FrontendEvent::Changed(h) => self.changed = Some(h),
             FrontendEvent::Created(h, c, s) => {
                 eprint!("client added ({h}): ");
                 print_config(&c);
@@ -291,23 +286,6 @@ impl<'a> Cli<'a> {
             eprintln!();
         }
     }
-
-    async fn send_request(&mut self, request: FrontendRequest) -> io::Result<()> {
-        let json = serde_json::to_string(&request).unwrap();
-        let bytes = json.as_bytes();
-        let len = bytes.len();
-        self.tx.write_u64(len as u64).await?;
-        self.tx.write_all(bytes).await?;
-        Ok(())
-    }
-
-    async fn await_event(&mut self) -> Result<FrontendEvent> {
-        let len = self.rx.read_u64().await?;
-        let mut buf = vec![0u8; len as usize];
-        self.rx.read_exact(&mut buf).await?;
-        let event: FrontendEvent = serde_json::from_slice(&buf)?;
-        Ok(event)
-    }
 }
 
 fn prompt() -> io::Result<()> {

lan-mouse-gtk/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "lan-mouse-gtk"
+description = "GTK4 / Libadwaita Frontend for lan-mouse"
+version = "0.1.0"
+edition = "2021"
+license = "GPL-3.0-or-later"
+repository = "https://github.com/feschber/lan-mouse"
+
+[dependencies]
+gtk = { package = "gtk4", version = "0.9.0", features = ["v4_2"] }
+adw = { package = "libadwaita", version = "0.7.0", features = ["v1_1"] }
+async-channel = { version = "2.1.1" }
+hostname = "0.4.0"
+log = "0.4.20"
+lan-mouse-ipc = { path = "../lan-mouse-ipc", version = "0.1.0" }
+
+[build-dependencies]
+glib-build-tools = { version = "0.20.0" }

lan-mouse-gtk/build.rs

@@ -0,0 +1,8 @@
+fn main() {
+    // composite_templates
+    glib_build_tools::compile_resources(
+        &["resources"],
+        "resources/resources.gresource.xml",
+        "lan-mouse.gresource",
+    );
+}

lan-mouse-gtk/src/client_object.rs

@@ -3,7 +3,7 @@ mod imp;
 use adw::subclass::prelude::*;
 use gtk::glib::{self, Object};
 
-use crate::client::{ClientConfig, ClientHandle, ClientState};
+use lan_mouse_ipc::{ClientConfig, ClientHandle, ClientState};
 
 glib::wrapper! {
     pub struct ClientObject(ObjectSubclass<imp::ClientObject>);

lan-mouse-gtk/src/client_object/imp.rs

@@ -5,7 +5,7 @@ use gtk::glib;
 use gtk::prelude::*;
 use gtk::subclass::prelude::*;
 
-use crate::client::ClientHandle;
+use lan_mouse_ipc::ClientHandle;
 
 use super::ClientData;
 

lan-mouse-gtk/src/client_row.rs

@@ -4,7 +4,7 @@ use adw::prelude::*;
 use adw::subclass::prelude::*;
 use gtk::glib::{self, Object};
 
-use crate::config::DEFAULT_PORT;
+use lan_mouse_ipc::DEFAULT_PORT;
 
 use super::ClientObject;
 

lan-mouse-gtk/src/lib.rs

@@ -2,16 +2,13 @@ mod client_object;
 mod client_row;
 mod window;
 
-use std::{
-    env,
-    io::{ErrorKind, Read},
-    process, str,
-};
+use std::{env, process, str};
 
-use crate::frontend::gtk::window::Window;
+use window::Window;
+
+use lan_mouse_ipc::{FrontendEvent, FrontendRequest};
 
 use adw::Application;
-use endi::{Endian, ReadBytes};
 use gtk::{
     gdk::Display, glib::clone, prelude::*, subclass::prelude::ObjectSubclassIsExt, IconTheme,
 };
@@ -19,8 +16,6 @@ use gtk::{gio, glib, prelude::ApplicationExt};
 
 use self::client_object::ClientObject;
 
-use super::FrontendEvent;
-
 pub fn run() -> glib::ExitCode {
     log::debug!("running gtk frontend");
     #[cfg(windows)]
@@ -65,15 +60,8 @@ fn load_icons() {
 
 fn build_ui(app: &Application) {
     log::debug!("connecting to lan-mouse-socket");
-    let mut rx = match super::wait_for_service() {
-        Ok(stream) => stream,
-        Err(e) => {
-            log::error!("could not connect to lan-mouse-socket: {e}");
-            process::exit(1);
-        }
-    };
-    let tx = match rx.try_clone() {
-        Ok(sock) => sock,
+    let (mut frontend_rx, frontend_tx) = match lan_mouse_ipc::connect() {
+        Ok(conn) => conn,
         Err(e) => {
             log::error!("{e}");
             process::exit(1);
@@ -84,35 +72,18 @@ fn build_ui(app: &Application) {
     let (sender, receiver) = async_channel::bounded(10);
 
     gio::spawn_blocking(move || {
-        match loop {
-            // read length
-            let len = match rx.read_u64(Endian::Big) {
-                Ok(l) => l,
-                Err(e) if e.kind() == ErrorKind::UnexpectedEof => break Ok(()),
-                Err(e) => break Err(e),
-            };
-
-            // read payload
-            let mut buf = vec![0u8; len as usize];
-            match rx.read_exact(&mut buf) {
-                Ok(_) => (),
-                Err(e) if e.kind() == ErrorKind::UnexpectedEof => break Ok(()),
-                Err(e) => break Err(e),
-            };
-
-            // parse json
-            let json = str::from_utf8(&buf).unwrap();
-            match serde_json::from_str(json) {
-                Ok(notify) => sender.send_blocking(notify).unwrap(),
-                Err(e) => log::error!("{e}"),
+        while let Some(e) = frontend_rx.next_event() {
+            match e {
+                Ok(e) => sender.send_blocking(e).unwrap(),
+                Err(e) => {
+                    log::error!("{e}");
+                    break;
+                }
             }
-        } {
-            Ok(()) => {}
-            Err(e) => log::error!("{e}"),
         }
     });
 
-    let window = Window::new(app, tx);
+    let window = Window::new(app, frontend_tx);
 
     glib::spawn_future_local(clone!(
         #[weak]
@@ -121,6 +92,9 @@ fn build_ui(app: &Application) {
             loop {
                 let notify = receiver.recv().await.unwrap_or_else(|_| process::exit(1));
                 match notify {
+                    FrontendEvent::Changed(handle) => {
+                        window.request(FrontendRequest::GetState(handle));
+                    }
                     FrontendEvent::Created(handle, client, state) => {
                         window.new_client(handle, client, state);
                     }

lan-mouse-gtk/src/window.rs

@@ -1,16 +1,7 @@
 mod imp;
 
-use std::io::Write;
-
-#[cfg(unix)]
-use std::os::unix::net::UnixStream;
-
-#[cfg(windows)]
-use std::net::TcpStream;
-
 use adw::prelude::*;
 use adw::subclass::prelude::*;
-use endi::{Endian, WriteBytes};
 use glib::{clone, Object};
 use gtk::{
     gio,
@@ -18,13 +9,12 @@ use gtk::{
     ListBox, NoSelection,
 };
 
-use crate::{
-    client::{ClientConfig, ClientHandle, ClientState, Position},
-    config::DEFAULT_PORT,
-    frontend::{gtk::client_object::ClientObject, FrontendRequest},
+use lan_mouse_ipc::{
+    ClientConfig, ClientHandle, ClientState, FrontendRequest, FrontendRequestWriter, Position,
+    DEFAULT_PORT,
 };
 
-use super::client_row::ClientRow;
+use super::{client_object::ClientObject, client_row::ClientRow};
 
 glib::wrapper! {
     pub struct Window(ObjectSubclass<imp::Window>)
@@ -34,13 +24,13 @@ glib::wrapper! {
 }
 
 impl Window {
-    pub(crate) fn new(
-        app: &adw::Application,
-        #[cfg(unix)] tx: UnixStream,
-        #[cfg(windows)] tx: TcpStream,
-    ) -> Self {
+    pub(crate) fn new(app: &adw::Application, conn: FrontendRequestWriter) -> Self {
         let window: Self = Object::builder().property("application", app).build();
-        window.imp().stream.borrow_mut().replace(tx);
+        window
+            .imp()
+            .frontend_request_writer
+            .borrow_mut()
+            .replace(conn);
         window
     }
 
@@ -257,7 +247,11 @@ impl Window {
     }
 
     pub fn request_client_state(&self, client: &ClientObject) {
-        self.request(FrontendRequest::GetState(client.handle()));
+        self.request_client_state_for(client.handle());
+    }
+
+    pub fn request_client_state_for(&self, handle: ClientHandle) {
+        self.request(FrontendRequest::GetState(handle));
     }
 
     pub fn request_client_create(&self) {
@@ -292,16 +286,10 @@ impl Window {
         self.request(FrontendRequest::Delete(client.handle()));
     }
 
-    pub fn request(&self, event: FrontendRequest) {
-        let json = serde_json::to_string(&event).unwrap();
-        log::debug!("requesting: {json}");
-        let mut stream = self.imp().stream.borrow_mut();
-        let stream = stream.as_mut().unwrap();
-        let bytes = json.as_bytes();
-        if let Err(e) = stream.write_u64(Endian::Big, bytes.len() as u64) {
-            log::error!("error sending message: {e}");
-        };
-        if let Err(e) = stream.write(bytes) {
+    pub fn request(&self, request: FrontendRequest) {
+        let mut requester = self.imp().frontend_request_writer.borrow_mut();
+        let requester = requester.as_mut().unwrap();
+        if let Err(e) = requester.request(request) {
             log::error!("error sending message: {e}");
         };
     }

lan-mouse-gtk/src/window/imp.rs

@@ -1,17 +1,12 @@
 use std::cell::{Cell, RefCell};
 
-#[cfg(windows)]
-use std::net::TcpStream;
-#[cfg(unix)]
-use std::os::unix::net::UnixStream;
-
 use adw::subclass::prelude::*;
 use adw::{prelude::*, ActionRow, PreferencesGroup, ToastOverlay};
 use glib::subclass::InitializingObject;
 use gtk::glib::clone;
 use gtk::{gdk, gio, glib, Button, CompositeTemplate, Entry, Label, ListBox};
 
-use crate::config::DEFAULT_PORT;
+use lan_mouse_ipc::{FrontendRequestWriter, DEFAULT_PORT};
 
 #[derive(CompositeTemplate, Default)]
 #[template(resource = "/de/feschber/LanMouse/window.ui")]
@@ -41,10 +36,7 @@ pub struct Window {
     #[template_child]
     pub input_capture_button: TemplateChild<Button>,
     pub clients: RefCell<Option<gio::ListStore>>,
-    #[cfg(unix)]
-    pub stream: RefCell<Option<UnixStream>>,
-    #[cfg(windows)]
-    pub stream: RefCell<Option<TcpStream>>,
+    pub frontend_request_writer: RefCell<Option<FrontendRequestWriter>>,
     pub port: Cell<u16>,
     pub capture_active: Cell<bool>,
     pub emulation_active: Cell<bool>,

lan-mouse-ipc/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "lan-mouse-ipc"
+description = "library for communication between lan-mouse service and frontends"
+version = "0.1.0"
+edition = "2021"
+license = "GPL-3.0-or-later"
+repository = "https://github.com/feschber/lan-mouse"
+
+[dependencies]
+futures = "0.3.30"
+log = "0.4.22"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0.107"
+thiserror = "1.0.63"
+tokio = { version = "1.32.0", features = ["net", "io-util", "time"] }
+tokio-stream = { version = "0.1.15", features = ["io-util"] }

lan-mouse-ipc/src/connect.rs

@@ -0,0 +1,89 @@
+use crate::{ConnectionError, FrontendEvent, FrontendRequest, IpcError};
+use std::{
+    cmp::min,
+    io::{self, prelude::*, BufReader, LineWriter, Lines},
+    thread,
+    time::Duration,
+};
+
+#[cfg(unix)]
+use std::os::unix::net::UnixStream;
+
+#[cfg(windows)]
+use std::net::TcpStream;
+
+pub struct FrontendEventReader {
+    #[cfg(unix)]
+    lines: Lines<BufReader<UnixStream>>,
+    #[cfg(windows)]
+    lines: Lines<BufReader<TcpStream>>,
+}
+
+pub struct FrontendRequestWriter {
+    #[cfg(unix)]
+    line_writer: LineWriter<UnixStream>,
+    #[cfg(windows)]
+    line_writer: LineWriter<TcpStream>,
+}
+
+impl FrontendEventReader {
+    pub fn next_event(&mut self) -> Option<Result<FrontendEvent, IpcError>> {
+        match self.lines.next()? {
+            Err(e) => Some(Err(e.into())),
+            Ok(l) => Some(serde_json::from_str(l.as_str()).map_err(|e| e.into())),
+        }
+    }
+}
+
+impl FrontendRequestWriter {
+    pub fn request(&mut self, request: FrontendRequest) -> Result<(), io::Error> {
+        let mut json = serde_json::to_string(&request).unwrap();
+        log::debug!("requesting: {json}");
+        json.push('\n');
+        self.line_writer.write_all(json.as_bytes())?;
+        Ok(())
+    }
+}
+
+pub fn connect() -> Result<(FrontendEventReader, FrontendRequestWriter), ConnectionError> {
+    let rx = wait_for_service()?;
+    let tx = rx.try_clone()?;
+    let buf_reader = BufReader::new(rx);
+    let lines = buf_reader.lines();
+    let line_writer = LineWriter::new(tx);
+    let reader = FrontendEventReader { lines };
+    let writer = FrontendRequestWriter { line_writer };
+    Ok((reader, writer))
+}
+
+/// wait for the lan-mouse socket to come online
+#[cfg(unix)]
+fn wait_for_service() -> Result<UnixStream, ConnectionError> {
+    let socket_path = crate::default_socket_path()?;
+    let mut duration = Duration::from_millis(10);
+    loop {
+        if let Ok(stream) = UnixStream::connect(&socket_path) {
+            break Ok(stream);
+        }
+        // a signaling mechanism or inotify could be used to
+        // improve this
+        thread::sleep(exponential_back_off(&mut duration));
+    }
+}
+
+#[cfg(windows)]
+fn wait_for_service() -> Result<TcpStream, ConnectionError> {
+    let mut duration = Duration::from_millis(10);
+    loop {
+        if let Ok(stream) = TcpStream::connect("127.0.0.1:5252") {
+            break Ok(stream);
+        }
+        thread::sleep(exponential_back_off(&mut duration));
+    }
+}
+
+fn exponential_back_off(duration: &mut Duration) -> Duration {
+    let new = duration.saturating_mul(2);
+    *duration = min(new, Duration::from_secs(1));
+    *duration
+}

lan-mouse-ipc/src/connect_async.rs

@@ -0,0 +1,104 @@
+use crate::{ConnectionError, FrontendEvent, FrontendRequest, IpcError};
+use std::{
+    cmp::min,
+    io,
+    task::{ready, Poll},
+    time::Duration,
+};
+
+use futures::{Stream, StreamExt};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf};
+use tokio_stream::wrappers::LinesStream;
+
+#[cfg(unix)]
+use tokio::net::UnixStream;
+
+#[cfg(windows)]
+use tokio::net::TcpStream;
+
+pub struct AsyncFrontendEventReader {
+    #[cfg(unix)]
+    lines_stream: LinesStream<BufReader<ReadHalf<UnixStream>>>,
+    #[cfg(windows)]
+    lines_stream: LinesStream<BufReader<ReadHalf<TcpStream>>>,
+}
+
+pub struct AsyncFrontendRequestWriter {
+    #[cfg(unix)]
+    tx: WriteHalf<UnixStream>,
+    #[cfg(windows)]
+    tx: WriteHalf<TcpStream>,
+}
+
+impl Stream for AsyncFrontendEventReader {
+    type Item = Result<FrontendEvent, IpcError>;
+
+    fn poll_next(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        let line = ready!(self.lines_stream.poll_next_unpin(cx));
+        let event = line.map(|l| {
+            l.map_err(Into::<IpcError>::into)
+                .and_then(|l| serde_json::from_str(l.as_str()).map_err(|e| e.into()))
+        });
+        Poll::Ready(event)
+    }
+}
+
+impl AsyncFrontendRequestWriter {
+    pub async fn request(&mut self, request: FrontendRequest) -> Result<(), io::Error> {
+        let mut json = serde_json::to_string(&request).unwrap();
+        log::debug!("requesting: {json}");
+        json.push('\n');
+        self.tx.write_all(json.as_bytes()).await?;
+        Ok(())
+    }
+}
+
+pub async fn connect_async(
+) -> Result<(AsyncFrontendEventReader, AsyncFrontendRequestWriter), ConnectionError> {
+    let stream = wait_for_service().await?;
+    #[cfg(unix)]
+    let (rx, tx): (ReadHalf<UnixStream>, WriteHalf<UnixStream>) = tokio::io::split(stream);
+    #[cfg(windows)]
+    let (rx, tx): (ReadHalf<TcpStream>, WriteHalf<TcpStream>) = tokio::io::split(stream);
+    let buf_reader = BufReader::new(rx);
+    let lines = buf_reader.lines();
+    let lines_stream = LinesStream::new(lines);
+    let reader = AsyncFrontendEventReader { lines_stream };
+    let writer = AsyncFrontendRequestWriter { tx };
+    Ok((reader, writer))
+}
+
+/// wait for the lan-mouse socket to come online
+#[cfg(unix)]
+async fn wait_for_service() -> Result<UnixStream, ConnectionError> {
+    let socket_path = crate::default_socket_path()?;
+    let mut duration = Duration::from_millis(10);
+    loop {
+        if let Ok(stream) = UnixStream::connect(&socket_path).await {
+            break Ok(stream);
+        }
+        // a signaling mechanism or inotify could be used to
+        // improve this
+        tokio::time::sleep(exponential_back_off(&mut duration)).await;
+    }
+}
+
+#[cfg(windows)]
+async fn wait_for_service() -> Result<TcpStream, ConnectionError> {
+    let mut duration = Duration::from_millis(10);
+    loop {
+        if let Ok(stream) = TcpStream::connect("127.0.0.1:5252").await {
+            break Ok(stream);
+        }
+        tokio::time::sleep(exponential_back_off(&mut duration)).await;
+    }
+}
+
+fn exponential_back_off(duration: &mut Duration) -> Duration {
+    let new = duration.saturating_mul(2);
+    *duration = min(new, Duration::from_secs(1));
+    *duration
+}

lan-mouse-ipc/src/lib.rs

@@ -0,0 +1,264 @@
+use std::{
+    collections::HashSet,
+    env::VarError,
+    fmt::Display,
+    io,
+    net::{IpAddr, SocketAddr},
+    str::FromStr,
+};
+use thiserror::Error;
+
+#[cfg(unix)]
+use std::{
+    env,
+    path::{Path, PathBuf},
+};
+
+use serde::{Deserialize, Serialize};
+
+mod connect;
+mod connect_async;
+mod listen;
+
+pub use connect::{connect, FrontendEventReader, FrontendRequestWriter};
+pub use connect_async::{connect_async, AsyncFrontendEventReader, AsyncFrontendRequestWriter};
+pub use listen::AsyncFrontendListener;
+
+#[derive(Debug, Error)]
+pub enum ConnectionError {
+    #[error(transparent)]
+    SocketPath(#[from] SocketPathError),
+    #[error(transparent)]
+    Io(#[from] io::Error),
+}
+
+#[derive(Debug, Error)]
+pub enum ListenerCreationError {
+    #[error("could not determine socket-path: `{0}`")]
+    SocketPath(#[from] SocketPathError),
+    #[error("service already running!")]
+    AlreadyRunning,
+    #[error("failed to bind lan-mouse socket: `{0}`")]
+    Bind(io::Error),
+}
+
+#[derive(Debug, Error)]
+pub enum IpcError {
+    #[error("io error occured: `{0}`")]
+    Io(#[from] io::Error),
+    #[error("invalid json: `{0}`")]
+    Json(#[from] serde_json::Error),
+    #[error(transparent)]
+    Connection(#[from] ConnectionError),
+    #[error(transparent)]
+    Listen(#[from] ListenerCreationError),
+}
+
+pub const DEFAULT_PORT: u16 = 4242;
+
+#[derive(Debug, Default, Eq, Hash, PartialEq, Clone, Copy, Serialize, Deserialize)]
+pub enum Position {
+    #[default]
+    Left,
+    Right,
+    Top,
+    Bottom,
+}
+
+#[derive(Debug, Error)]
+#[error("not a valid position: {pos}")]
+pub struct PositionParseError {
+    pos: String,
+}
+
+impl FromStr for Position {
+    type Err = PositionParseError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "left" => Ok(Self::Left),
+            "right" => Ok(Self::Right),
+            "top" => Ok(Self::Top),
+            "bottom" => Ok(Self::Bottom),
+            _ => Err(PositionParseError { pos: s.into() }),
+        }
+    }
+}
+
+impl Display for Position {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                Position::Left => "left",
+                Position::Right => "right",
+                Position::Top => "top",
+                Position::Bottom => "bottom",
+            }
+        )
+    }
+}
+
+impl TryFrom<&str> for Position {
+    type Error = ();
+
+    fn try_from(s: &str) -> Result<Self, Self::Error> {
+        match s {
+            "left" => Ok(Position::Left),
+            "right" => Ok(Position::Right),
+            "top" => Ok(Position::Top),
+            "bottom" => Ok(Position::Bottom),
+            _ => Err(()),
+        }
+    }
+}
+
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
+pub struct ClientConfig {
+    /// hostname of this client
+    pub hostname: Option<String>,
+    /// fix ips, determined by the user
+    pub fix_ips: Vec<IpAddr>,
+    /// both active_addr and addrs can be None / empty so port needs to be stored seperately
+    pub port: u16,
+    /// position of a client on screen
+    pub pos: Position,
+    /// enter hook
+    pub cmd: Option<String>,
+}
+
+impl Default for ClientConfig {
+    fn default() -> Self {
+        Self {
+            port: DEFAULT_PORT,
+            hostname: Default::default(),
+            fix_ips: Default::default(),
+            pos: Default::default(),
+            cmd: None,
+        }
+    }
+}
+
+pub type ClientHandle = u64;
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct ClientState {
+    /// events should be sent to and received from the client
+    pub active: bool,
+    /// `active` address of the client, used to send data to.
+    /// This should generally be the socket address where data
+    /// was last received from.
+    pub active_addr: Option<SocketAddr>,
+    /// tracks whether or not the client is responding to pings
+    pub alive: bool,
+    /// ips from dns
+    pub dns_ips: Vec<IpAddr>,
+    /// all ip addresses associated with a particular client
+    /// e.g. Laptops usually have at least an ethernet and a wifi port
+    /// which have different ip addresses
+    pub ips: HashSet<IpAddr>,
+    /// client has pressed keys
+    pub has_pressed_keys: bool,
+    /// dns resolving in progress
+    pub resolving: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum FrontendEvent {
+    /// client state has changed, new state must be requested via [`FrontendRequest::GetState`]
+    Changed(ClientHandle),
+    /// a client was created
+    Created(ClientHandle, ClientConfig, ClientState),
+    /// no such client
+    NoSuchClient(ClientHandle),
+    /// state changed
+    State(ClientHandle, ClientConfig, ClientState),
+    /// the client was deleted
+    Deleted(ClientHandle),
+    /// new port, reason of failure (if failed)
+    PortChanged(u16, Option<String>),
+    /// list of all clients, used for initial state synchronization
+    Enumerate(Vec<(ClientHandle, ClientConfig, ClientState)>),
+    /// an error occured
+    Error(String),
+    /// capture status
+    CaptureStatus(Status),
+    /// emulation status
+    EmulationStatus(Status),
+}
+
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
+pub enum FrontendRequest {
+    /// activate/deactivate client
+    Activate(ClientHandle, bool),
+    /// add a new client
+    Create,
+    /// change the listen port (recreate udp listener)
+    ChangePort(u16),
+    /// remove a client
+    Delete(ClientHandle),
+    /// request an enumeration of all clients
+    Enumerate(),
+    /// resolve dns
+    ResolveDns(ClientHandle),
+    /// update hostname
+    UpdateHostname(ClientHandle, Option<String>),
+    /// update port
+    UpdatePort(ClientHandle, u16),
+    /// update position
+    UpdatePosition(ClientHandle, Position),
+    /// update fix-ips
+    UpdateFixIps(ClientHandle, Vec<IpAddr>),
+    /// request the state of the given client
+    GetState(ClientHandle),
+    /// request reenabling input capture
+    EnableCapture,
+    /// request reenabling input emulation
+    EnableEmulation,
+    /// synchronize all state
+    Sync,
+}
+
+#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)]
+pub enum Status {
+    #[default]
+    Disabled,
+    Enabled,
+}
+
+impl From<Status> for bool {
+    fn from(status: Status) -> Self {
+        match status {
+            Status::Enabled => true,
+            Status::Disabled => false,
+        }
+    }
+}
+
+#[cfg(unix)]
+const LAN_MOUSE_SOCKET_NAME: &str = "lan-mouse-socket.sock";
+
+#[derive(Debug, Error)]
+pub enum SocketPathError {
+    #[error("could not determine $XDG_RUNTIME_DIR: `{0}`")]
+    XdgRuntimeDirNotFound(VarError),
+    #[error("could not determine $HOME: `{0}`")]
+    HomeDirNotFound(VarError),
+}
+
+#[cfg(all(unix, not(target_os = "macos")))]
+pub fn default_socket_path() -> Result<PathBuf, SocketPathError> {
+    let xdg_runtime_dir =
+        env::var("XDG_RUNTIME_DIR").map_err(SocketPathError::XdgRuntimeDirNotFound)?;
+    Ok(Path::new(xdg_runtime_dir.as_str()).join(LAN_MOUSE_SOCKET_NAME))
+}
+
+#[cfg(all(unix, target_os = "macos"))]
+pub fn default_socket_path() -> Result<PathBuf, SocketPathError> {
+    let home = env::var("HOME").map_err(SocketPathError::HomeDirNotFound)?;
+    Ok(Path::new(home.as_str())
+        .join("Library")
+        .join("Caches")
+        .join(LAN_MOUSE_SOCKET_NAME))
+}

lan-mouse-ipc/src/listen.rs

@@ -0,0 +1,147 @@
+use futures::{stream::SelectAll, Stream, StreamExt};
+#[cfg(unix)]
+use std::path::PathBuf;
+use std::{
+    io::ErrorKind,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf};
+use tokio_stream::wrappers::LinesStream;
+
+#[cfg(unix)]
+use tokio::net::UnixListener;
+#[cfg(unix)]
+use tokio::net::UnixStream;
+
+#[cfg(windows)]
+use tokio::net::TcpListener;
+#[cfg(windows)]
+use tokio::net::TcpStream;
+
+use crate::{FrontendEvent, FrontendRequest, IpcError, ListenerCreationError};
+
+pub struct AsyncFrontendListener {
+    #[cfg(windows)]
+    listener: TcpListener,
+    #[cfg(unix)]
+    listener: UnixListener,
+    #[cfg(unix)]
+    socket_path: PathBuf,
+    #[cfg(unix)]
+    line_streams: SelectAll<LinesStream<BufReader<ReadHalf<UnixStream>>>>,
+    #[cfg(windows)]
+    line_streams: SelectAll<LinesStream<BufReader<ReadHalf<TcpStream>>>>,
+    #[cfg(unix)]
+    tx_streams: Vec<WriteHalf<UnixStream>>,
+    #[cfg(windows)]
+    tx_streams: Vec<WriteHalf<TcpStream>>,
+}
+
+impl AsyncFrontendListener {
+    pub async fn new() -> Result<Self, ListenerCreationError> {
+        #[cfg(unix)]
+        let (socket_path, listener) = {
+            let socket_path = crate::default_socket_path()?;
+
+            log::debug!("remove socket: {:?}", socket_path);
+            if socket_path.exists() {
+                // try to connect to see if some other instance
+                // of lan-mouse is already running
+                match UnixStream::connect(&socket_path).await {
+                    // connected -> lan-mouse is already running
+                    Ok(_) => return Err(ListenerCreationError::AlreadyRunning),
+                    // lan-mouse is not running but a socket was left behind
+                    Err(e) => {
+                        log::debug!("{socket_path:?}: {e} - removing left behind socket");
+                        let _ = std::fs::remove_file(&socket_path);
+                    }
+                }
+            }
+            let listener = match UnixListener::bind(&socket_path) {
+                Ok(ls) => ls,
+                // some other lan-mouse instance has bound the socket in the meantime
+                Err(e) if e.kind() == ErrorKind::AddrInUse => {
+                    return Err(ListenerCreationError::AlreadyRunning)
+                }
+                Err(e) => return Err(ListenerCreationError::Bind(e)),
+            };
+            (socket_path, listener)
+        };
+
+        #[cfg(windows)]
+        let listener = match TcpListener::bind("127.0.0.1:5252").await {
+            Ok(ls) => ls,
+            // some other lan-mouse instance has bound the socket in the meantime
+            Err(e) if e.kind() == ErrorKind::AddrInUse => {
+                return Err(ListenerCreationError::AlreadyRunning)
+            }
+            Err(e) => return Err(ListenerCreationError::Bind(e)),
+        };
+
+        let adapter = Self {
+            listener,
+            #[cfg(unix)]
+            socket_path,
+            line_streams: SelectAll::new(),
+            tx_streams: vec![],
+        };
+
+        Ok(adapter)
+    }
+
+    pub async fn broadcast(&mut self, notify: FrontendEvent) {
+        // encode event
+        let mut json = serde_json::to_string(&notify).unwrap();
+        json.push('\n');
+
+        let mut keep = vec![];
+        // TODO do simultaneously
+        for tx in self.tx_streams.iter_mut() {
+            // write len + payload
+            if tx.write(json.as_bytes()).await.is_err() {
+                keep.push(false);
+                continue;
+            }
+            keep.push(true);
+        }
+
+        // could not find a better solution because async
+        let mut keep = keep.into_iter();
+        self.tx_streams.retain(|_| keep.next().unwrap());
+    }
+}
+
+#[cfg(unix)]
+impl Drop for AsyncFrontendListener {
+    fn drop(&mut self) {
+        log::debug!("remove socket: {:?}", self.socket_path);
+        let _ = std::fs::remove_file(&self.socket_path);
+    }
+}
+
+impl Stream for AsyncFrontendListener {
+    type Item = Result<FrontendRequest, IpcError>;
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if let Poll::Ready(Some(Ok(l))) = self.line_streams.poll_next_unpin(cx) {
+            let request = serde_json::from_str(l.as_str()).map_err(|e| e.into());
+            return Poll::Ready(Some(request));
+        }
+        let mut sync = false;
+        while let Poll::Ready(Ok((stream, _))) = self.listener.poll_accept(cx) {
+            let (rx, tx) = tokio::io::split(stream);
+            let buf_reader = BufReader::new(rx);
+            let lines = buf_reader.lines();
+            let lines = LinesStream::new(lines);
+            self.line_streams.push(lines);
+            self.tx_streams.push(tx);
+            sync = true;
+        }
+        if sync {
+            Poll::Ready(Some(Ok(FrontendRequest::Sync)))
+        } else {
+            Poll::Pending
+        }
+    }
+}

nix/default.nix

@@ -12,6 +12,7 @@ rustPlatform.buildRustPackage {
   version = version;
 
   nativeBuildInputs = with pkgs; [
+    git
     pkg-config
     cmake
     makeWrapper
@@ -23,9 +24,11 @@ rustPlatform.buildRustPackage {
     gtk4
     libadwaita
     xorg.libXtst
-  ] ++ lib.optionals stdenv.isDarwin [
-    darwin.apple_sdk_11_0.frameworks.CoreGraphics
-  ];
+  ] ++ lib.optionals stdenv.isDarwin
+  (with darwin.apple_sdk_11_0.frameworks; [
+    CoreGraphics
+    ApplicationServices
+  ]);
 
   src = builtins.path {
     name = pname;

src/capture_test.rs

@@ -2,21 +2,9 @@ use crate::config::Config;
 use futures::StreamExt;
 use input_capture::{self, CaptureError, CaptureEvent, InputCapture, InputCaptureError, Position};
 use input_event::{Event, KeyboardEvent};
-use tokio::task::LocalSet;
 
-pub fn run() -> anyhow::Result<()> {
+pub async fn run(config: Config) -> Result<(), InputCaptureError> {
     log::info!("running input capture test");
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_io()
-        .enable_time()
-        .build()?;
-
-    let config = Config::new()?;
-
-    Ok(runtime.block_on(LocalSet::new().run_until(input_capture_test(config)))?)
-}
-
-async fn input_capture_test(config: Config) -> Result<(), InputCaptureError> {
     log::info!("creating input capture");
     let backend = config.capture_backend.map(|b| b.into());
     loop {

src/client.rs

@@ -1,135 +1,8 @@
-use std::{
-    collections::HashSet,
-    fmt::Display,
-    net::{IpAddr, SocketAddr},
-    str::FromStr,
-};
+use std::net::SocketAddr;
 
-use serde::{Deserialize, Serialize};
 use slab::Slab;
-use thiserror::Error;
 
-use crate::config::DEFAULT_PORT;
-use input_capture;
-
-#[derive(Debug, Default, Eq, Hash, PartialEq, Clone, Copy, Serialize, Deserialize)]
-pub enum Position {
-    #[default]
-    Left,
-    Right,
-    Top,
-    Bottom,
-}
-
-impl From<Position> for input_capture::Position {
-    fn from(position: Position) -> input_capture::Position {
-        match position {
-            Position::Left => input_capture::Position::Left,
-            Position::Right => input_capture::Position::Right,
-            Position::Top => input_capture::Position::Top,
-            Position::Bottom => input_capture::Position::Bottom,
-        }
-    }
-}
-
-#[derive(Debug, Error)]
-#[error("not a valid position: {pos}")]
-pub struct PositionParseError {
-    pos: String,
-}
-
-impl FromStr for Position {
-    type Err = PositionParseError;
-
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
-        match s {
-            "left" => Ok(Self::Left),
-            "right" => Ok(Self::Right),
-            "top" => Ok(Self::Top),
-            "bottom" => Ok(Self::Bottom),
-            _ => Err(PositionParseError { pos: s.into() }),
-        }
-    }
-}
-
-impl Display for Position {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "{}",
-            match self {
-                Position::Left => "left",
-                Position::Right => "right",
-                Position::Top => "top",
-                Position::Bottom => "bottom",
-            }
-        )
-    }
-}
-
-impl TryFrom<&str> for Position {
-    type Error = ();
-
-    fn try_from(s: &str) -> Result<Self, Self::Error> {
-        match s {
-            "left" => Ok(Position::Left),
-            "right" => Ok(Position::Right),
-            "top" => Ok(Position::Top),
-            "bottom" => Ok(Position::Bottom),
-            _ => Err(()),
-        }
-    }
-}
-
-#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
-pub struct ClientConfig {
-    /// hostname of this client
-    pub hostname: Option<String>,
-    /// fix ips, determined by the user
-    pub fix_ips: Vec<IpAddr>,
-    /// both active_addr and addrs can be None / empty so port needs to be stored seperately
-    pub port: u16,
-    /// position of a client on screen
-    pub pos: Position,
-    /// enter hook
-    pub cmd: Option<String>,
-}
-
-impl Default for ClientConfig {
-    fn default() -> Self {
-        Self {
-            port: DEFAULT_PORT,
-            hostname: Default::default(),
-            fix_ips: Default::default(),
-            pos: Default::default(),
-            cmd: None,
-        }
-    }
-}
-
-pub type ClientHandle = u64;
-
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
-pub struct ClientState {
-    /// events should be sent to and received from the client
-    pub active: bool,
-    /// `active` address of the client, used to send data to.
-    /// This should generally be the socket address where data
-    /// was last received from.
-    pub active_addr: Option<SocketAddr>,
-    /// tracks whether or not the client is responding to pings
-    pub alive: bool,
-    /// ips from dns
-    pub dns_ips: Vec<IpAddr>,
-    /// all ip addresses associated with a particular client
-    /// e.g. Laptops usually have at least an ethernet and a wifi port
-    /// which have different ip addresses
-    pub ips: HashSet<IpAddr>,
-    /// client has pressed keys
-    pub has_pressed_keys: bool,
-    /// dns resolving in progress
-    pub resolving: bool,
-}
+use lan_mouse_ipc::{ClientConfig, ClientHandle, ClientState, Position};
 
 #[derive(Default)]
 pub struct ClientManager {

src/config.rs

@@ -1,22 +1,20 @@
-use anyhow::Result;
 use clap::{Parser, ValueEnum};
 use serde::{Deserialize, Serialize};
-use std::collections::HashSet;
-use std::env;
+use std::env::{self, VarError};
 use std::fmt::Display;
+use std::fs;
 use std::net::IpAddr;
-use std::{error::Error, fs};
+use std::{collections::HashSet, io};
+use thiserror::Error;
 use toml;
 
-use crate::client::Position;
+use lan_mouse_ipc::{Position, DEFAULT_PORT};
 
 use input_event::scancode::{
     self,
     Linux::{KeyLeftAlt, KeyLeftCtrl, KeyLeftMeta, KeyLeftShift},
 };
 
-pub const DEFAULT_PORT: u16 = 4242;
-
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ConfigToml {
     pub capture_backend: Option<CaptureBackend>,
@@ -42,7 +40,7 @@ pub struct TomlClient {
 }
 
 impl ConfigToml {
-    pub fn new(path: &str) -> Result<ConfigToml, Box<dyn Error>> {
+    pub fn new(path: &str) -> Result<ConfigToml, ConfigError> {
         let config = fs::read_to_string(path)?;
         log::info!("using config: \"{path}\"");
         Ok(toml::from_str::<_>(&config)?)
@@ -50,7 +48,7 @@ impl ConfigToml {
 }
 
 #[derive(Parser, Debug)]
-#[command(author, version, about, long_about = None)]
+#[command(author, version=env!("GIT_DESCRIBE"), about, long_about = None)]
 struct CliArgs {
     /// the listen port for lan-mouse
     #[arg(short, long)]
@@ -231,11 +229,21 @@ pub struct ConfigClient {
     pub enter_hook: Option<String>,
 }
 
+#[derive(Debug, Error)]
+pub enum ConfigError {
+    #[error(transparent)]
+    Toml(#[from] toml::de::Error),
+    #[error(transparent)]
+    Io(#[from] io::Error),
+    #[error(transparent)]
+    Var(#[from] VarError),
+}
+
 const DEFAULT_RELEASE_KEYS: [scancode::Linux; 4] =
     [KeyLeftCtrl, KeyLeftShift, KeyLeftMeta, KeyLeftAlt];
 
 impl Config {
-    pub fn new() -> Result<Self> {
+    pub fn new() -> Result<Self, ConfigError> {
         let args = CliArgs::parse();
         let config_file = "config.toml";
         #[cfg(unix)]

src/dns.rs

@@ -1,9 +1,10 @@
+use local_channel::mpsc::Receiver;
 use std::net::IpAddr;
-use tokio::sync::mpsc::Receiver;
 
 use hickory_resolver::{error::ResolveError, TokioAsyncResolver};
 
-use crate::{client::ClientHandle, server::Server};
+use crate::server::Server;
+use lan_mouse_ipc::ClientHandle;
 
 pub(crate) struct DnsResolver {
     resolver: TokioAsyncResolver,

src/emulation_test.rs

@@ -1,30 +1,19 @@
 use crate::config::Config;
-use anyhow::Result;
-use input_emulation::InputEmulation;
+use input_emulation::{InputEmulation, InputEmulationError};
 use input_event::{Event, PointerEvent};
 use std::f64::consts::PI;
 use std::time::{Duration, Instant};
-use tokio::task::LocalSet;
-
-pub fn run() -> Result<()> {
-    log::info!("running input emulation test");
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_io()
-        .enable_time()
-        .build()?;
-
-    let config = Config::new()?;
-
-    runtime.block_on(LocalSet::new().run_until(input_emulation_test(config)))
-}
 
 const FREQUENCY_HZ: f64 = 1.0;
 const RADIUS: f64 = 100.0;
 
-async fn input_emulation_test(config: Config) -> Result<()> {
+pub async fn run(config: Config) -> Result<(), InputEmulationError> {
+    log::info!("running input emulation test");
+
     let backend = config.emulation_backend.map(|b| b.into());
     let mut emulation = InputEmulation::new(backend).await?;
     emulation.create(0).await;
+
     let start = Instant::now();
     let mut offset = (0, 0);
     loop {

src/frontend.rs

@@ -1,307 +0,0 @@
-use anyhow::{anyhow, Result};
-use std::{cmp::min, io::ErrorKind, net::IpAddr, str, time::Duration};
-
-#[cfg(unix)]
-use std::{
-    env,
-    path::{Path, PathBuf},
-};
-
-use tokio::io::ReadHalf;
-use tokio::io::{AsyncReadExt, AsyncWriteExt, WriteHalf};
-
-#[cfg(unix)]
-use tokio::net::UnixListener;
-#[cfg(unix)]
-use tokio::net::UnixStream;
-
-#[cfg(windows)]
-use tokio::net::TcpListener;
-#[cfg(windows)]
-use tokio::net::TcpStream;
-
-use serde::{Deserialize, Serialize};
-
-use crate::{
-    client::{ClientConfig, ClientHandle, ClientState, Position},
-    config::{Config, Frontend},
-};
-
-/// cli frontend
-pub mod cli;
-
-/// gtk frontend
-#[cfg(feature = "gtk")]
-pub mod gtk;
-
-pub fn run_frontend(config: &Config) -> Result<()> {
-    match config.frontend {
-        #[cfg(feature = "gtk")]
-        Frontend::Gtk => {
-            gtk::run();
-        }
-        #[cfg(not(feature = "gtk"))]
-        Frontend::Gtk => panic!("gtk frontend requested but feature not enabled!"),
-        Frontend::Cli => {
-            cli::run()?;
-        }
-    };
-    Ok(())
-}
-
-fn exponential_back_off(duration: &mut Duration) -> &Duration {
-    let new = duration.saturating_mul(2);
-    *duration = min(new, Duration::from_secs(1));
-    duration
-}
-
-/// wait for the lan-mouse socket to come online
-#[cfg(unix)]
-pub fn wait_for_service() -> Result<std::os::unix::net::UnixStream> {
-    let socket_path = FrontendListener::socket_path()?;
-    let mut duration = Duration::from_millis(1);
-    loop {
-        use std::os::unix::net::UnixStream;
-        if let Ok(stream) = UnixStream::connect(&socket_path) {
-            break Ok(stream);
-        }
-        // a signaling mechanism or inotify could be used to
-        // improve this
-        std::thread::sleep(*exponential_back_off(&mut duration));
-    }
-}
-
-#[cfg(windows)]
-pub fn wait_for_service() -> Result<std::net::TcpStream> {
-    let mut duration = Duration::from_millis(1);
-    loop {
-        use std::net::TcpStream;
-        if let Ok(stream) = TcpStream::connect("127.0.0.1:5252") {
-            break Ok(stream);
-        }
-        std::thread::sleep(*exponential_back_off(&mut duration));
-    }
-}
-
-#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
-pub enum FrontendRequest {
-    /// activate/deactivate client
-    Activate(ClientHandle, bool),
-    /// add a new client
-    Create,
-    /// change the listen port (recreate udp listener)
-    ChangePort(u16),
-    /// remove a client
-    Delete(ClientHandle),
-    /// request an enumeration of all clients
-    Enumerate(),
-    /// resolve dns
-    ResolveDns(ClientHandle),
-    /// update hostname
-    UpdateHostname(ClientHandle, Option<String>),
-    /// update port
-    UpdatePort(ClientHandle, u16),
-    /// update position
-    UpdatePosition(ClientHandle, Position),
-    /// update fix-ips
-    UpdateFixIps(ClientHandle, Vec<IpAddr>),
-    /// request the state of the given client
-    GetState(ClientHandle),
-    /// request reenabling input capture
-    EnableCapture,
-    /// request reenabling input emulation
-    EnableEmulation,
-}
-
-#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)]
-pub enum Status {
-    #[default]
-    Disabled,
-    Enabled,
-}
-
-impl From<Status> for bool {
-    fn from(status: Status) -> Self {
-        match status {
-            Status::Enabled => true,
-            Status::Disabled => false,
-        }
-    }
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub enum FrontendEvent {
-    /// a client was created
-    Created(ClientHandle, ClientConfig, ClientState),
-    /// no such client
-    NoSuchClient(ClientHandle),
-    /// state changed
-    State(ClientHandle, ClientConfig, ClientState),
-    /// the client was deleted
-    Deleted(ClientHandle),
-    /// new port, reason of failure (if failed)
-    PortChanged(u16, Option<String>),
-    /// list of all clients, used for initial state synchronization
-    Enumerate(Vec<(ClientHandle, ClientConfig, ClientState)>),
-    /// an error occured
-    Error(String),
-    /// capture status
-    CaptureStatus(Status),
-    /// emulation status
-    EmulationStatus(Status),
-}
-
-pub struct FrontendListener {
-    #[cfg(windows)]
-    listener: TcpListener,
-    #[cfg(unix)]
-    listener: UnixListener,
-    #[cfg(unix)]
-    socket_path: PathBuf,
-    #[cfg(unix)]
-    tx_streams: Vec<WriteHalf<UnixStream>>,
-    #[cfg(windows)]
-    tx_streams: Vec<WriteHalf<TcpStream>>,
-}
-
-impl FrontendListener {
-    #[cfg(all(unix, not(target_os = "macos")))]
-    pub fn socket_path() -> Result<PathBuf> {
-        let xdg_runtime_dir = match env::var("XDG_RUNTIME_DIR") {
-            Ok(d) => d,
-            Err(e) => return Err(anyhow!("could not find XDG_RUNTIME_DIR: {e}")),
-        };
-        let xdg_runtime_dir = Path::new(xdg_runtime_dir.as_str());
-        Ok(xdg_runtime_dir.join("lan-mouse-socket.sock"))
-    }
-
-    #[cfg(all(unix, target_os = "macos"))]
-    pub fn socket_path() -> Result<PathBuf> {
-        let home = match env::var("HOME") {
-            Ok(d) => d,
-            Err(e) => return Err(anyhow!("could not find HOME: {e}")),
-        };
-        let home = Path::new(home.as_str());
-        let path = home
-            .join("Library")
-            .join("Caches")
-            .join("lan-mouse-socket.sock");
-        Ok(path)
-    }
-
-    pub async fn new() -> Option<Result<Self>> {
-        #[cfg(unix)]
-        let (socket_path, listener) = {
-            let socket_path = match Self::socket_path() {
-                Ok(path) => path,
-                Err(e) => return Some(Err(e)),
-            };
-
-            log::debug!("remove socket: {:?}", socket_path);
-            if socket_path.exists() {
-                // try to connect to see if some other instance
-                // of lan-mouse is already running
-                match UnixStream::connect(&socket_path).await {
-                    // connected -> lan-mouse is already running
-                    Ok(_) => return None,
-                    // lan-mouse is not running but a socket was left behind
-                    Err(e) => {
-                        log::debug!("{socket_path:?}: {e} - removing left behind socket");
-                        let _ = std::fs::remove_file(&socket_path);
-                    }
-                }
-            }
-            let listener = match UnixListener::bind(&socket_path) {
-                Ok(ls) => ls,
-                // some other lan-mouse instance has bound the socket in the meantime
-                Err(e) if e.kind() == ErrorKind::AddrInUse => return None,
-                Err(e) => return Some(Err(anyhow!("failed to bind lan-mouse-socket: {e}"))),
-            };
-            (socket_path, listener)
-        };
-
-        #[cfg(windows)]
-        let listener = match TcpListener::bind("127.0.0.1:5252").await {
-            Ok(ls) => ls,
-            // some other lan-mouse instance has bound the socket in the meantime
-            Err(e) if e.kind() == ErrorKind::AddrInUse => return None,
-            Err(e) => return Some(Err(anyhow!("failed to bind lan-mouse-socket: {e}"))),
-        };
-
-        let adapter = Self {
-            listener,
-            #[cfg(unix)]
-            socket_path,
-            tx_streams: vec![],
-        };
-
-        Some(Ok(adapter))
-    }
-
-    #[cfg(unix)]
-    pub async fn accept(&mut self) -> Result<ReadHalf<UnixStream>> {
-        let stream = self.listener.accept().await?.0;
-        let (rx, tx) = tokio::io::split(stream);
-        self.tx_streams.push(tx);
-        Ok(rx)
-    }
-
-    #[cfg(windows)]
-    pub async fn accept(&mut self) -> Result<ReadHalf<TcpStream>> {
-        let stream = self.listener.accept().await?.0;
-        let (rx, tx) = tokio::io::split(stream);
-        self.tx_streams.push(tx);
-        Ok(rx)
-    }
-
-    pub(crate) async fn broadcast(&mut self, notify: FrontendEvent) {
-        // encode event
-        let json = serde_json::to_string(&notify).unwrap();
-        let payload = json.as_bytes();
-        let len = payload.len().to_be_bytes();
-        log::debug!("broadcasting event to streams: {json}");
-        let mut keep = vec![];
-        // TODO do simultaneously
-        for tx in self.tx_streams.iter_mut() {
-            // write len + payload
-            if tx.write(&len).await.is_err() {
-                keep.push(false);
-                continue;
-            }
-            if tx.write(payload).await.is_err() {
-                keep.push(false);
-                continue;
-            }
-            keep.push(true);
-        }
-
-        // could not find a better solution because async
-        let mut keep = keep.into_iter();
-        self.tx_streams.retain(|_| keep.next().unwrap());
-    }
-}
-
-#[cfg(unix)]
-impl Drop for FrontendListener {
-    fn drop(&mut self) {
-        log::debug!("remove socket: {:?}", self.socket_path);
-        let _ = std::fs::remove_file(&self.socket_path);
-    }
-}
-
-#[cfg(unix)]
-pub async fn wait_for_request(stream: &mut ReadHalf<UnixStream>) -> Result<FrontendRequest> {
-    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])?)
-}
-
-#[cfg(windows)]
-pub async fn wait_for_request(stream: &mut ReadHalf<TcpStream>) -> Result<FrontendRequest> {
-    let len = stream.read_u64().await?;
-    let mut buf = [0u8; 256];
-    stream.read_exact(&mut buf[..len as usize]).await?;
-    Ok(serde_json::from_slice(&buf[..len as usize])?)
-}

src/lib.rs

@@ -5,4 +5,3 @@ pub mod server;
 
 pub mod capture_test;
 pub mod emulation_test;
-pub mod frontend;

src/main.rs

@@ -1,11 +1,37 @@
-use anyhow::Result;
-use std::process::{self, Child, Command};
-
 use env_logger::Env;
-use lan_mouse::{capture_test, config::Config, emulation_test, frontend, server::Server};
-
+use input_capture::InputCaptureError;
+use input_emulation::InputEmulationError;
+use lan_mouse::{
+    capture_test,
+    config::{Config, ConfigError, Frontend},
+    emulation_test,
+    server::{Server, ServiceError},
+};
+use lan_mouse_ipc::IpcError;
+use std::{
+    future::Future,
+    io,
+    process::{self, Child, Command},
+};
+use thiserror::Error;
 use tokio::task::LocalSet;
 
+#[derive(Debug, Error)]
+enum LanMouseError {
+    #[error(transparent)]
+    Service(#[from] ServiceError),
+    #[error(transparent)]
+    IpcError(#[from] IpcError),
+    #[error(transparent)]
+    Config(#[from] ConfigError),
+    #[error(transparent)]
+    Io(#[from] io::Error),
+    #[error(transparent)]
+    Capture(#[from] InputCaptureError),
+    #[error(transparent)]
+    Emulation(#[from] InputEmulationError),
+}
+
 pub fn main() {
     // init logging
     let env = Env::default().filter_or("LAN_MOUSE_LOG_LEVEL", "info");
@@ -17,32 +43,24 @@ pub fn main() {
     }
 }
 
-pub fn start_service() -> Result<Child> {
-    let child = Command::new(std::env::current_exe()?)
-        .args(std::env::args().skip(1))
-        .arg("--daemon")
-        .spawn()?;
-    Ok(child)
-}
-
-pub fn run() -> Result<()> {
+fn run() -> Result<(), LanMouseError> {
     // parse config file + cli args
     let config = Config::new()?;
     log::debug!("{config:?}");
     log::info!("release bind: {:?}", config.release_bind);
 
     if config.test_capture {
-        capture_test::run()?;
+        run_async(capture_test::run(config))?;
     } else if config.test_emulation {
-        emulation_test::run()?;
+        run_async(emulation_test::run(config))?;
     } else if config.daemon {
         // if daemon is specified we run the service
-        run_service(config)?;
+        run_async(run_service(config))?;
     } else {
         //  otherwise start the service as a child process and
         //  run a frontend
         let mut service = start_service()?;
-        frontend::run_frontend(&config)?;
+        run_frontend(&config)?;
         #[cfg(unix)]
         {
             // on unix we give the service a chance to terminate gracefully
@@ -55,10 +73,14 @@ pub fn run() -> Result<()> {
         service.kill()?;
     }
 
-    anyhow::Ok(())
+    Ok(())
 }
 
-fn run_service(config: Config) -> Result<()> {
+fn run_async<F, E>(f: F) -> Result<(), LanMouseError>
+where
+    F: Future<Output = Result<(), E>>,
+    LanMouseError: From<E>,
+{
     // create single threaded tokio runtime
     let runtime = tokio::runtime::Builder::new_current_thread()
         .enable_io()
@@ -66,15 +88,35 @@ fn run_service(config: Config) -> Result<()> {
         .build()?;
 
     // run async event loop
-    runtime.block_on(LocalSet::new().run_until(async {
-        // run main loop
-        log::info!("Press {:?} to release the mouse", config.release_bind);
+    Ok(runtime.block_on(LocalSet::new().run_until(f))?)
+}
 
-        let mut server = Server::new(config);
-        server.run().await?;
+fn start_service() -> Result<Child, io::Error> {
+    let child = Command::new(std::env::current_exe()?)
+        .args(std::env::args().skip(1))
+        .arg("--daemon")
+        .spawn()?;
+    Ok(child)
+}
 
-        log::debug!("service exiting");
-        anyhow::Ok(())
-    }))?;
+async fn run_service(config: Config) -> Result<(), ServiceError> {
+    log::info!("Press {:?} to release the mouse", config.release_bind);
+    Server::new(config).run().await?;
+    log::info!("service exited!");
+    Ok(())
+}
+
+fn run_frontend(config: &Config) -> Result<(), IpcError> {
+    match config.frontend {
+        #[cfg(feature = "gtk")]
+        Frontend::Gtk => {
+            lan_mouse_gtk::run();
+        }
+        #[cfg(not(feature = "gtk"))]
+        Frontend::Gtk => panic!("gtk frontend requested but feature not enabled!"),
+        Frontend::Cli => {
+            lan_mouse_cli::run()?;
+        }
+    };
     Ok(())
 }

src/server.rs

@@ -1,37 +1,27 @@
 use capture_task::CaptureRequest;
 use emulation_task::EmulationRequest;
+use futures::StreamExt;
+use hickory_resolver::error::ResolveError;
+use local_channel::mpsc::{channel, Sender};
 use log;
 use std::{
     cell::{Cell, RefCell},
     collections::{HashSet, VecDeque},
-    io::ErrorKind,
+    io,
     net::{IpAddr, SocketAddr},
     rc::Rc,
 };
-use tokio::{
-    io::ReadHalf,
-    join, signal,
-    sync::{
-        mpsc::{channel, Sender},
-        Notify,
-    },
-    task::JoinHandle,
-};
+use thiserror::Error;
+use tokio::{join, signal, sync::Notify};
 use tokio_util::sync::CancellationToken;
 
-use crate::{
-    client::{ClientConfig, ClientHandle, ClientManager, ClientState, Position},
-    config::Config,
-    dns::DnsResolver,
-    frontend::{self, FrontendEvent, FrontendListener, FrontendRequest, Status},
+use crate::{client::ClientManager, config::Config, dns::DnsResolver};
+
+use lan_mouse_ipc::{
+    AsyncFrontendListener, ClientConfig, ClientHandle, ClientState, FrontendEvent, FrontendRequest,
+    ListenerCreationError, Position, Status,
 };
 
-#[cfg(unix)]
-use tokio::net::UnixStream;
-
-#[cfg(windows)]
-use tokio::net::TcpStream;
-
 mod capture_task;
 mod emulation_task;
 mod network_task;
@@ -48,6 +38,16 @@ enum State {
     AwaitAck,
 }
 
+#[derive(Debug, Error)]
+pub enum ServiceError {
+    #[error(transparent)]
+    Dns(#[from] ResolveError),
+    #[error(transparent)]
+    Listen(#[from] ListenerCreationError),
+    #[error(transparent)]
+    Io(#[from] io::Error),
+}
+
 #[derive(Clone)]
 pub struct Server {
     active_client: Rc<Cell<Option<ClientHandle>>>,
@@ -58,7 +58,6 @@ pub struct Server {
     notifies: Rc<Notifies>,
     config: Rc<Config>,
     pending_frontend_events: Rc<RefCell<VecDeque<FrontendEvent>>>,
-    pending_dns_requests: Rc<RefCell<VecDeque<ClientHandle>>>,
     capture_status: Rc<Cell<Status>>,
     emulation_status: Rc<Cell<Status>>,
 }
@@ -70,7 +69,6 @@ struct Notifies {
     ping: Notify,
     port_changed: Notify,
     frontend_event_pending: Notify,
-    dns_request_pending: Notify,
     cancel: CancellationToken,
 }
 
@@ -114,40 +112,32 @@ impl Server {
             release_bind,
             notifies,
             pending_frontend_events: Rc::new(RefCell::new(VecDeque::new())),
-            pending_dns_requests: Rc::new(RefCell::new(VecDeque::new())),
             capture_status: Default::default(),
             emulation_status: Default::default(),
         }
     }
 
-    pub async fn run(&mut self) -> anyhow::Result<()> {
+    pub async fn run(&mut self) -> Result<(), ServiceError> {
         // create frontend communication adapter, exit if already running
-        let mut frontend = match FrontendListener::new().await {
-            Some(f) => f?,
-            None => {
+        let mut frontend = match AsyncFrontendListener::new().await {
+            Ok(f) => f,
+            Err(ListenerCreationError::AlreadyRunning) => {
                 log::info!("service already running, exiting");
                 return Ok(());
             }
+            e => e?,
         };
 
-        let (capture_tx, capture_rx) = channel(1); /* requests for input capture */
-        let (emulation_tx, emulation_rx) = channel(1); /* emulation requests */
-        let (udp_recv_tx, udp_recv_rx) = channel(1); /* udp receiver */
-        let (udp_send_tx, udp_send_rx) = channel(1); /* udp sender */
-        let (request_tx, mut request_rx) = channel(1); /* frontend requests */
-        let (dns_tx, dns_rx) = channel(1); /* dns requests */
+        let (capture_tx, capture_rx) = channel(); /* requests for input capture */
+        let (emulation_tx, emulation_rx) = channel(); /* emulation requests */
+        let (udp_recv_tx, udp_recv_rx) = channel(); /* udp receiver */
+        let (udp_send_tx, udp_send_rx) = channel(); /* udp sender */
+        let (dns_tx, dns_rx) = channel(); /* dns requests */
 
-        // udp task
         let network = network_task::new(self.clone(), udp_recv_tx.clone(), udp_send_rx).await?;
-
-        // input capture
         let capture = capture_task::new(self.clone(), capture_rx, udp_send_tx.clone());
-
-        // input emulation
         let emulation =
             emulation_task::new(self.clone(), emulation_rx, udp_recv_rx, udp_send_tx.clone());
-
-        // create dns resolver
         let resolver = DnsResolver::new(dns_rx)?;
         let dns_task = tokio::task::spawn_local(resolver.run(self.clone()));
 
@@ -160,30 +150,22 @@ impl Server {
         );
 
         for handle in self.active_clients() {
-            self.request_dns(handle);
+            dns_tx.send(handle).expect("channel closed");
         }
 
-        log::info!("running service");
-
-        let mut join_handles = vec![];
-
         loop {
             tokio::select! {
-                stream = frontend.accept() => {
-                    match stream {
-                        Ok(s) => join_handles.push(handle_frontend_stream(self.notifies.cancel.clone(), s, request_tx.clone())),
-                        Err(e) => log::warn!("error accepting frontend connection: {e}"),
+                request = frontend.next() => {
+                    let request = match request {
+                        Some(Ok(r)) => r,
+                        Some(Err(e)) => {
+                            log::error!("error receiving request: {e}");
+                            continue;
+                        }
+                        None => break,
                     };
-                    self.enumerate();
-                    self.notify_frontend(FrontendEvent::EmulationStatus(self.emulation_status.get()));
-                    self.notify_frontend(FrontendEvent::CaptureStatus(self.capture_status.get()));
-                    self.notify_frontend(FrontendEvent::PortChanged(self.port.get(), None));
-                }
-                request = request_rx.recv() => {
-                    let request = request.expect("channel closed");
-                    log::debug!("received frontend request: {request:?}");
-                    self.handle_request(&capture_tx.clone(), &emulation_tx.clone(), request).await;
-                    log::debug!("handled frontend request");
+                    log::debug!("handle frontend request: {request:?}");
+                    self.handle_request(&capture_tx.clone(), &emulation_tx.clone(), request, &dns_tx);
                 }
                 _ = self.notifies.frontend_event_pending.notified() => {
                     while let Some(event) = {
@@ -194,15 +176,6 @@ impl Server {
                         frontend.broadcast(event).await;
                     }
                 },
-                _ = self.notifies.dns_request_pending.notified() => {
-                    while let Some(request) = {
-                        /* need to drop borrow before next iteration! */
-                        let request = self.pending_dns_requests.borrow_mut().pop_front();
-                        request
-                    } {
-                        dns_tx.send(request).await.expect("channel closed");
-                    }
-                }
                 _ = self.cancelled() => break,
                 r = signal::ctrl_c() => {
                     r.expect("failed to wait for CTRL+C");
@@ -213,15 +186,7 @@ impl Server {
 
         log::info!("terminating service");
 
-        assert!(!capture_tx.is_closed());
-        assert!(!emulation_tx.is_closed());
-        assert!(!udp_recv_tx.is_closed());
-        assert!(!udp_send_tx.is_closed());
-        assert!(!request_tx.is_closed());
-        assert!(!dns_tx.is_closed());
-
         self.cancel();
-        futures::future::join_all(join_handles).await;
         let _ = join!(capture, dns_task, emulation, network, ping);
 
         Ok(())
@@ -245,6 +210,7 @@ impl Server {
     }
 
     fn notify_capture(&self) {
+        log::info!("received capture enable request");
         self.notifies.capture.notify_waiters()
     }
 
@@ -253,6 +219,7 @@ impl Server {
     }
 
     fn notify_emulation(&self) {
+        log::info!("received emulation enable request");
         self.notifies.emulation.notify_waiters()
     }
 
@@ -279,10 +246,7 @@ impl Server {
     }
 
     pub(crate) fn client_updated(&self, handle: ClientHandle) {
-        let state = self.client_manager.borrow().get(handle).cloned();
-        if let Some((config, state)) = state {
-            self.notify_frontend(FrontendEvent::State(handle, config, state));
-        }
+        self.notify_frontend(FrontendEvent::Changed(handle));
     }
 
     fn active_clients(&self) -> Vec<ClientHandle> {
@@ -294,55 +258,49 @@ impl Server {
             .collect()
     }
 
-    fn request_dns(&self, handle: ClientHandle) {
-        self.pending_dns_requests.borrow_mut().push_back(handle);
-        self.notifies.dns_request_pending.notify_one();
-    }
-
-    async fn handle_request(
+    fn handle_request(
         &self,
         capture: &Sender<CaptureRequest>,
         emulate: &Sender<EmulationRequest>,
         event: FrontendRequest,
+        dns: &Sender<ClientHandle>,
     ) -> bool {
         log::debug!("frontend: {event:?}");
         match event {
-            FrontendRequest::EnableCapture => {
-                log::info!("received capture enable request");
-                self.notify_capture();
-            }
-            FrontendRequest::EnableEmulation => {
-                log::info!("received emulation enable request");
-                self.notify_emulation();
-            }
+            FrontendRequest::EnableCapture => self.notify_capture(),
+            FrontendRequest::EnableEmulation => self.notify_emulation(),
             FrontendRequest::Create => {
-                let handle = self.add_client().await;
-                self.request_dns(handle);
+                self.add_client();
             }
             FrontendRequest::Activate(handle, active) => {
                 if active {
-                    self.activate_client(capture, emulate, handle).await;
+                    self.activate_client(capture, emulate, handle);
                 } else {
-                    self.deactivate_client(capture, emulate, handle).await;
+                    self.deactivate_client(capture, emulate, handle);
                 }
             }
             FrontendRequest::ChangePort(port) => self.request_port_change(port),
             FrontendRequest::Delete(handle) => {
-                self.remove_client(capture, emulate, handle).await;
+                self.remove_client(capture, emulate, handle);
                 self.notify_frontend(FrontendEvent::Deleted(handle));
             }
             FrontendRequest::Enumerate() => self.enumerate(),
             FrontendRequest::GetState(handle) => self.broadcast_client(handle),
-            FrontendRequest::UpdateFixIps(handle, fix_ips) => {
-                self.update_fix_ips(handle, fix_ips);
-                self.request_dns(handle);
+            FrontendRequest::UpdateFixIps(handle, fix_ips) => self.update_fix_ips(handle, fix_ips),
+            FrontendRequest::UpdateHostname(handle, host) => {
+                self.update_hostname(handle, host, dns)
             }
-            FrontendRequest::UpdateHostname(handle, host) => self.update_hostname(handle, host),
             FrontendRequest::UpdatePort(handle, port) => self.update_port(handle, port),
             FrontendRequest::UpdatePosition(handle, pos) => {
-                self.update_pos(handle, capture, emulate, pos).await;
+                self.update_pos(handle, capture, emulate, pos)
+            }
+            FrontendRequest::ResolveDns(handle) => dns.send(handle).expect("channel closed"),
+            FrontendRequest::Sync => {
+                self.enumerate();
+                self.notify_frontend(FrontendEvent::EmulationStatus(self.emulation_status.get()));
+                self.notify_frontend(FrontendEvent::CaptureStatus(self.capture_status.get()));
+                self.notify_frontend(FrontendEvent::PortChanged(self.port.get(), None));
             }
-            FrontendRequest::ResolveDns(handle) => self.request_dns(handle),
         };
         false
     }
@@ -357,7 +315,7 @@ impl Server {
         self.notify_frontend(FrontendEvent::Enumerate(clients));
     }
 
-    async fn add_client(&self) -> ClientHandle {
+    fn add_client(&self) -> ClientHandle {
         let handle = self.client_manager.borrow_mut().add_client();
         log::info!("added client {handle}");
         let (c, s) = self.client_manager.borrow().get(handle).unwrap().clone();
@@ -365,41 +323,40 @@ impl Server {
         handle
     }
 
-    async fn deactivate_client(
+    fn deactivate_client(
         &self,
         capture: &Sender<CaptureRequest>,
         emulate: &Sender<EmulationRequest>,
         handle: ClientHandle,
     ) {
-        log::debug!("deactivating client {handle}");
         match self.client_manager.borrow_mut().get_mut(handle) {
-            Some((_, s)) => s.active = false,
             None => return,
+            Some((_, s)) if !s.active => return,
+            Some((_, s)) => s.active = false,
         };
 
-        let _ = capture.send(CaptureRequest::Destroy(handle)).await;
-        let _ = emulate.send(EmulationRequest::Destroy(handle)).await;
-        log::debug!("deactivating client {handle} done");
+        let _ = capture.send(CaptureRequest::Destroy(handle));
+        let _ = emulate.send(EmulationRequest::Destroy(handle));
+        self.client_updated(handle);
+        log::info!("deactivated client {handle}");
     }
 
-    async fn activate_client(
+    fn activate_client(
         &self,
         capture: &Sender<CaptureRequest>,
         emulate: &Sender<EmulationRequest>,
         handle: ClientHandle,
     ) {
-        log::debug!("activating client");
         /* deactivate potential other client at this position */
         let pos = match self.client_manager.borrow().get(handle) {
-            Some((client, _)) => client.pos,
             None => return,
+            Some((_, s)) if s.active => return,
+            Some((client, _)) => client.pos,
         };
 
         let other = self.client_manager.borrow_mut().find_client(pos);
         if let Some(other) = other {
-            if other != handle {
-                self.deactivate_client(capture, emulate, other).await;
-            }
+            self.deactivate_client(capture, emulate, other);
         }
 
         /* activate the client */
@@ -410,14 +367,15 @@ impl Server {
         };
 
         /* notify emulation, capture and frontends */
-        let _ = capture
-            .send(CaptureRequest::Create(handle, pos.into()))
-            .await;
-        let _ = emulate.send(EmulationRequest::Create(handle)).await;
-        log::debug!("activating client {handle} done");
+        let _ = capture.send(CaptureRequest::Create(handle, to_capture_pos(pos)));
+        let _ = emulate.send(EmulationRequest::Create(handle));
+
+        self.client_updated(handle);
+
+        log::info!("activated client {handle} ({pos})");
     }
 
-    async fn remove_client(
+    fn remove_client(
         &self,
         capture: &Sender<CaptureRequest>,
         emulate: &Sender<EmulationRequest>,
@@ -433,8 +391,8 @@ impl Server {
         };
 
         if active {
-            let _ = capture.send(CaptureRequest::Destroy(handle)).await;
-            let _ = emulate.send(EmulationRequest::Destroy(handle)).await;
+            let _ = capture.send(CaptureRequest::Destroy(handle));
+            let _ = emulate.send(EmulationRequest::Destroy(handle));
         }
     }
 
@@ -449,6 +407,7 @@ impl Server {
             c.fix_ips = fix_ips;
         };
         self.update_ips(handle);
+        self.client_updated(handle);
     }
 
     pub(crate) fn update_dns_ips(&self, handle: ClientHandle, dns_ips: Vec<IpAddr>) {
@@ -456,6 +415,7 @@ impl Server {
             s.dns_ips = dns_ips;
         };
         self.update_ips(handle);
+        self.client_updated(handle);
     }
 
     fn update_ips(&self, handle: ClientHandle) {
@@ -469,7 +429,12 @@ impl Server {
         }
     }
 
-    fn update_hostname(&self, handle: ClientHandle, hostname: Option<String>) {
+    fn update_hostname(
+        &self,
+        handle: ClientHandle,
+        hostname: Option<String>,
+        dns: &Sender<ClientHandle>,
+    ) {
         let mut client_manager = self.client_manager.borrow_mut();
         let Some((c, s)) = client_manager.get_mut(handle) else {
             return;
@@ -478,10 +443,13 @@ impl Server {
         // hostname changed
         if c.hostname != hostname {
             c.hostname = hostname;
-            s.ips = HashSet::from_iter(c.fix_ips.iter().cloned());
             s.active_addr = None;
-            self.request_dns(handle);
+            s.dns_ips.clear();
+            drop(client_manager);
+            self.update_ips(handle);
+            dns.send(handle).expect("channel closed");
         }
+        self.client_updated(handle);
     }
 
     fn update_port(&self, handle: ClientHandle, port: u16) {
@@ -496,7 +464,7 @@ impl Server {
         }
     }
 
-    async fn update_pos(
+    fn update_pos(
         &self,
         handle: ClientHandle,
         capture: &Sender<CaptureRequest>,
@@ -510,20 +478,19 @@ impl Server {
             };
 
             let changed = c.pos != pos;
+            if changed {
+                log::info!("update pos {handle} {} -> {}", c.pos, pos);
+            }
             c.pos = pos;
             (changed, s.active)
         };
 
         // update state in event input emulator & input capture
         if changed {
+            self.deactivate_client(capture, emulate, handle);
             if active {
-                let _ = capture.send(CaptureRequest::Destroy(handle)).await;
-                let _ = emulate.send(EmulationRequest::Destroy(handle)).await;
+                self.activate_client(capture, emulate, handle);
             }
-            let _ = capture
-                .send(CaptureRequest::Create(handle, pos.into()))
-                .await;
-            let _ = emulate.send(EmulationRequest::Create(handle)).await;
         }
     }
 
@@ -556,7 +523,7 @@ impl Server {
         self.client_updated(handle);
     }
 
-    pub(crate) fn get_hostname(&self, handle: u64) -> Option<String> {
+    pub(crate) fn get_hostname(&self, handle: ClientHandle) -> Option<String> {
         self.client_manager
             .borrow_mut()
             .get_mut(handle)
@@ -572,12 +539,12 @@ impl Server {
         self.state.replace(state);
     }
 
-    fn set_active(&self, handle: Option<u64>) {
+    fn set_active(&self, handle: Option<ClientHandle>) {
         log::debug!("active client => {handle:?}");
         self.active_client.replace(handle);
     }
 
-    fn active_addr(&self, handle: u64) -> Option<SocketAddr> {
+    fn active_addr(&self, handle: ClientHandle) -> Option<SocketAddr> {
         self.client_manager
             .borrow()
             .get(handle)
@@ -585,41 +552,11 @@ impl Server {
     }
 }
 
-async fn listen_frontend(
-    request_tx: Sender<FrontendRequest>,
-    #[cfg(unix)] mut stream: ReadHalf<UnixStream>,
-    #[cfg(windows)] mut stream: ReadHalf<TcpStream>,
-) {
-    use std::io;
-    loop {
-        let request = frontend::wait_for_request(&mut stream).await;
-        match request {
-            Ok(request) => {
-                let _ = request_tx.send(request).await;
-            }
-            Err(e) => {
-                if let Some(e) = e.downcast_ref::<io::Error>() {
-                    if e.kind() == ErrorKind::UnexpectedEof {
-                        return;
-                    }
-                }
-                log::error!("error reading frontend event: {e}");
-                return;
-            }
-        }
+fn to_capture_pos(pos: Position) -> input_capture::Position {
+    match pos {
+        Position::Left => input_capture::Position::Left,
+        Position::Right => input_capture::Position::Right,
+        Position::Top => input_capture::Position::Top,
+        Position::Bottom => input_capture::Position::Bottom,
     }
 }
-
-fn handle_frontend_stream(
-    cancel: CancellationToken,
-    #[cfg(unix)] stream: ReadHalf<UnixStream>,
-    #[cfg(windows)] stream: ReadHalf<TcpStream>,
-    request_tx: Sender<FrontendRequest>,
-) -> JoinHandle<()> {
-    tokio::task::spawn_local(async move {
-        tokio::select! {
-            _ = listen_frontend(request_tx, stream) => {},
-            _ = cancel.cancelled() => {},
-        }
-    })
-}

src/server/capture_task.rs

@@ -1,18 +1,16 @@
 use futures::StreamExt;
 use lan_mouse_proto::ProtoEvent;
+use local_channel::mpsc::{Receiver, Sender};
 use std::net::SocketAddr;
 
-use tokio::{
-    process::Command,
-    sync::mpsc::{Receiver, Sender},
-    task::JoinHandle,
-};
+use tokio::{process::Command, task::JoinHandle};
 
 use input_capture::{
     self, CaptureError, CaptureEvent, CaptureHandle, InputCapture, InputCaptureError, Position,
 };
 
-use crate::{client::ClientHandle, frontend::Status, server::State};
+use crate::server::State;
+use lan_mouse_ipc::{ClientHandle, Status};
 
 use super::Server;
 
@@ -88,7 +86,7 @@ async fn do_capture(
         )
     });
     for (handle, pos) in clients {
-        capture.create(handle, pos.into()).await?;
+        capture.create(handle, to_capture_pos(pos)).await?;
     }
 
     loop {
@@ -161,7 +159,7 @@ async fn handle_capture_event(
             /* released capture */
             State::Receiving => ProtoEvent::Leave(0),
         };
-        sender_tx.send((event, addr)).await.expect("sender closed");
+        sender_tx.send((event, addr)).expect("sender closed");
     };
 
     Ok(())
@@ -197,3 +195,12 @@ fn spawn_hook_command(server: &Server, handle: ClientHandle) {
         }
     });
 }
+
+fn to_capture_pos(pos: lan_mouse_ipc::Position) -> input_capture::Position {
+    match pos {
+        lan_mouse_ipc::Position::Left => input_capture::Position::Left,
+        lan_mouse_ipc::Position::Right => input_capture::Position::Right,
+        lan_mouse_ipc::Position::Top => input_capture::Position::Top,
+        lan_mouse_ipc::Position::Bottom => input_capture::Position::Bottom,
+    }
+}

src/server/emulation_task.rs

@@ -1,17 +1,14 @@
+use local_channel::mpsc::{Receiver, Sender};
 use std::net::SocketAddr;
 
 use lan_mouse_proto::ProtoEvent;
-use tokio::{
-    sync::mpsc::{Receiver, Sender},
-    task::JoinHandle,
-};
+use tokio::task::JoinHandle;
 
-use crate::{
-    client::{ClientHandle, ClientManager},
-    frontend::Status,
-    server::State,
-};
+use lan_mouse_ipc::ClientHandle;
+
+use crate::{client::ClientManager, server::State};
 use input_emulation::{self, EmulationError, EmulationHandle, InputEmulation, InputEmulationError};
+use lan_mouse_ipc::Status;
 
 use super::{network_task::NetworkError, Server};
 
@@ -140,7 +137,7 @@ async fn handle_incoming_event(
     match (event, addr) {
         (ProtoEvent::Pong, _) => { /* ignore pong events */ }
         (ProtoEvent::Ping, addr) => {
-            let _ = sender_tx.send((ProtoEvent::Pong, addr)).await;
+            let _ = sender_tx.send((ProtoEvent::Pong, addr));
         }
         (ProtoEvent::Leave(_), _) => emulate.release_keys(handle).await?,
         (ProtoEvent::Ack(_), _) => server.set_state(State::Sending),
@@ -148,7 +145,6 @@ async fn handle_incoming_event(
             server.set_state(State::Receiving);
             sender_tx
                 .send((ProtoEvent::Ack(0), addr))
-                .await
                 .expect("no channel")
         }
         (ProtoEvent::Input(e), _) => {

src/server/network_task.rs

@@ -1,11 +1,8 @@
+use local_channel::mpsc::{Receiver, Sender};
 use std::{io, net::SocketAddr};
 
 use thiserror::Error;
-use tokio::{
-    net::UdpSocket,
-    sync::mpsc::{Receiver, Sender},
-    task::JoinHandle,
-};
+use tokio::{net::UdpSocket, task::JoinHandle};
 
 use super::Server;
 use lan_mouse_proto::{ProtoEvent, ProtocolError};
@@ -65,7 +62,7 @@ async fn udp_receiver(
 ) {
     loop {
         let event = receive_event(socket).await;
-        receiver_tx.send(event).await.expect("channel closed");
+        receiver_tx.send(event).expect("channel closed");
     }
 }
 

src/server/ping_task.rs

@@ -1,9 +1,10 @@
 use std::{net::SocketAddr, time::Duration};
 
 use lan_mouse_proto::ProtoEvent;
-use tokio::{sync::mpsc::Sender, task::JoinHandle};
+use local_channel::mpsc::Sender;
+use tokio::task::JoinHandle;
 
-use crate::client::ClientHandle;
+use lan_mouse_ipc::ClientHandle;
 
 use super::{capture_task::CaptureRequest, emulation_task::EmulationRequest, Server, State};
 
@@ -85,7 +86,7 @@ async fn ping_task(
 
             // ping clients
             for addr in ping_addrs {
-                if sender_ch.send((ProtoEvent::Ping, addr)).await.is_err() {
+                if sender_ch.send((ProtoEvent::Ping, addr)).is_err() {
                     break;
                 }
             }
@@ -122,14 +123,14 @@ async fn ping_task(
             if receiving {
                 for h in unresponsive_clients {
                     log::warn!("device not responding, releasing keys!");
-                    let _ = emulate_notify.send(EmulationRequest::ReleaseKeys(h)).await;
+                    let _ = emulate_notify.send(EmulationRequest::ReleaseKeys(h));
                 }
             } else {
                 // release pointer if the active client has not responded
                 if !unresponsive_clients.is_empty() {
                     log::warn!("client not responding, releasing pointer!");
                     server.state.replace(State::Receiving);
-                    let _ = capture_notify.send(CaptureRequest::Release).await;
+                    let _ = capture_notify.send(CaptureRequest::Release);
                 }
             }
         }