produce events in dummy capture-backend

* [WIP] MacOS inputcapture

---------

Co-authored-by: Ferdinand Schober <ferdinand.schober@fau.de>
Co-authored-by: Ferdinand Schober <ferdinandschober20@gmail.com>

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",
@@ -1324,6 +1329,7 @@ dependencies = [
  "lan-mouse-proto",
  "libadwaita",
  "libc",
+ "local-channel",
  "log",
  "serde",
  "serde_json",
@@ -1403,6 +1409,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"

Cargo.toml

@@ -50,6 +50,7 @@ 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"

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

@@ -217,7 +217,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);
+    }
+}

nix/default.nix

@@ -23,9 +23,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/dns.rs

@@ -1,5 +1,5 @@
+use local_channel::mpsc::Receiver;
 use std::net::IpAddr;
-use tokio::sync::mpsc::Receiver;
 
 use hickory_resolver::{error::ResolveError, TokioAsyncResolver};
 

src/server.rs

@@ -1,5 +1,6 @@
 use capture_task::CaptureRequest;
 use emulation_task::EmulationRequest;
+use local_channel::mpsc::{channel, Sender};
 use log;
 use std::{
     cell::{Cell, RefCell},
@@ -8,15 +9,7 @@ use std::{
     net::{IpAddr, SocketAddr},
     rc::Rc,
 };
-use tokio::{
-    io::ReadHalf,
-    join, signal,
-    sync::{
-        mpsc::{channel, Sender},
-        Notify,
-    },
-    task::JoinHandle,
-};
+use tokio::{io::ReadHalf, join, signal, sync::Notify, task::JoinHandle};
 use tokio_util::sync::CancellationToken;
 
 use crate::{
@@ -130,12 +123,12 @@ impl Server {
             }
         };
 
-        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 (request_tx, mut request_rx) = channel(); /* frontend requests */
+        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?;
@@ -200,7 +193,7 @@ impl Server {
                         let request = self.pending_dns_requests.borrow_mut().pop_front();
                         request
                     } {
-                        dns_tx.send(request).await.expect("channel closed");
+                        dns_tx.send(request).expect("channel closed");
                     }
                 }
                 _ = self.cancelled() => break,
@@ -213,13 +206,6 @@ 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);
@@ -377,8 +363,8 @@ impl Server {
             None => return,
         };
 
-        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));
         log::debug!("deactivating client {handle} done");
     }
 
@@ -410,10 +396,8 @@ impl Server {
         };
 
         /* notify emulation, capture and frontends */
-        let _ = capture
-            .send(CaptureRequest::Create(handle, pos.into()))
-            .await;
-        let _ = emulate.send(EmulationRequest::Create(handle)).await;
+        let _ = capture.send(CaptureRequest::Create(handle, pos.into()));
+        let _ = emulate.send(EmulationRequest::Create(handle));
         log::debug!("activating client {handle} done");
     }
 
@@ -433,8 +417,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));
         }
     }
 
@@ -517,13 +501,11 @@ impl Server {
         // update state in event input emulator & input capture
         if changed {
             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));
             }
-            let _ = capture
-                .send(CaptureRequest::Create(handle, pos.into()))
-                .await;
-            let _ = emulate.send(EmulationRequest::Create(handle)).await;
+            let _ = capture.send(CaptureRequest::Create(handle, pos.into()));
+            let _ = emulate.send(EmulationRequest::Create(handle));
         }
     }
 
@@ -595,7 +577,7 @@ async fn listen_frontend(
         let request = frontend::wait_for_request(&mut stream).await;
         match request {
             Ok(request) => {
-                let _ = request_tx.send(request).await;
+                let _ = request_tx.send(request);
             }
             Err(e) => {
                 if let Some(e) = e.downcast_ref::<io::Error>() {

src/server/capture_task.rs

@@ -1,12 +1,9 @@
 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,
@@ -161,7 +158,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(())

src/server/emulation_task.rs

@@ -1,10 +1,8 @@
+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},
@@ -140,7 +138,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 +146,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,7 +1,8 @@
 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;
 
@@ -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);
                 }
             }
         }