remove dispatch workaround

Cargo.lock

@@ -1226,6 +1226,7 @@ dependencies = [
  "input-event",
  "keycode",
  "log",
+ "once_cell",
  "reis",
  "thiserror",
  "tokio",

input-emulation/Cargo.toml

@@ -14,6 +14,7 @@ log = "0.4.22"
 input-event = { path = "../input-event", version = "0.1.0" }
 thiserror = "1.0.61"
 tokio = { version = "1.32.0", features = ["io-util", "io-std", "macros", "net", "process", "rt", "sync", "signal"] }
+once_cell = "1.19.0"
 
 [target.'cfg(all(unix, not(target_os="macos")))'.dependencies]
 wayland-client = { version="0.31.1", optional = true }

input-emulation/src/error.rs

@@ -8,6 +8,9 @@ use wayland_client::{
     ConnectError, DispatchError,
 };
 
+#[cfg(all(unix, feature = "libei", not(target_os = "macos")))]
+use reis::tokio::HandshakeError;
+
 #[derive(Debug, Error)]
 pub enum EmulationCreationError {
     #[cfg(all(unix, feature = "wayland", not(target_os = "macos")))]
@@ -104,6 +107,7 @@ impl Display for WlrootsEmulationCreationError {
 pub enum LibeiEmulationCreationError {
     Ashpd(#[from] ashpd::Error),
     Io(#[from] std::io::Error),
+    Handshake(#[from] HandshakeError),
 }
 
 #[cfg(all(unix, feature = "libei", not(target_os = "macos")))]
@@ -112,6 +116,7 @@ impl Display for LibeiEmulationCreationError {
         match self {
             LibeiEmulationCreationError::Ashpd(e) => write!(f, "xdg-desktop-portal: {e}"),
             LibeiEmulationCreationError::Io(e) => write!(f, "io error: {e}"),
+            LibeiEmulationCreationError::Handshake(e) => write!(f, "error in libei handshake: {e}"),
         }
     }
 }

input-emulation/src/lib.rs

@@ -1,5 +1,5 @@
 use async_trait::async_trait;
-use std::{fmt::Display, future};
+use std::fmt::Display;
 
 use input_event::Event;
 
@@ -73,11 +73,6 @@ pub trait InputEmulation: Send {
     async fn consume(&mut self, event: Event, handle: EmulationHandle);
     async fn create(&mut self, handle: EmulationHandle);
     async fn destroy(&mut self, handle: EmulationHandle);
-    /// this function is waited on continuously and can be used to handle events
-    async fn dispatch(&mut self) -> Result<()> {
-        let _: () = future::pending().await;
-        Ok(())
-    }
 }
 
 pub async fn create_backend(

input-emulation/src/libei.rs

@@ -1,10 +1,17 @@
 use anyhow::{anyhow, Result};
+use futures::StreamExt;
+use once_cell::sync::Lazy;
 use std::{
     collections::HashMap,
     io,
     os::{fd::OwnedFd, unix::net::UnixStream},
+    sync::{
+        atomic::{AtomicU32, Ordering},
+        Arc, RwLock,
+    },
     time::{SystemTime, UNIX_EPOCH},
 };
+use tokio::task::JoinHandle;
 
 use ashpd::{
     desktop::{
@@ -14,34 +21,49 @@ use ashpd::{
     WindowIdentifier,
 };
 use async_trait::async_trait;
-use futures::StreamExt;
 
 use reis::{
-    ei::{self, button::ButtonState, handshake::ContextType, keyboard::KeyState},
+    ei::{
-    tokio::EiEventStream,
+        self, button::ButtonState, handshake::ContextType, keyboard::KeyState, Button, Keyboard,
-    PendingRequestResult,
+        Pointer, Scroll,
+    },
+    event::{DeviceCapability, DeviceEvent, EiEvent, SeatEvent},
+    tokio::{ei_handshake, EiConvertEventStream, EiEventStream},
 };
 
 use input_event::{Event, KeyboardEvent, PointerEvent};
 
 use super::{error::LibeiEmulationCreationError, EmulationHandle, InputEmulation};
 
+static INTERFACES: Lazy<HashMap<&'static str, u32>> = Lazy::new(|| {
+    let mut m = HashMap::new();
+    m.insert("ei_connection", 1);
+    m.insert("ei_callback", 1);
+    m.insert("ei_pingpong", 1);
+    m.insert("ei_seat", 1);
+    m.insert("ei_device", 2);
+    m.insert("ei_pointer", 1);
+    m.insert("ei_pointer_absolute", 1);
+    m.insert("ei_scroll", 1);
+    m.insert("ei_button", 1);
+    m.insert("ei_keyboard", 1);
+    m.insert("ei_touchscreen", 1);
+    m
+});
+
+#[derive(Clone, Default)]
+struct Devices {
+    pointer: Arc<RwLock<Option<(ei::Device, ei::Pointer)>>>,
+    scroll: Arc<RwLock<Option<(ei::Device, ei::Scroll)>>>,
+    button: Arc<RwLock<Option<(ei::Device, ei::Button)>>>,
+    keyboard: Arc<RwLock<Option<(ei::Device, ei::Keyboard)>>>,
+}
+
 pub struct LibeiEmulation {
-    handshake: bool,
     context: ei::Context,
-    events: EiEventStream,
+    devices: Devices,
-    pointer: Option<(ei::Device, ei::Pointer)>,
+    serial: AtomicU32,
-    has_pointer: bool,
+    ei_task: JoinHandle<Result<()>>,
-    scroll: Option<(ei::Device, ei::Scroll)>,
-    has_scroll: bool,
-    button: Option<(ei::Device, ei::Button)>,
-    has_button: bool,
-    keyboard: Option<(ei::Device, ei::Keyboard)>,
-    has_keyboard: bool,
-    capabilities: HashMap<String, u64>,
-    capability_mask: u64,
-    sequence: u32,
-    serial: u32,
 }
 
 async fn get_ei_fd() -> Result<OwnedFd, ashpd::Error> {
@@ -77,37 +99,44 @@ async fn get_ei_fd() -> Result<OwnedFd, ashpd::Error> {
 
 impl LibeiEmulation {
     pub async fn new() -> Result<Self, LibeiEmulationCreationError> {
-        // fd is owned by the message, so we need to dup it
         let eifd = get_ei_fd().await?;
         let stream = UnixStream::from(eifd);
-        // let stream = UnixStream::connect("/run/user/1000/eis-0")?;
         stream.set_nonblocking(true)?;
         let context = ei::Context::new(stream)?;
         context.flush().map_err(|e| io::Error::new(e.kind(), e))?;
-        let events = EiEventStream::new(context.clone())?;
+        let mut events = EiEventStream::new(context.clone())?;
+        let handshake = ei_handshake(
+            &mut events,
+            "de.feschber.LanMouse",
+            ContextType::Sender,
+            &INTERFACES,
+        )
+        .await?;
+        let events = EiConvertEventStream::new(events, handshake.serial);
+        let devices = Devices::default();
+        let ei_task =
+            tokio::task::spawn_local(ei_event_handler(events, context.clone(), devices.clone()));
+
+        let serial = AtomicU32::new(handshake.serial);
+
         Ok(Self {
-            handshake: false,
+            serial,
             context,
-            events,
+            ei_task,
-            pointer: None,
+            devices,
-            button: None,
-            scroll: None,
-            keyboard: None,
-            has_pointer: false,
-            has_button: false,
-            has_scroll: false,
-            has_keyboard: false,
-            capabilities: HashMap::new(),
-            capability_mask: 0,
-            sequence: 0,
-            serial: 0,
         })
     }
 }
 
+impl Drop for LibeiEmulation {
+    fn drop(&mut self) {
+        self.ei_task.abort();
+    }
+}
+
 #[async_trait]
 impl InputEmulation for LibeiEmulation {
-    async fn consume(&mut self, event: Event, _client_handle: EmulationHandle) {
+    async fn consume(&mut self, event: Event, _handle: EmulationHandle) {
         let now = SystemTime::now()
             .duration_since(UNIX_EPOCH)
             .unwrap()
@@ -119,12 +148,10 @@ impl InputEmulation for LibeiEmulation {
                     relative_x,
                     relative_y,
                 } => {
-                    if !self.has_pointer {
+                    let pointer_device = self.devices.pointer.read().unwrap();
-                        return;
+                    if let Some((d, p)) = pointer_device.as_ref() {
-                    }
-                    if let Some((d, p)) = self.pointer.as_mut() {
                         p.motion_relative(relative_x as f32, relative_y as f32);
-                        d.frame(self.serial, now);
+                        d.frame(self.serial.load(Ordering::SeqCst), now);
                     }
                 }
                 PointerEvent::Button {
@@ -132,10 +159,8 @@ impl InputEmulation for LibeiEmulation {
                     button,
                     state,
                 } => {
-                    if !self.has_button {
+                    let button_device = self.devices.button.read().unwrap();
-                        return;
+                    if let Some((d, b)) = button_device.as_ref() {
-                    }
-                    if let Some((d, b)) = self.button.as_mut() {
                         b.button(
                             button,
                             match state {
@@ -143,7 +168,7 @@ impl InputEmulation for LibeiEmulation {
                                 _ => ButtonState::Press,
                             },
                         );
-                        d.frame(self.serial, now);
+                        d.frame(self.serial.load(Ordering::SeqCst), now);
                     }
                 }
                 PointerEvent::Axis {
@@ -151,27 +176,23 @@ impl InputEmulation for LibeiEmulation {
                     axis,
                     value,
                 } => {
-                    if !self.has_scroll {
+                    let scroll_device = self.devices.scroll.read().unwrap();
-                        return;
+                    if let Some((d, s)) = scroll_device.as_ref() {
-                    }
-                    if let Some((d, s)) = self.scroll.as_mut() {
                         match axis {
                             0 => s.scroll(0., value as f32),
                             _ => s.scroll(value as f32, 0.),
                         }
-                        d.frame(self.serial, now);
+                        d.frame(self.serial.load(Ordering::SeqCst), now);
                     }
                 }
                 PointerEvent::AxisDiscrete120 { axis, value } => {
-                    if !self.has_scroll {
+                    let scroll_device = self.devices.scroll.read().unwrap();
-                        return;
+                    if let Some((d, s)) = scroll_device.as_ref() {
-                    }
-                    if let Some((d, s)) = self.scroll.as_mut() {
                         match axis {
                             0 => s.scroll_discrete(0, value),
                             _ => s.scroll_discrete(value, 0),
                         }
-                        d.frame(self.serial, now);
+                        d.frame(self.serial.load(Ordering::SeqCst), now);
                     }
                 }
                 PointerEvent::Frame {} => {}
@@ -182,10 +203,8 @@ impl InputEmulation for LibeiEmulation {
                     key,
                     state,
                 } => {
-                    if !self.has_keyboard {
+                    let keyboard_device = self.devices.keyboard.read().unwrap();
-                        return;
+                    if let Some((d, k)) = keyboard_device.as_ref() {
-                    }
-                    if let Some((d, k)) = &mut self.keyboard {
                         k.key(
                             key,
                             match state {
@@ -193,7 +212,7 @@ impl InputEmulation for LibeiEmulation {
                                 _ => KeyState::Press,
                             },
                         );
-                        d.frame(self.serial, now);
+                        d.frame(self.serial.load(Ordering::SeqCst), now);
                     }
                 }
                 KeyboardEvent::Modifiers { .. } => {}
@@ -203,195 +222,111 @@ impl InputEmulation for LibeiEmulation {
         self.context.flush().unwrap();
     }
 
-    async fn dispatch(&mut self) -> Result<()> {
-        let event = match self.events.next().await {
-            Some(e) => e?,
-            None => return Err(anyhow!("libei connection lost")),
-        };
-        let event = match event {
-            PendingRequestResult::Request(result) => result,
-            PendingRequestResult::ParseError(e) => {
-                return Err(anyhow!("libei protocol violation: {e}"))
-            }
-            PendingRequestResult::InvalidObject(e) => return Err(anyhow!("invalid object {e}")),
-        };
-        match event {
-            ei::Event::Handshake(handshake, request) => match request {
-                ei::handshake::Event::HandshakeVersion { version } => {
-                    if self.handshake {
-                        return Ok(());
-                    }
-                    log::info!("libei version {}", version);
-                    // sender means we are sending events _to_ the eis server
-                    handshake.handshake_version(version); // FIXME
-                    handshake.context_type(ContextType::Sender);
-                    handshake.name("ei-demo-client");
-                    handshake.interface_version("ei_connection", 1);
-                    handshake.interface_version("ei_callback", 1);
-                    handshake.interface_version("ei_pingpong", 1);
-                    handshake.interface_version("ei_seat", 1);
-                    handshake.interface_version("ei_device", 2);
-                    handshake.interface_version("ei_pointer", 1);
-                    handshake.interface_version("ei_pointer_absolute", 1);
-                    handshake.interface_version("ei_scroll", 1);
-                    handshake.interface_version("ei_button", 1);
-                    handshake.interface_version("ei_keyboard", 1);
-                    handshake.interface_version("ei_touchscreen", 1);
-                    handshake.finish();
-                    self.handshake = true;
-                }
-                ei::handshake::Event::InterfaceVersion { name, version } => {
-                    log::debug!("handshake: Interface {name} @ {version}");
-                }
-                ei::handshake::Event::Connection { serial, connection } => {
-                    connection.sync(1);
-                    self.serial = serial;
-                }
-                _ => unreachable!(),
-            },
-            ei::Event::Connection(_connection, request) => match request {
-                ei::connection::Event::Seat { seat } => {
-                    log::debug!("connected to seat: {seat:?}");
-                }
-                ei::connection::Event::Ping { ping } => {
-                    ping.done(0);
-                }
-                ei::connection::Event::Disconnected {
-                    last_serial: _,
-                    reason,
-                    explanation,
-                } => {
-                    log::debug!("ei - disconnected: reason: {reason:?}: {explanation}")
-                }
-                ei::connection::Event::InvalidObject {
-                    last_serial,
-                    invalid_id,
-                } => {
-                    return Err(anyhow!(
-                        "invalid object: id: {invalid_id}, serial: {last_serial}"
-                    ));
-                }
-                _ => unreachable!(),
-            },
-            ei::Event::Device(device, request) => match request {
-                ei::device::Event::Destroyed { serial } => {
-                    log::debug!("device destroyed: {device:?} - serial: {serial}")
-                }
-                ei::device::Event::Name { name } => {
-                    log::debug!("device name: {name}")
-                }
-                ei::device::Event::DeviceType { device_type } => {
-                    log::debug!("device type: {device_type:?}")
-                }
-                ei::device::Event::Dimensions { width, height } => {
-                    log::debug!("device dimensions: {width}x{height}")
-                }
-                ei::device::Event::Region {
-                    offset_x,
-                    offset_y,
-                    width,
-                    hight,
-                    scale,
-                } => log::debug!(
-                    "device region: {width}x{hight} @ ({offset_x},{offset_y}), scale: {scale}"
-                ),
-                ei::device::Event::Interface { object } => {
-                    log::debug!("device interface: {object:?}");
-                    if object.interface().eq("ei_pointer") {
-                        log::debug!("GOT POINTER DEVICE");
-                        self.pointer.replace((device, object.downcast().unwrap()));
-                    } else if object.interface().eq("ei_button") {
-                        log::debug!("GOT BUTTON DEVICE");
-                        self.button.replace((device, object.downcast().unwrap()));
-                    } else if object.interface().eq("ei_scroll") {
-                        log::debug!("GOT SCROLL DEVICE");
-                        self.scroll.replace((device, object.downcast().unwrap()));
-                    } else if object.interface().eq("ei_keyboard") {
-                        log::debug!("GOT KEYBOARD DEVICE");
-                        self.keyboard.replace((device, object.downcast().unwrap()));
-                    }
-                }
-                ei::device::Event::Done => {
-                    log::debug!("device: done {device:?}");
-                }
-                ei::device::Event::Resumed { serial } => {
-                    self.serial = serial;
-                    device.start_emulating(serial, self.sequence);
-                    self.sequence += 1;
-                    log::debug!("resumed: {device:?}");
-                    if let Some((d, _)) = &mut self.pointer {
-                        if d == &device {
-                            log::debug!("pointer resumed {serial}");
-                            self.has_pointer = true;
-                        }
-                    }
-                    if let Some((d, _)) = &mut self.button {
-                        if d == &device {
-                            log::debug!("button resumed {serial}");
-                            self.has_button = true;
-                        }
-                    }
-                    if let Some((d, _)) = &mut self.scroll {
-                        if d == &device {
-                            log::debug!("scroll resumed {serial}");
-                            self.has_scroll = true;
-                        }
-                    }
-                    if let Some((d, _)) = &mut self.keyboard {
-                        if d == &device {
-                            log::debug!("keyboard resumed {serial}");
-                            self.has_keyboard = true;
-                        }
-                    }
-                }
-                ei::device::Event::Paused { serial } => {
-                    self.has_pointer = false;
-                    self.has_button = false;
-                    self.serial = serial;
-                }
-                ei::device::Event::StartEmulating { serial, sequence } => {
-                    log::debug!("start emulating {serial}, {sequence}")
-                }
-                ei::device::Event::StopEmulating { serial } => {
-                    log::debug!("stop emulating {serial}")
-                }
-                ei::device::Event::Frame { serial, timestamp } => {
-                    log::debug!("frame: {serial}, {timestamp}");
-                }
-                ei::device::Event::RegionMappingId { mapping_id } => {
-                    log::debug!("RegionMappingId {mapping_id}")
-                }
-                e => log::debug!("invalid event: {e:?}"),
-            },
-            ei::Event::Seat(seat, request) => match request {
-                ei::seat::Event::Destroyed { serial } => {
-                    self.serial = serial;
-                    log::debug!("seat destroyed: {seat:?}");
-                }
-                ei::seat::Event::Name { name } => {
-                    log::debug!("seat name: {name}");
-                }
-                ei::seat::Event::Capability { mask, interface } => {
-                    log::debug!("seat capabilities: {mask}, interface: {interface:?}");
-                    self.capabilities.insert(interface, mask);
-                    self.capability_mask |= mask;
-                }
-                ei::seat::Event::Done => {
-                    log::debug!("seat done");
-                    log::debug!("binding capabilities: {}", self.capability_mask);
-                    seat.bind(self.capability_mask);
-                }
-                ei::seat::Event::Device { device } => {
-                    log::debug!("seat: new device - {device:?}");
-                }
-                _ => todo!(),
-            },
-            e => log::debug!("unhandled event: {e:?}"),
-        }
-        self.context.flush()?;
-        Ok(())
-    }
-
     async fn create(&mut self, _: EmulationHandle) {}
     async fn destroy(&mut self, _: EmulationHandle) {}
 }
+
+async fn ei_event_handler(
+    mut events: EiConvertEventStream,
+    context: ei::Context,
+    devices: Devices,
+) -> Result<()> {
+    loop {
+        let event = events
+            .next()
+            .await
+            .ok_or(anyhow!("ei stream closed"))?
+            .map_err(|e| anyhow!("libei error: {e:?}"))?;
+        const CAPABILITIES: &[DeviceCapability] = &[
+            DeviceCapability::Pointer,
+            DeviceCapability::PointerAbsolute,
+            DeviceCapability::Keyboard,
+            DeviceCapability::Touch,
+            DeviceCapability::Scroll,
+            DeviceCapability::Button,
+        ];
+        log::debug!("{event:?}");
+        match event {
+            EiEvent::Disconnected(e) => {
+                log::debug!("ei disconnected: {e:?}");
+                break;
+            }
+            EiEvent::SeatAdded(e) => {
+                e.seat().bind_capabilities(CAPABILITIES);
+            }
+            EiEvent::SeatRemoved(e) => {
+                log::debug!("seat removed: {:?}", e.seat());
+            }
+            EiEvent::DeviceAdded(e) => {
+                let device_type = e.device().device_type();
+                log::debug!("device added: {device_type:?}");
+                e.device().device();
+                let device = e.device();
+                log::info!("GOT A DEVICE: {device:?}");
+                if let Some(pointer) = e.device().interface::<Pointer>() {
+                    log::info!("GOT POINTER");
+                    devices
+                        .pointer
+                        .write()
+                        .unwrap()
+                        .replace((device.device().clone(), pointer));
+                }
+                if let Some(keyboard) = e.device().interface::<Keyboard>() {
+                    log::info!("GOT KEYBOARD");
+                    devices
+                        .keyboard
+                        .write()
+                        .unwrap()
+                        .replace((device.device().clone(), keyboard));
+                }
+                if let Some(scroll) = e.device().interface::<Scroll>() {
+                    log::info!("GOT SCROLL");
+                    devices
+                        .scroll
+                        .write()
+                        .unwrap()
+                        .replace((device.device().clone(), scroll));
+                }
+                if let Some(button) = e.device().interface::<Button>() {
+                    log::info!("GOT BUTTON");
+                    devices
+                        .button
+                        .write()
+                        .unwrap()
+                        .replace((device.device().clone(), button));
+                }
+            }
+            EiEvent::DeviceRemoved(e) => {
+                log::debug!("device removed: {:?}", e.device().device_type());
+            }
+            EiEvent::DevicePaused(e) => {
+                log::debug!("device paused: {:?}", e.device().device_type());
+            }
+            EiEvent::DeviceResumed(e) => {
+                log::debug!("device resumed: {:?}", e.device().device_type());
+                e.device().device().start_emulating(0, 0);
+            }
+            EiEvent::KeyboardModifiers(e) => {
+                log::debug!("modifiers: {e:?}");
+            }
+            // only for receiver context
+            // EiEvent::Frame(_) => { },
+            // EiEvent::DeviceStartEmulating(_) => { },
+            // EiEvent::DeviceStopEmulating(_) => { },
+            // EiEvent::PointerMotion(_) => { },
+            // EiEvent::PointerMotionAbsolute(_) => { },
+            // EiEvent::Button(_) => { },
+            // EiEvent::ScrollDelta(_) => { },
+            // EiEvent::ScrollStop(_) => { },
+            // EiEvent::ScrollCancel(_) => { },
+            // EiEvent::ScrollDiscrete(_) => { },
+            // EiEvent::KeyboardKey(_) => { },
+            // EiEvent::TouchDown(_) => { },
+            // EiEvent::TouchUp(_) => { },
+            // EiEvent::TouchMotion(_) => { },
+            _ => unreachable!("unexpected ei event"),
+        }
+        context.flush()?;
+    }
+    Ok(())
+}

src/emulation_test.rs

@@ -27,22 +27,27 @@ async fn input_emulation_test(config: Config) -> Result<()> {
     let start = Instant::now();
     let mut offset = (0, 0);
     loop {
-        tokio::select! {
+        tokio::time::sleep(Duration::from_millis(1)).await;
-            _ = emulation.dispatch() => {}
+        let elapsed = start.elapsed();
-            _ = tokio::time::sleep(Duration::from_millis(1)) => {
+        let elapsed_sec_f64 = elapsed.as_secs_f64();
-                let elapsed = start.elapsed();
+        let second_fraction = elapsed_sec_f64 - elapsed_sec_f64 as u64 as f64;
-                let elapsed_sec_f64 = elapsed.as_secs_f64();
+        let radians = second_fraction * 2. * PI * FREQUENCY_HZ;
-                let second_fraction = elapsed_sec_f64 - elapsed_sec_f64 as u64 as f64;
+        let new_offset_f = (radians.cos() * RADIUS * 2., (radians * 2.).sin() * RADIUS);
-                let radians = second_fraction * 2. * PI * FREQUENCY_HZ;
+        let new_offset = (new_offset_f.0 as i32, new_offset_f.1 as i32);
-                let new_offset_f = (radians.cos() * RADIUS * 2., (radians * 2.).sin() * RADIUS);
+        if new_offset != offset {
-                let new_offset = (new_offset_f.0 as i32, new_offset_f.1 as i32);
+            let relative_motion = (new_offset.0 - offset.0, new_offset.1 - offset.1);
-                if new_offset != offset {
+            offset = new_offset;
-                    let relative_motion = (new_offset.0 - offset.0, new_offset.1 - offset.1);
+            let (relative_x, relative_y) = (relative_motion.0 as f64, relative_motion.1 as f64);
-                    offset = new_offset;
+            emulation
-                    let (relative_x, relative_y) = (relative_motion.0 as f64, relative_motion.1 as f64);
+                .consume(
-                    emulation.consume(Event::Pointer(PointerEvent::Motion {time: 0, relative_x, relative_y }), 0).await;
+                    Event::Pointer(PointerEvent::Motion {
-                }
+                        time: 0,
-            }
+                        relative_x,
+                        relative_y,
+                    }),
+                    0,
+                )
+                .await;
         }
     }
 }

src/server/emulation_task.rs

@@ -55,9 +55,6 @@ pub fn new(
                         None => break,
                     }
                 }
-                res = emulate.dispatch() => {
-                    res?;
-                }
             }
         }