chore: Release lan-mouse version 0.2.1

major update:
- remove threading overhead by resorting to an event driven design with mio as a backend for epoll
- Clients can now have an arbitrary amount of ip adresses and lan-mouse will automatically choose the correct one
- -> seemless switching between ethernet and wifi
- cli frontend + frontend adapter for future frontends

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "lan-mouse"
 description = "Software KVM Switch / mouse & keyboard sharing software for Local Area Networks"
-version = "0.1.1-alpha.1"
+version = "0.2.1"
 edition = "2021"
 license = "GPL-3.0-or-later"
 repository = "https://github.com/ferdinandschober/lan-mouse"
@@ -17,9 +17,13 @@ tempfile = "3.6"
 trust-dns-resolver = "0.22"
 memmap = "0.7"
 toml = "0.7"
-serde = "1.0"
-serde_derive = "1.0"
+serde = { version = "1.0", features = ["derive"] }
 anyhow = "1.0.71"
+log = "0.4.20"
+env_logger = "0.10.0"
+mio = { version = "0.8", features = ["os-ext"] }
+libc = "0.2.148"
+serde_json = "1.0.107"
 
 [target.'cfg(unix)'.dependencies]
 wayland-client = { version="0.30.2", optional = true }
@@ -27,15 +31,18 @@ wayland-protocols = { version="0.30.0", features=["client", "staging", "unstable
 wayland-protocols-wlr = { version="0.1.0", features=["client"], optional = true }
 wayland-protocols-misc = { version="0.1.0", features=["client"], optional = true }
 wayland-protocols-plasma = { version="0.1.0", features=["client"], optional = true }
+mio-signals = "0.2.0"
 x11 = { version = "2.21.0", features = ["xlib", "xtest"], optional = true }
+gtk = { package = "gtk4", version = "0.7.2", features = ["v4_8"], optional = true }
+adw = { package = "libadwaita", version = "0.5.2", features = ["v1_3"], optional = true }
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3.9", features = ["winuser"] }
 
-
 [features]
 default = ["wayland", "x11", "xdg_desktop_portal", "libei"]
 wayland = ["dep:wayland-client", "dep:wayland-protocols", "dep:wayland-protocols-wlr", "dep:wayland-protocols-misc", "dep:wayland-protocols-plasma"]
 x11 = ["dep:x11"]
 xdg_desktop_portal = []
 libei = []
+gtk = ["dep:gtk", "dep:adw"]

config.toml

@@ -1,22 +1,20 @@
 # example configuration
 
-# optional port
+# optional port (defaults to 4242)
 port = 4242
-# optional backend override
-backend = "wlroots"
 
 # define a client on the right side with host name "iridium"
 [right]
 # hostname
 host_name = "iridium"
-# optional ip address
-ip = "192.168.178.141"
+# optional list of (known) ip addresses
+ips = ["192.168.178.141"]
 # optional port (defaults to 4242)
 port = 4242
 
 # define a client on the left side with IP address 192.168.178.189
-#
-# when an IP address is specified, it takes priority
-# and host_name can be omitted
 [left]
-ip = "192.168.178.189"
+# The hostname is optional: When no hostname is specified,
+# at least one ip address needs to be specified.
+host_name = "thorium"
+ips = ["192.168.178.189"]

deny.toml

@@ -0,0 +1,272 @@
+# This template contains all of the possible sections and their default values
+
+# Note that all fields that take a lint level have these possible values:
+# * deny - An error will be produced and the check will fail
+# * warn - A warning will be produced, but the check will not fail
+# * allow - No warning or error will be produced, though in some cases a note
+# will be
+
+# The values provided in this template are the default values that will be used
+# when any section or field is not specified in your own configuration
+
+# Root options
+
+# If 1 or more target triples (and optionally, target_features) are specified,
+# only the specified targets will be checked when running `cargo deny check`.
+# This means, if a particular package is only ever used as a target specific
+# dependency, such as, for example, the `nix` crate only being used via the
+# `target_family = "unix"` configuration, that only having windows targets in
+# this list would mean the nix crate, as well as any of its exclusive
+# dependencies not shared by any other crates, would be ignored, as the target
+# list here is effectively saying which targets you are building for.
+targets = [
+    # The triple can be any string, but only the target triples built in to
+    # rustc (as of 1.40) can be checked against actual config expressions
+    #{ triple = "x86_64-unknown-linux-musl" },
+    # You can also specify which target_features you promise are enabled for a
+    # particular target. target_features are currently not validated against
+    # the actual valid features supported by the target architecture.
+    #{ triple = "wasm32-unknown-unknown", features = ["atomics"] },
+]
+# When creating the dependency graph used as the source of truth when checks are
+# executed, this field can be used to prune crates from the graph, removing them
+# from the view of cargo-deny. This is an extremely heavy hammer, as if a crate
+# is pruned from the graph, all of its dependencies will also be pruned unless
+# they are connected to another crate in the graph that hasn't been pruned,
+# so it should be used with care. The identifiers are [Package ID Specifications]
+# (https://doc.rust-lang.org/cargo/reference/pkgid-spec.html)
+#exclude = []
+# If true, metadata will be collected with `--all-features`. Note that this can't
+# be toggled off if true, if you want to conditionally enable `--all-features` it
+# is recommended to pass `--all-features` on the cmd line instead
+all-features = false
+# If true, metadata will be collected with `--no-default-features`. The same
+# caveat with `all-features` applies
+no-default-features = false
+# If set, these feature will be enabled when collecting metadata. If `--features`
+# is specified on the cmd line they will take precedence over this option.
+#features = []
+# When outputting inclusion graphs in diagnostics that include features, this
+# option can be used to specify the depth at which feature edges will be added.
+# This option is included since the graphs can be quite large and the addition
+# of features from the crate(s) to all of the graph roots can be far too verbose.
+# This option can be overridden via `--feature-depth` on the cmd line
+feature-depth = 1
+
+# This section is considered when running `cargo deny check advisories`
+# More documentation for the advisories section can be found here:
+# https://embarkstudios.github.io/cargo-deny/checks/advisories/cfg.html
+[advisories]
+# The path where the advisory database is cloned/fetched into
+db-path = "~/.cargo/advisory-db"
+# The url(s) of the advisory databases to use
+db-urls = ["https://github.com/rustsec/advisory-db"]
+# The lint level for security vulnerabilities
+vulnerability = "deny"
+# The lint level for unmaintained crates
+unmaintained = "warn"
+# The lint level for crates that have been yanked from their source registry
+yanked = "warn"
+# The lint level for crates with security notices. Note that as of
+# 2019-12-17 there are no security notice advisories in
+# https://github.com/rustsec/advisory-db
+notice = "warn"
+# A list of advisory IDs to ignore. Note that ignored advisories will still
+# output a note when they are encountered.
+ignore = [
+    #"RUSTSEC-0000-0000",
+]
+# Threshold for security vulnerabilities, any vulnerability with a CVSS score
+# lower than the range specified will be ignored. Note that ignored advisories
+# will still output a note when they are encountered.
+# * None - CVSS Score 0.0
+# * Low - CVSS Score 0.1 - 3.9
+# * Medium - CVSS Score 4.0 - 6.9
+# * High - CVSS Score 7.0 - 8.9
+# * Critical - CVSS Score 9.0 - 10.0
+#severity-threshold =
+
+# If this is true, then cargo deny will use the git executable to fetch advisory database.
+# If this is false, then it uses a built-in git library.
+# Setting this to true can be helpful if you have special authentication requirements that cargo-deny does not support.
+# See Git Authentication for more information about setting up git authentication.
+#git-fetch-with-cli = true
+
+# This section is considered when running `cargo deny check licenses`
+# More documentation for the licenses section can be found here:
+# https://embarkstudios.github.io/cargo-deny/checks/licenses/cfg.html
+[licenses]
+# The lint level for crates which do not have a detectable license
+unlicensed = "deny"
+# List of explicitly allowed licenses
+# See https://spdx.org/licenses/ for list of possible licenses
+# [possible values: any SPDX 3.11 short identifier (+ optional exception)].
+allow = [
+    "MIT",
+	"BSD-3-Clause",
+	"ISC",
+    "Apache-2.0",
+    "Apache-2.0 WITH LLVM-exception",
+	"Unicode-DFS-2016",
+]
+# List of explicitly disallowed licenses
+# See https://spdx.org/licenses/ for list of possible licenses
+# [possible values: any SPDX 3.11 short identifier (+ optional exception)].
+deny = [
+    #"Nokia",
+]
+# Lint level for licenses considered copyleft
+copyleft = "warn"
+# Blanket approval or denial for OSI-approved or FSF Free/Libre licenses
+# * both - The license will be approved if it is both OSI-approved *AND* FSF
+# * either - The license will be approved if it is either OSI-approved *OR* FSF
+# * osi-only - The license will be approved if is OSI-approved *AND NOT* FSF
+# * fsf-only - The license will be approved if is FSF *AND NOT* OSI-approved
+# * neither - This predicate is ignored and the default lint level is used
+allow-osi-fsf-free = "neither"
+# Lint level used when no other predicates are matched
+# 1. License isn't in the allow or deny lists
+# 2. License isn't copyleft
+# 3. License isn't OSI/FSF, or allow-osi-fsf-free = "neither"
+default = "deny"
+# The confidence threshold for detecting a license from license text.
+# The higher the value, the more closely the license text must be to the
+# canonical license text of a valid SPDX license file.
+# [possible values: any between 0.0 and 1.0].
+confidence-threshold = 0.8
+# Allow 1 or more licenses on a per-crate basis, so that particular licenses
+# aren't accepted for every possible crate as with the normal allow list
+exceptions = [
+    # Each entry is the crate and version constraint, and its specific allow
+    # list
+    #{ allow = ["Zlib"], name = "adler32", version = "*" },
+]
+
+# Some crates don't have (easily) machine readable licensing information,
+# adding a clarification entry for it allows you to manually specify the
+# licensing information
+#[[licenses.clarify]]
+# The name of the crate the clarification applies to
+#name = "ring"
+# The optional version constraint for the crate
+#version = "*"
+# The SPDX expression for the license requirements of the crate
+#expression = "MIT AND ISC AND OpenSSL"
+# One or more files in the crate's source used as the "source of truth" for
+# the license expression. If the contents match, the clarification will be used
+# when running the license check, otherwise the clarification will be ignored
+# and the crate will be checked normally, which may produce warnings or errors
+# depending on the rest of your configuration
+#license-files = [
+    # Each entry is a crate relative path, and the (opaque) hash of its contents
+    #{ path = "LICENSE", hash = 0xbd0eed23 }
+#]
+
+[licenses.private]
+# If true, ignores workspace crates that aren't published, or are only
+# published to private registries.
+# To see how to mark a crate as unpublished (to the official registry),
+# visit https://doc.rust-lang.org/cargo/reference/manifest.html#the-publish-field.
+ignore = false
+# One or more private registries that you might publish crates to, if a crate
+# is only published to private registries, and ignore is true, the crate will
+# not have its license(s) checked
+registries = [
+    #"https://sekretz.com/registry
+]
+
+# This section is considered when running `cargo deny check bans`.
+# More documentation about the 'bans' section can be found here:
+# https://embarkstudios.github.io/cargo-deny/checks/bans/cfg.html
+[bans]
+# Lint level for when multiple versions of the same crate are detected
+multiple-versions = "warn"
+# Lint level for when a crate version requirement is `*`
+wildcards = "allow"
+# The graph highlighting used when creating dotgraphs for crates
+# with multiple versions
+# * lowest-version - The path to the lowest versioned duplicate is highlighted
+# * simplest-path - The path to the version with the fewest edges is highlighted
+# * all - Both lowest-version and simplest-path are used
+highlight = "all"
+# The default lint level for `default` features for crates that are members of
+# the workspace that is being checked. This can be overriden by allowing/denying
+# `default` on a crate-by-crate basis if desired.
+workspace-default-features = "allow"
+# The default lint level for `default` features for external crates that are not
+# members of the workspace. This can be overriden by allowing/denying `default`
+# on a crate-by-crate basis if desired.
+external-default-features = "allow"
+# List of crates that are allowed. Use with care!
+allow = [
+    #{ name = "ansi_term", version = "=0.11.0" },
+]
+# List of crates to deny
+deny = [
+    # Each entry the name of a crate and a version range. If version is
+    # not specified, all versions will be matched.
+    #{ name = "ansi_term", version = "=0.11.0" },
+    #
+    # Wrapper crates can optionally be specified to allow the crate when it
+    # is a direct dependency of the otherwise banned crate
+    #{ name = "ansi_term", version = "=0.11.0", wrappers = [] },
+]
+
+# List of features to allow/deny
+# Each entry the name of a crate and a version range. If version is
+# not specified, all versions will be matched.
+#[[bans.features]]
+#name = "reqwest"
+# Features to not allow
+#deny = ["json"]
+# Features to allow
+#allow = [
+#    "rustls",
+#    "__rustls",
+#    "__tls",
+#    "hyper-rustls",
+#    "rustls",
+#    "rustls-pemfile",
+#    "rustls-tls-webpki-roots",
+#    "tokio-rustls",
+#    "webpki-roots",
+#]
+# If true, the allowed features must exactly match the enabled feature set. If
+# this is set there is no point setting `deny`
+#exact = true
+
+# Certain crates/versions that will be skipped when doing duplicate detection.
+skip = [
+    #{ name = "ansi_term", version = "=0.11.0" },
+]
+# Similarly to `skip` allows you to skip certain crates during duplicate
+# detection. Unlike skip, it also includes the entire tree of transitive
+# dependencies starting at the specified crate, up to a certain depth, which is
+# by default infinite.
+skip-tree = [
+    #{ name = "ansi_term", version = "=0.11.0", depth = 20 },
+]
+
+# This section is considered when running `cargo deny check sources`.
+# More documentation about the 'sources' section can be found here:
+# https://embarkstudios.github.io/cargo-deny/checks/sources/cfg.html
+[sources]
+# Lint level for what to happen when a crate from a crate registry that is not
+# in the allow list is encountered
+unknown-registry = "warn"
+# Lint level for what to happen when a crate from a git repository that is not
+# in the allow list is encountered
+unknown-git = "warn"
+# List of URLs for allowed crate registries. Defaults to the crates.io index
+# if not specified. If it is specified but empty, no registries are allowed.
+allow-registry = ["https://github.com/rust-lang/crates.io-index"]
+# List of URLs for allowed Git repositories
+allow-git = []
+
+[sources.allow-org]
+# 1 or more github.com organizations to allow git sources for
+github = [""]
+# 1 or more gitlab.com organizations to allow git sources for
+gitlab = [""]
+# 1 or more bitbucket.org organizations to allow git sources for
+bitbucket = [""]

src/backend/consumer/libei.rs

@@ -1,9 +1,18 @@
-use std::sync::mpsc::Receiver;
+use crate::consumer::EventConsumer;
 
-use crate::{event::Event, client::{ClientHandle, Client}};
+pub struct LibeiConsumer {}
 
-
-
-pub(crate) fn run(_consume_rx: Receiver<(Event, ClientHandle)>, _clients: Vec<Client>) {
-    todo!()
+impl LibeiConsumer {
+    pub fn new() -> Self { Self {  } }
+}
+
+impl EventConsumer for LibeiConsumer {
+    fn consume(&self, _: crate::event::Event, _: crate::client::ClientHandle) {
+        log::error!("libei backend not yet implemented!");
+        todo!()
+    }
+
+    fn notify(&mut self, _: crate::client::ClientEvent) {
+        todo!()
+    }
 }

src/backend/consumer/windows.rs

@@ -1,6 +1,4 @@
-use std::sync::mpsc::Receiver;
-
-use crate::event::{KeyboardEvent, PointerEvent};
+use crate::{event::{KeyboardEvent, PointerEvent}, consumer::EventConsumer};
 use winapi::{
     self,
     um::winuser::{INPUT, INPUT_MOUSE, LPINPUT, MOUSEEVENTF_MOVE, MOUSEINPUT,
@@ -16,10 +14,45 @@ use winapi::{
 };
 
 use crate::{
-    client::{Client, ClientHandle},
+    client::{ClientEvent, ClientHandle},
     event::Event,
 };
 
+
+pub struct WindowsConsumer {}
+
+impl WindowsConsumer {
+    pub fn new() -> Self { Self {  } }
+}
+
+impl EventConsumer for WindowsConsumer {
+    fn consume(&self, event: Event, _: ClientHandle) {
+        match event {
+            Event::Pointer(pointer_event) => match pointer_event {
+                PointerEvent::Motion {
+                    time: _,
+                    relative_x,
+                    relative_y,
+                } => {
+                    rel_mouse(relative_x as i32, relative_y as i32);
+                }
+                PointerEvent::Button { time:_, button, state } => { mouse_button(button, state)}
+                PointerEvent::Axis { time:_, axis, value } => { scroll(axis, value) }
+                PointerEvent::Frame {} => {}
+            },
+            Event::Keyboard(keyboard_event) => match keyboard_event {
+                KeyboardEvent::Key { time:_, key, state } => { key_event(key, state) }
+                KeyboardEvent::Modifiers { .. } => {}
+            },
+            _ => {}
+        }
+    }
+
+    fn notify(&mut self, _: ClientEvent) {
+        // nothing to do
+    }
+}
+
 fn send_mouse_input(mi: MOUSEINPUT) {
     unsafe {
         let mut input = INPUT {
@@ -114,27 +147,3 @@ fn send_keyboard_input(ki: KEYBDINPUT) {
         winapi::um::winuser::SendInput(1 as u32, &mut input, std::mem::size_of::<INPUT>() as i32);
     }
 }
-
-pub fn run(event_rx: Receiver<(Event, ClientHandle)>, _clients: Vec<Client>) {
-    loop {
-        match event_rx.recv().expect("event receiver unavailable").0 {
-            Event::Pointer(pointer_event) => match pointer_event {
-                PointerEvent::Motion {
-                    time: _,
-                    relative_x,
-                    relative_y,
-                } => {
-                    rel_mouse(relative_x as i32, relative_y as i32);
-                }
-                PointerEvent::Button { time:_, button, state } => { mouse_button(button, state)}
-                PointerEvent::Axis { time:_, axis, value } => { scroll(axis, value) }
-                PointerEvent::Frame {} => {}
-            },
-            Event::Keyboard(keyboard_event) => match keyboard_event {
-                KeyboardEvent::Key { time:_, key, state } => { key_event(key, state) }
-                KeyboardEvent::Modifiers { .. } => {}
-            },
-            Event::Release() => {}
-        }
-    }
-}

src/backend/consumer/wlroots.rs

@@ -1,16 +1,15 @@
-use crate::client::{Client, ClientHandle};
-use crate::request::{self, Request};
+use wayland_client::WEnum;
+use crate::client::{ClientHandle, ClientEvent};
+use crate::consumer::EventConsumer;
 use std::collections::HashMap;
-use std::sync::mpsc::Receiver;
-use std::time::Duration;
-use std::{io, thread};
-use std::{
-    io::{BufWriter, Write},
-    os::unix::prelude::AsRawFd,
-};
+use std::os::fd::OwnedFd;
+use std::os::unix::prelude::AsRawFd;
 
+use anyhow::{Result, anyhow};
 use wayland_client::globals::BindError;
 use wayland_client::protocol::wl_pointer::{Axis, ButtonState};
+use wayland_client::protocol::wl_keyboard::{self, WlKeyboard};
+use wayland_client::protocol::wl_seat::WlSeat;
 use wayland_protocols_wlr::virtual_pointer::v1::client::{
     zwlr_virtual_pointer_manager_v1::ZwlrVirtualPointerManagerV1 as VpManager,
     zwlr_virtual_pointer_v1::ZwlrVirtualPointerV1 as Vp,
@@ -30,8 +29,6 @@ use wayland_client::{
     Connection, Dispatch, EventQueue, QueueHandle,
 };
 
-use tempfile;
-
 use crate::event::{Event, KeyboardEvent, PointerEvent};
 
 enum VirtualInputManager {
@@ -39,27 +36,31 @@ enum VirtualInputManager {
     Kde { fake_input: OrgKdeKwinFakeInput },
 }
 
-// App State, implements Dispatch event handlers
-struct App {
+struct State {
+    keymap: Option<(u32, OwnedFd, u32)>,
     input_for_client: HashMap<ClientHandle, VirtualInput>,
     seat: wl_seat::WlSeat,
-    event_rx: Receiver<(Event, ClientHandle)>,
     virtual_input_manager: VirtualInputManager,
-    queue: EventQueue<Self>,
     qh: QueueHandle<Self>,
 }
 
-pub fn run(event_rx: Receiver<(Event, ClientHandle)>, clients: Vec<Client>) {
-    let mut app = App::new(event_rx, clients);
-    app.run();
+// App State, implements Dispatch event handlers
+pub(crate) struct WlrootsConsumer {
+    state: State,
+    queue: EventQueue<State>,
 }
 
-impl App {
-    pub fn new(event_rx: Receiver<(Event, ClientHandle)>, clients: Vec<Client>) -> Self {
+impl WlrootsConsumer {
+    pub fn new() -> Result<Self> {
         let conn = Connection::connect_to_env().unwrap();
-        let (globals, queue) = registry_queue_init::<App>(&conn).unwrap();
+        let (globals, queue) = registry_queue_init::<State>(&conn).unwrap();
         let qh = queue.handle();
 
+        let seat: wl_seat::WlSeat = match globals.bind(&qh, 7..=8, ()) {
+            Ok(wl_seat) => wl_seat,
+            Err(_) => return Err(anyhow!("wl_seat >= v7 not supported")),
+        };
+
         let vpm: Result<VpManager, BindError> = globals.bind(&qh, 1..=1, ());
         let vkm: Result<VkManager, BindError> = globals.bind(&qh, 1..=1, ());
         let fake_input: Result<OrgKdeKwinFakeInput, BindError> = globals.bind(&qh, 4..=4, ());
@@ -74,10 +75,11 @@ impl App {
                 VirtualInputManager::Kde { fake_input }
             }
             (Err(e1), Err(e2), Err(e3)) => {
-                eprintln!("zwlr_virtual_pointer_v1: {e1}");
-                eprintln!("zwp_virtual_keyboard_v1: {e2}");
-                eprintln!("org_kde_kwin_fake_input: {e3}");
-                panic!("neither wlroots nor kde input emulation protocol supported!")
+                log::warn!("zwlr_virtual_pointer_v1: {e1}");
+                log::warn!("zwp_virtual_keyboard_v1: {e2}");
+                log::warn!("org_kde_kwin_fake_input: {e3}");
+                log::error!("neither wlroots nor kde input emulation protocol supported!");
+                return Err(anyhow!("could not create event consumer"));
             }
             _ => {
                 panic!()
@@ -85,86 +87,76 @@ impl App {
         };
 
         let input_for_client: HashMap<ClientHandle, VirtualInput> = HashMap::new();
-        let seat: wl_seat::WlSeat = globals.bind(&qh, 7..=8, ()).unwrap();
-        let mut app = App {
-            input_for_client,
-            seat,
-            event_rx,
-            virtual_input_manager,
+
+        let mut consumer = WlrootsConsumer {
+            state: State {
+                keymap: None,
+                input_for_client,
+                seat,
+                virtual_input_manager,
+                qh,
+            },
             queue,
-            qh,
         };
-        for client in clients {
-            app.add_client(client);
+        while consumer.state.keymap.is_none() {
+            consumer.queue.blocking_dispatch(&mut consumer.state).unwrap();
         }
-        app
+        // let fd = unsafe { &File::from_raw_fd(consumer.state.keymap.unwrap().1.as_raw_fd()) };
+        // let mmap = unsafe { MmapOptions::new().map_copy(fd).unwrap() };
+        // log::debug!("{:?}", &mmap[..100]);
+        Ok(consumer)
     }
+}
 
-    pub fn run(&mut self) {
-        loop {
-            let (event, client) = self.event_rx.recv().expect("event receiver unavailable");
-            if let Some(virtual_input) = self.input_for_client.get(&client) {
-                virtual_input.consume_event(event).unwrap();
-                if let Err(e) = self.queue.flush() {
-                    eprintln!("{}", e);
-                }
-            }
-        }
-    }
-
-    fn add_client(&mut self, client: Client) {
+impl State {
+    fn add_client(&mut self, client: ClientHandle) {
         // create virtual input devices
         match &self.virtual_input_manager {
             VirtualInputManager::Wlroots { vpm, vkm } => {
                 let pointer: Vp = vpm.create_virtual_pointer(None, &self.qh, ());
                 let keyboard: Vk = vkm.create_virtual_keyboard(&self.seat, &self.qh, ());
 
-                // receive keymap from device
-                eprint!("\rconnecting to {} ", client.addr);
-                let mut attempts = 0;
-                let data = loop {
-                    let result = request::request_data(client.addr, Request::KeyMap);
-                    eprint!("\rconnecting to {} ", client.addr);
-                    for _ in 0..attempts {
-                        eprint!(".");
-                    }
-                    match result {
-                        Ok(data) => break data,
-                        Err(e) => {
-                            eprint!(" - {}", e);
-                        }
-                    }
-                    io::stderr().flush().unwrap();
-                    thread::sleep(Duration::from_millis(500));
-                    attempts += 1;
-                };
-
-                eprint!("\rconnecting to {} ", client.addr);
-                for _ in 0..attempts {
-                    eprint!(".");
+                // TODO: use server side keymap
+                if let Some((format, fd, size)) = self.keymap.as_ref() {
+                    keyboard.keymap(*format, fd.as_raw_fd(), *size);
+                } else {
+                    panic!("no keymap");
                 }
-                eprintln!(" done!                          ");
-
-                // TODO use shm_open
-                let f = tempfile::tempfile().unwrap();
-                let mut buf = BufWriter::new(&f);
-                buf.write_all(&data[..]).unwrap();
-                buf.flush().unwrap();
-                keyboard.keymap(1, f.as_raw_fd(), data.len() as u32);
 
                 let vinput = VirtualInput::Wlroots { pointer, keyboard };
 
-                self.input_for_client.insert(client.handle, vinput);
+                self.input_for_client.insert(client, vinput);
             }
             VirtualInputManager::Kde { fake_input } => {
                 let fake_input = fake_input.clone();
                 let vinput = VirtualInput::Kde { fake_input };
-                self.input_for_client.insert(client.handle, vinput);
+                self.input_for_client.insert(client, vinput);
             }
         }
     }
 }
 
+impl EventConsumer for WlrootsConsumer {
+    fn consume(&self, event: Event, client_handle: ClientHandle) {
+        if let Some(virtual_input) = self.state.input_for_client.get(&client_handle) {
+            virtual_input.consume_event(event).unwrap();
+            if let Err(e) = self.queue.flush() {
+                log::error!("{}", e);
+            }
+        }
+    }
+
+    fn notify(&mut self, client_event: ClientEvent) {
+        if let ClientEvent::Create(client, _) = client_event {
+            self.state.add_client(client);
+            if let Err(e) = self.queue.flush() {
+                log::error!("{}", e);
+            }
+        }
+    }
+}
+
+
 enum VirtualInput {
     Wlroots { pointer: Vp, keyboard: Vk },
     Kde { fake_input: OrgKdeKwinFakeInput },
@@ -184,7 +176,6 @@ impl VirtualInput {
                         keyboard: _,
                     } => {
                         pointer.motion(time, relative_x, relative_y);
-                        pointer.frame();
                     }
                     VirtualInput::Kde { fake_input } => {
                         fake_input.pointer_motion(relative_y, relative_y);
@@ -202,7 +193,6 @@ impl VirtualInput {
                             keyboard: _,
                         } => {
                             pointer.button(time, button, state);
-                            pointer.frame();
                         }
                         VirtualInput::Kde { fake_input } => {
                             fake_input.button(button, state as u32);
@@ -261,36 +251,64 @@ impl VirtualInput {
                     VirtualInput::Kde { fake_input: _ } => {}
                 },
             },
-            Event::Release() => match self {
-                VirtualInput::Wlroots {
-                    pointer: _,
-                    keyboard,
-                } => {
-                    keyboard.modifiers(77, 0, 0, 0);
-                    keyboard.modifiers(0, 0, 0, 0);
-                }
-                VirtualInput::Kde { fake_input: _ } => {}
-            },
+            _ => {},
         }
         Ok(())
     }
 }
 
-delegate_noop!(App: Vp);
-delegate_noop!(App: Vk);
-delegate_noop!(App: VpManager);
-delegate_noop!(App: VkManager);
-delegate_noop!(App: wl_seat::WlSeat);
-delegate_noop!(App: OrgKdeKwinFakeInput);
+delegate_noop!(State: Vp);
+delegate_noop!(State: Vk);
+delegate_noop!(State: VpManager);
+delegate_noop!(State: VkManager);
+delegate_noop!(State: OrgKdeKwinFakeInput);
 
-impl Dispatch<wl_registry::WlRegistry, GlobalListContents> for App {
+impl Dispatch<wl_registry::WlRegistry, GlobalListContents> for State {
     fn event(
-        _: &mut App,
+        _: &mut State,
         _: &wl_registry::WlRegistry,
         _: wl_registry::Event,
         _: &GlobalListContents,
         _: &Connection,
-        _: &QueueHandle<App>,
+        _: &QueueHandle<State>,
     ) {
     }
 }
+
+impl Dispatch<WlKeyboard, ()> for State {
+    fn event(
+        state: &mut Self,
+        _: &WlKeyboard,
+        event: <WlKeyboard as wayland_client::Proxy>::Event,
+        _: &(),
+        _: &Connection,
+        _: &QueueHandle<Self>,
+    ) {
+        match event {
+            wl_keyboard::Event::Keymap { format, fd, size } => {
+                state.keymap = Some((u32::from(format), fd, size));
+            }
+            _ => {},
+        }
+    }
+}
+
+impl Dispatch<WlSeat, ()> for State {
+    fn event(
+        _: &mut Self,
+        seat: &WlSeat,
+        event: <WlSeat as wayland_client::Proxy>::Event,
+        _: &(),
+        _: &Connection,
+        qhandle: &QueueHandle<Self>,
+    ) {
+        if let wl_seat::Event::Capabilities {
+            capabilities: WEnum::Value(capabilities),
+        } = event
+        {
+            if capabilities.contains(wl_seat::Capability::Keyboard) {
+                seat.get_keyboard(qhandle, ());
+            }
+        }
+    }
+}

src/backend/consumer/x11.rs

@@ -1,49 +1,57 @@
-use std::{ptr, sync::mpsc::Receiver};
+use std::ptr;
 use x11::{xlib, xtest};
 
 use crate::{
-    client::{Client, ClientHandle},
-    event::Event,
+    client::ClientHandle,
+    event::Event, consumer::EventConsumer,
 };
 
-fn open_display() -> Option<*mut xlib::Display> {
-    unsafe {
-        match xlib::XOpenDisplay(ptr::null()) {
-            d if d == ptr::null::<xlib::Display>() as *mut xlib::Display => None,
-            display => Some(display),
+pub struct X11Consumer {
+    display: *mut xlib::Display,
+}
+
+impl X11Consumer {
+    pub fn new() -> Self {
+        let display = unsafe {
+            match xlib::XOpenDisplay(ptr::null()) {
+                d if d == ptr::null::<xlib::Display>() as *mut xlib::Display => None,
+                display => Some(display),
+            }
+        };
+        let display = display.expect("could not open display");
+        Self { display }
+    }
+
+    fn relative_motion(&self, dx: i32, dy: i32) {
+        unsafe {
+            xtest::XTestFakeRelativeMotionEvent(self.display, dx, dy, 0, 0);
+            xlib::XFlush(self.display);
         }
     }
 }
 
-fn relative_motion(display: *mut xlib::Display, dx: i32, dy: i32) {
-    unsafe {
-        xtest::XTestFakeRelativeMotionEvent(display, dx, dy, 0, 0);
-        xlib::XFlush(display);
-    }
-}
-
-pub fn run(event_rx: Receiver<(Event, ClientHandle)>, _clients: Vec<Client>) {
-    let display = match open_display() {
-        None => panic!("could not open display!"),
-        Some(display) => display,
-    };
-
-    loop {
-        match event_rx.recv().expect("event receiver unavailable").0 {
+impl EventConsumer for X11Consumer {
+    fn consume(&self, event: Event, _: ClientHandle) {
+        match event {
             Event::Pointer(pointer_event) => match pointer_event {
                 crate::event::PointerEvent::Motion {
                     time: _,
                     relative_x,
                     relative_y,
                 } => {
-                    relative_motion(display, relative_x as i32, relative_y as i32);
+                    self.relative_motion(relative_x as i32, relative_y as i32);
                 }
                 crate::event::PointerEvent::Button { .. } => {}
                 crate::event::PointerEvent::Axis { .. } => {}
                 crate::event::PointerEvent::Frame {} => {}
             },
             Event::Keyboard(_) => {}
-            Event::Release() => {}
+            _ => {}
         }
     }
+
+    fn notify(&mut self, _: crate::client::ClientEvent) {
+        // for our purposes it does not matter what client sent the event
+    }
 }
+

src/backend/consumer/xdg_desktop_portal.rs

@@ -1,9 +1,15 @@
-use std::sync::mpsc::Receiver;
+use crate::consumer::EventConsumer;
 
-use crate::{event::Event, client::{ClientHandle, Client}};
+pub struct DesktopPortalConsumer {}
 
-
-
-pub(crate) fn run(_consume_rx: Receiver<(Event, ClientHandle)>, _clients: Vec<Client>) {
-    todo!()
+impl DesktopPortalConsumer {
+    pub fn new() -> Self { Self {  } }
+}
+
+impl EventConsumer for DesktopPortalConsumer {
+    fn consume(&self, _: crate::event::Event, _: crate::client::ClientHandle) {
+        log::error!("xdg_desktop_portal backend not yet implemented!");
+    }
+
+    fn notify(&mut self, _: crate::client::ClientEvent) {}
 }

src/backend/producer/wayland.rs

@@ -1,18 +1,15 @@
-use crate::{
-    client::{Client, ClientHandle, Position},
-    request,
-};
+use crate::{client::{ClientHandle, Position, ClientEvent}, producer::EventProducer};
+use mio::{event::Source, unix::SourceFd};
 
+use std::{os::fd::RawFd, vec::Drain, io::ErrorKind};
 use memmap::MmapOptions;
+use anyhow::{anyhow, Result};
 
 use std::{
     fs::File,
     io::{BufWriter, Write},
     os::unix::prelude::{AsRawFd, FromRawFd},
     rc::Rc,
-    sync::mpsc::SyncSender,
-    thread,
-    time::Duration,
 };
 
 use wayland_protocols::wp::{
@@ -36,14 +33,14 @@ use wayland_protocols_wlr::layer_shell::v1::client::{
 };
 
 use wayland_client::{
-    backend::WaylandError,
+    backend::{WaylandError, ReadEventsGuard},
     delegate_noop,
     globals::{registry_queue_init, GlobalListContents},
     protocol::{
         wl_buffer, wl_compositor, wl_keyboard, wl_pointer, wl_region, wl_registry, wl_seat, wl_shm,
         wl_shm_pool, wl_surface,
     },
-    Connection, Dispatch, DispatchError, QueueHandle, WEnum,
+    Connection, Dispatch, DispatchError, QueueHandle, WEnum, EventQueue,
 };
 
 use tempfile;
@@ -60,17 +57,22 @@ struct Globals {
     layer_shell: ZwlrLayerShellV1,
 }
 
-struct App {
-    running: bool,
+struct State {
     pointer_lock: Option<ZwpLockedPointerV1>,
     rel_pointer: Option<ZwpRelativePointerV1>,
     shortcut_inhibitor: Option<ZwpKeyboardShortcutsInhibitorV1>,
     client_for_window: Vec<(Rc<Window>, ClientHandle)>,
     focused: Option<(Rc<Window>, ClientHandle)>,
     g: Globals,
-    tx: SyncSender<(Event, ClientHandle)>,
-    server: request::Server,
+    wayland_fd: RawFd,
+    read_guard: Option<ReadEventsGuard>,
     qh: QueueHandle<Self>,
+    pending_events: Vec<(ClientHandle, Event)>,
+}
+
+pub struct WaylandEventProducer {
+    state: State,
+    queue: EventQueue<State>,
 }
 
 struct Window {
@@ -80,7 +82,7 @@ struct Window {
 }
 
 impl Window {
-    fn new(g: &Globals, qh: &QueueHandle<App>, pos: Position) -> Window {
+    fn new(g: &Globals, qh: &QueueHandle<State>, pos: Position) -> Window {
         let (width, height) = (1, 1440);
         let mut file = tempfile::tempfile().unwrap();
         draw(&mut file, (width, height));
@@ -127,80 +129,6 @@ impl Window {
     }
 }
 
-pub fn run(tx: SyncSender<(Event, ClientHandle)>, server: request::Server, clients: Vec<Client>) {
-    let conn = Connection::connect_to_env().expect("could not connect to wayland compositor");
-    let (g, mut queue) =
-        registry_queue_init::<App>(&conn).expect("failed to initialize wl_registry");
-    let qh = queue.handle();
-
-    let compositor: wl_compositor::WlCompositor = g
-        .bind(&qh, 4..=5, ())
-        .expect("wl_compositor >= v4 not supported");
-    let shm: wl_shm::WlShm = g.bind(&qh, 1..=1, ()).expect("wl_shm v1 not supported");
-    let layer_shell: ZwlrLayerShellV1 = g
-        .bind(&qh, 3..=4, ())
-        .expect("zwlr_layer_shell_v1 >= v3 not supported - required to display a surface at the edge of the screen");
-    let seat: wl_seat::WlSeat = g.bind(&qh, 7..=8, ()).expect("wl_seat >= v7 not supported");
-    let pointer_constraints: ZwpPointerConstraintsV1 = g
-        .bind(&qh, 1..=1, ())
-        .expect("zwp_pointer_constraints_v1 not supported");
-    let relative_pointer_manager: ZwpRelativePointerManagerV1 = g
-        .bind(&qh, 1..=1, ())
-        .expect("zwp_relative_pointer_manager_v1 not supported");
-    let shortcut_inhibit_manager: ZwpKeyboardShortcutsInhibitManagerV1 = g
-        .bind(&qh, 1..=1, ())
-        .expect("zwp_keyboard_shortcuts_inhibit_manager_v1 not supported");
-
-    let g = Globals {
-        compositor,
-        shm,
-        layer_shell,
-        seat,
-        pointer_constraints,
-        relative_pointer_manager,
-        shortcut_inhibit_manager,
-    };
-
-    let client_for_window = Vec::new();
-
-    let mut app = App {
-        running: true,
-        g,
-        pointer_lock: None,
-        rel_pointer: None,
-        shortcut_inhibitor: None,
-        client_for_window,
-        focused: None,
-        tx,
-        server,
-        qh,
-    };
-
-    for client in clients {
-        app.add_client(client.handle, client.pos);
-    }
-
-    while app.running {
-        match queue.blocking_dispatch(&mut app) {
-            Ok(_) => {}
-            Err(DispatchError::Backend(WaylandError::Io(e))) => {
-                eprintln!("Wayland Error: {}", e);
-                thread::sleep(Duration::from_millis(500));
-            }
-            Err(DispatchError::Backend(e)) => {
-                panic!("{}", e);
-            }
-            Err(DispatchError::BadMessage {
-                sender_id,
-                interface,
-                opcode,
-            }) => {
-                panic!("bad message {}, {} , {}", sender_id, interface, opcode);
-            }
-        }
-    }
-}
-
 fn draw(f: &mut File, (width, height): (u32, u32)) {
     let mut buf = BufWriter::new(f);
     for _ in 0..height {
@@ -210,13 +138,91 @@ fn draw(f: &mut File, (width, height): (u32, u32)) {
     }
 }
 
-impl App {
+impl WaylandEventProducer {
+    pub fn new() -> Result<Self> {
+        let conn = Connection::connect_to_env().expect("could not connect to wayland compositor");
+        let (g, queue) =
+            registry_queue_init::<State>(&conn).expect("failed to initialize wl_registry");
+        let qh = queue.handle();
+
+        let compositor: wl_compositor::WlCompositor = match g.bind(&qh, 4..=5, ()) {
+            Ok(compositor) => compositor,
+            Err(_) => return Err(anyhow!("wl_compositor >= v4 not supported")),
+        };
+
+        let shm: wl_shm::WlShm = match g.bind(&qh, 1..=1, ()) {
+            Ok(wl_shm) => wl_shm,
+            Err(_) => return Err(anyhow!("wl_shm v1 not supported")),
+        };
+
+        let layer_shell: ZwlrLayerShellV1 = match g.bind(&qh, 3..=4, ()) {
+            Ok(layer_shell) => layer_shell,
+            Err(_) => return Err(anyhow!("zwlr_layer_shell_v1 >= v3 not supported - required to display a surface at the edge of the screen")),
+        };
+
+        let seat: wl_seat::WlSeat = match g.bind(&qh, 7..=8, ()) {
+            Ok(wl_seat) => wl_seat,
+            Err(_) => return Err(anyhow!("wl_seat >= v7 not supported")),
+        };
+
+        let pointer_constraints: ZwpPointerConstraintsV1 = match g.bind(&qh, 1..=1, ()) {
+            Ok(pointer_constraints) => pointer_constraints,
+            Err(_) => return Err(anyhow!("zwp_pointer_constraints_v1 not supported")),
+        };
+
+        let relative_pointer_manager: ZwpRelativePointerManagerV1 = match g.bind(&qh, 1..=1, ()) {
+            Ok(relative_pointer_manager) => relative_pointer_manager,
+            Err(_) => return Err(anyhow!("zwp_relative_pointer_manager_v1 not supported")),
+        };
+
+        let shortcut_inhibit_manager: ZwpKeyboardShortcutsInhibitManagerV1 = match g.bind(&qh, 1..=1, ()) {
+            Ok(shortcut_inhibit_manager) => shortcut_inhibit_manager,
+            Err(_) => return Err(anyhow!("zwp_keyboard_shortcuts_inhibit_manager_v1 not supported")),
+        };
+
+        let g = Globals {
+            compositor,
+            shm,
+            layer_shell,
+            seat,
+            pointer_constraints,
+            relative_pointer_manager,
+            shortcut_inhibit_manager,
+        };
+
+        // flush outgoing events
+        queue.flush()?;
+
+        // prepare reading wayland events
+        let read_guard = queue.prepare_read()?;
+        let wayland_fd = read_guard.connection_fd().as_raw_fd();
+        let read_guard = Some(read_guard);
+
+        Ok(WaylandEventProducer {
+            queue,
+            state: State {
+                g,
+                pointer_lock: None,
+                rel_pointer: None,
+                shortcut_inhibitor: None,
+                client_for_window: Vec::new(),
+                focused: None,
+                qh,
+                wayland_fd,
+                read_guard,
+                pending_events: vec![],
+            }
+        })
+    }
+}
+
+impl State {
     fn grab(
         &mut self,
         surface: &wl_surface::WlSurface,
         pointer: &wl_pointer::WlPointer,
         serial: u32,
-        qh: &QueueHandle<App>,
+        qh: &QueueHandle<State>,
     ) {
         let (window, _) = self.focused.as_ref().unwrap();
 
@@ -235,7 +241,7 @@ impl App {
                 surface,
                 pointer,
                 None,
-                Lifetime::Oneshot,
+                Lifetime::Persistent,
                 qh,
                 (),
             ));
@@ -263,7 +269,10 @@ impl App {
 
     fn ungrab(&mut self) {
         // get focused client
-        let (window, _client) = self.focused.as_ref().unwrap();
+        let (window, _client) = match self.focused.as_ref() {
+            Some(focused) => focused,
+            None => return,
+        };
 
         // ungrab surface
         window
@@ -271,7 +280,7 @@ impl App {
             .set_keyboard_interactivity(KeyboardInteractivity::None);
         window.surface.commit();
 
-        // release pointer
+        // destroy pointer lock
         if let Some(pointer_lock) = &self.pointer_lock {
             pointer_lock.destroy();
             self.pointer_lock = None;
@@ -283,7 +292,7 @@ impl App {
             self.rel_pointer = None;
         }
 
-        // release shortcut inhibitor
+        // destroy shortcut inhibitor
         if let Some(shortcut_inhibitor) = &self.shortcut_inhibitor {
             shortcut_inhibitor.destroy();
             self.shortcut_inhibitor = None;
@@ -296,7 +305,126 @@ impl App {
     }
 }
 
-impl Dispatch<wl_seat::WlSeat, ()> for App {
+impl Source for WaylandEventProducer {
+    fn register(
+        &mut self,
+        registry: &mio::Registry,
+        token: mio::Token,
+        interests: mio::Interest,
+    ) -> std::io::Result<()> {
+        SourceFd(&self.state.wayland_fd).register(registry, token, interests)
+    }
+
+    fn reregister(
+        &mut self,
+        registry: &mio::Registry,
+        token: mio::Token,
+        interests: mio::Interest,
+    ) -> std::io::Result<()> {
+        SourceFd(&self.state.wayland_fd).reregister(registry, token, interests)
+    }
+
+    fn deregister(&mut self, registry: &mio::Registry) -> std::io::Result<()> {
+        SourceFd(&self.state.wayland_fd).deregister(registry)
+    }
+}
+impl WaylandEventProducer {
+    fn read(&mut self) -> bool {
+        match self.state.read_guard.take().unwrap().read() {
+            Ok(_) => true,
+            Err(WaylandError::Io(e)) if e.kind() == ErrorKind::WouldBlock => false,
+            Err(WaylandError::Io(e)) => {
+                log::error!("error reading from wayland socket: {e}");
+                false
+            }
+            Err(WaylandError::Protocol(e)) => {
+                panic!("wayland protocol violation: {e}")
+            }
+        }
+    }
+
+    fn prepare_read(&mut self) {
+        match self.queue.prepare_read() {
+            Ok(r) => self.state.read_guard = Some(r),
+            Err(WaylandError::Io(e)) => {
+                log::error!("error preparing read from wayland socket: {e}")
+            }
+            Err(WaylandError::Protocol(e)) => {
+                panic!("wayland Protocol violation: {e}")
+            }
+        };
+    }
+
+    fn dispatch_events(&mut self) {
+        match self.queue.dispatch_pending(&mut self.state) {
+            Ok(_) => {}
+            Err(DispatchError::Backend(WaylandError::Io(e))) => {
+                log::error!("Wayland Error: {}", e);
+            }
+            Err(DispatchError::Backend(e)) => {
+                panic!("backend error: {}", e);
+            }
+            Err(DispatchError::BadMessage {
+                sender_id,
+                interface,
+                opcode,
+            }) => {
+                panic!("bad message {}, {} , {}", sender_id, interface, opcode);
+            }
+        }
+    }
+
+    fn flush_events(&mut self) {
+        // flush outgoing events
+        match self.queue.flush() {
+            Ok(_) => (),
+            Err(e) => match e {
+                WaylandError::Io(e) => {
+                    log::error!("error writing to wayland socket: {e}")
+                },
+                WaylandError::Protocol(e) => {
+                    panic!("wayland protocol violation: {e}")
+                },
+            },
+        }
+    }
+}
+
+impl EventProducer for WaylandEventProducer {
+
+    fn read_events(&mut self) -> Drain<(ClientHandle, Event)> {
+        // read events
+        while self.read() {
+            // prepare next read
+            self.prepare_read();
+        }
+        // dispatch the events
+        self.dispatch_events();
+
+        // flush outgoing events
+        self.flush_events();
+
+        // prepare for the next read
+        self.prepare_read();
+
+        // return the events
+        self.state.pending_events.drain(..)
+    }
+
+    fn notify(&mut self, client_event: ClientEvent) {
+        if let ClientEvent::Create(handle, pos) = client_event {
+            self.state.add_client(handle, pos);
+            self.flush_events();
+        }
+    }
+
+    fn release(&mut self) {
+        self.state.ungrab();
+        self.flush_events();
+    }
+}
+
+impl Dispatch<wl_seat::WlSeat, ()> for State {
     fn event(
         _: &mut Self,
         seat: &wl_seat::WlSeat,
@@ -319,7 +447,7 @@ impl Dispatch<wl_seat::WlSeat, ()> for App {
     }
 }
 
-impl Dispatch<wl_pointer::WlPointer, ()> for App {
+impl Dispatch<wl_pointer::WlPointer, ()> for State {
     fn event(
         app: &mut Self,
         pointer: &wl_pointer::WlPointer,
@@ -328,6 +456,7 @@ impl Dispatch<wl_pointer::WlPointer, ()> for App {
         _: &Connection,
         qh: &QueueHandle<Self>,
     ) {
+
         match event {
             wl_pointer::Event::Enter {
                 serial,
@@ -336,6 +465,7 @@ impl Dispatch<wl_pointer::WlPointer, ()> for App {
                 surface_y: _,
             } => {
                 // get client corresponding to the focused surface
+                log::trace!("produce: enter()");
 
                 {
                     let (window, client) = app
@@ -351,9 +481,10 @@ impl Dispatch<wl_pointer::WlPointer, ()> for App {
                     .iter()
                     .find(|(w, _c)| w.surface == surface)
                     .unwrap();
-                app.tx.send((Event::Release(), *client)).unwrap();
+                app.pending_events.push((*client, Event::Release()));
             }
             wl_pointer::Event::Leave { .. } => {
+                log::trace!("produce: leave()");
                 app.ungrab();
             }
             wl_pointer::Event::Button {
@@ -362,43 +493,43 @@ impl Dispatch<wl_pointer::WlPointer, ()> for App {
                 button,
                 state,
             } => {
+                log::trace!("produce: button()");
                 let (_, client) = app.focused.as_ref().unwrap();
-                app.tx
-                    .send((
-                        Event::Pointer(PointerEvent::Button {
-                            time,
-                            button,
-                            state: u32::from(state),
-                        }),
-                        *client,
-                    ))
-                    .unwrap();
+                app.pending_events.push((
+                    *client,
+                    Event::Pointer(PointerEvent::Button {
+                        time,
+                        button,
+                        state: u32::from(state),
+                    }),
+                ));
             }
             wl_pointer::Event::Axis { time, axis, value } => {
+                log::trace!("produce: scroll()");
                 let (_, client) = app.focused.as_ref().unwrap();
-                app.tx
-                    .send((
-                        Event::Pointer(PointerEvent::Axis {
-                            time,
-                            axis: u32::from(axis) as u8,
-                            value,
-                        }),
-                        *client,
-                    ))
-                    .unwrap();
+                app.pending_events.push((
+                    *client,
+                    Event::Pointer(PointerEvent::Axis {
+                        time,
+                        axis: u32::from(axis) as u8,
+                        value,
+                    }),
+                ));
             }
             wl_pointer::Event::Frame {} => {
+                log::trace!("produce: frame()");
                 let (_, client) = app.focused.as_ref().unwrap();
-                app.tx
-                    .send((Event::Pointer(PointerEvent::Frame {}), *client))
-                    .unwrap();
+                app.pending_events.push((
+                    *client,
+                    Event::Pointer(PointerEvent::Frame {}),
+                ));
             }
             _ => {}
         }
     }
 }
 
-impl Dispatch<wl_keyboard::WlKeyboard, ()> for App {
+impl Dispatch<wl_keyboard::WlKeyboard, ()> for State {
     fn event(
         app: &mut Self,
         _: &wl_keyboard::WlKeyboard,
@@ -419,16 +550,14 @@ impl Dispatch<wl_keyboard::WlKeyboard, ()> for App {
                 state,
             } => {
                 if let Some(client) = client {
-                    app.tx
-                        .send((
-                            Event::Keyboard(KeyboardEvent::Key {
-                                time,
-                                key,
-                                state: u32::from(state) as u8,
-                            }),
-                            *client,
-                        ))
-                        .unwrap();
+                    app.pending_events.push((
+                        *client,
+                        Event::Keyboard(KeyboardEvent::Key {
+                            time,
+                            key,
+                            state: u32::from(state) as u8,
+                        }),
+                    ));
                 }
             }
             wl_keyboard::Event::Modifiers {
@@ -439,17 +568,15 @@ impl Dispatch<wl_keyboard::WlKeyboard, ()> for App {
                 group,
             } => {
                 if let Some(client) = client {
-                    app.tx
-                        .send((
-                            Event::Keyboard(KeyboardEvent::Modifiers {
-                                mods_depressed,
-                                mods_latched,
-                                mods_locked,
-                                group,
-                            }),
-                            *client,
-                        ))
-                        .unwrap();
+                    app.pending_events.push((
+                        *client,
+                        Event::Keyboard(KeyboardEvent::Modifiers {
+                            mods_depressed,
+                            mods_latched,
+                            mods_locked,
+                            group,
+                        }),
+                    ));
                 }
                 if mods_depressed == 77 {
                     // ctrl shift super alt
@@ -462,15 +589,15 @@ impl Dispatch<wl_keyboard::WlKeyboard, ()> for App {
                 size: _,
             } => {
                 let fd = unsafe { &File::from_raw_fd(fd.as_raw_fd()) };
-                let mmap = unsafe { MmapOptions::new().map_copy(fd).unwrap() };
-                app.server.offer_data(request::Request::KeyMap, mmap);
+                let _mmap = unsafe { MmapOptions::new().map_copy(fd).unwrap() };
+                // TODO keymap
             }
             _ => (),
         }
     }
 }
 
-impl Dispatch<ZwpRelativePointerV1, ()> for App {
+impl Dispatch<ZwpRelativePointerV1, ()> for State {
     fn event(
         app: &mut Self,
         _: &ZwpRelativePointerV1,
@@ -488,24 +615,23 @@ impl Dispatch<ZwpRelativePointerV1, ()> for App {
             dy_unaccel: surface_y,
         } = event
         {
+            log::trace!("produce: motion()");
             if let Some((_window, client)) = &app.focused {
                 let time = (((utime_hi as u64) << 32 | utime_lo as u64) / 1000) as u32;
-                app.tx
-                    .send((
-                        Event::Pointer(PointerEvent::Motion {
-                            time,
-                            relative_x: surface_x,
-                            relative_y: surface_y,
-                        }),
-                        *client,
-                    ))
-                    .unwrap();
+                app.pending_events.push((
+                    *client,
+                    Event::Pointer(PointerEvent::Motion {
+                        time,
+                        relative_x: surface_x,
+                        relative_y: surface_y,
+                    }),
+                ));
             }
         }
     }
 }
 
-impl Dispatch<ZwlrLayerSurfaceV1, ()> for App {
+impl Dispatch<ZwlrLayerSurfaceV1, ()> for State {
     fn event(
         app: &mut Self,
         layer_surface: &ZwlrLayerSurfaceV1,
@@ -523,9 +649,8 @@ impl Dispatch<ZwlrLayerSurfaceV1, ()> for App {
             // client corresponding to the layer_surface
             let surface = &window.surface;
             let buffer = &window.buffer;
-            surface.commit();
-            layer_surface.ack_configure(serial);
             surface.attach(Some(&buffer), 0, 0);
+            layer_surface.ack_configure(serial);
             surface.commit();
         }
     }
@@ -533,7 +658,7 @@ impl Dispatch<ZwlrLayerSurfaceV1, ()> for App {
 
 // delegate wl_registry events to App itself
 // delegate_dispatch!(App: [wl_registry::WlRegistry: GlobalListContents] => App);
-impl Dispatch<wl_registry::WlRegistry, GlobalListContents> for App {
+impl Dispatch<wl_registry::WlRegistry, GlobalListContents> for State {
     fn event(
         _state: &mut Self,
         _proxy: &wl_registry::WlRegistry,
@@ -546,17 +671,17 @@ impl Dispatch<wl_registry::WlRegistry, GlobalListContents> for App {
 }
 
 // don't emit any events
-delegate_noop!(App: wl_region::WlRegion);
-delegate_noop!(App: wl_shm_pool::WlShmPool);
-delegate_noop!(App: wl_compositor::WlCompositor);
-delegate_noop!(App: ZwlrLayerShellV1);
-delegate_noop!(App: ZwpRelativePointerManagerV1);
-delegate_noop!(App: ZwpKeyboardShortcutsInhibitManagerV1);
-delegate_noop!(App: ZwpPointerConstraintsV1);
+delegate_noop!(State: wl_region::WlRegion);
+delegate_noop!(State: wl_shm_pool::WlShmPool);
+delegate_noop!(State: wl_compositor::WlCompositor);
+delegate_noop!(State: ZwlrLayerShellV1);
+delegate_noop!(State: ZwpRelativePointerManagerV1);
+delegate_noop!(State: ZwpKeyboardShortcutsInhibitManagerV1);
+delegate_noop!(State: ZwpPointerConstraintsV1);
 
 // ignore events
-delegate_noop!(App: ignore wl_shm::WlShm);
-delegate_noop!(App: ignore wl_buffer::WlBuffer);
-delegate_noop!(App: ignore wl_surface::WlSurface);
-delegate_noop!(App: ignore ZwpKeyboardShortcutsInhibitorV1);
-delegate_noop!(App: ignore ZwpLockedPointerV1);
+delegate_noop!(State: ignore wl_shm::WlShm);
+delegate_noop!(State: ignore wl_buffer::WlBuffer);
+delegate_noop!(State: ignore wl_surface::WlSurface);
+delegate_noop!(State: ignore ZwpKeyboardShortcutsInhibitorV1);
+delegate_noop!(State: ignore ZwpLockedPointerV1);

src/backend/producer/windows.rs

@@ -1,11 +1,58 @@
-use std::sync::mpsc::SyncSender;
+use std::vec::Drain;
+
+use mio::{Token, Registry};
+use mio::event::Source;
+use std::io::Result;
 
 use crate::{
-    client::{Client, ClientHandle},
+    client::{ClientHandle, ClientEvent},
     event::Event,
-    request::Server,
+    producer::EventProducer,
 };
 
-pub fn run(_produce_tx: SyncSender<(Event, ClientHandle)>, _server: Server, _clients: Vec<Client>) {
-    todo!();
+pub struct WindowsProducer {
+    pending_events: Vec<(ClientHandle, Event)>,
+}
+
+impl Source for WindowsProducer {
+    fn register(
+        &mut self,
+        _registry: &Registry,
+        _token: Token,
+        _interests: mio::Interest,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    fn reregister(
+        &mut self,
+        _registry: &Registry,
+        _token: Token,
+        _interests: mio::Interest,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    fn deregister(&mut self, _registry: &Registry) -> Result<()> {
+        Ok(())
+    }
+}
+
+
+impl EventProducer for WindowsProducer {
+    fn notify(&mut self, _: ClientEvent) { }
+
+    fn read_events(&mut self) -> Drain<(ClientHandle, Event)> {
+        self.pending_events.drain(..)
+    }
+
+    fn release(&mut self) { }
+}
+
+impl WindowsProducer {
+    pub(crate) fn new() -> Self {
+        Self {
+            pending_events: vec![],
+        }
+    }
 }

src/backend/producer/x11.rs

@@ -1,9 +1,55 @@
-use std::sync::mpsc::SyncSender;
+use std::vec::Drain;
 
-use crate::client::Client;
-use crate::event::Event;
-use crate::request::Server;
+use mio::{Token, Registry};
+use mio::event::Source;
+use std::io::Result;
 
-pub fn run(_produce_tx: SyncSender<(Event, u32)>, _request_server: Server, _clients: Vec<Client>) {
-    todo!()
+use crate::producer::EventProducer;
+
+use crate::{client::{ClientHandle, ClientEvent}, event::Event};
+
+pub struct X11Producer {
+    pending_events: Vec<(ClientHandle, Event)>,
+}
+
+impl X11Producer {
+    pub fn new() -> Self {
+        Self {
+            pending_events: vec![],
+        }
+    }
+}
+
+impl Source for X11Producer {
+    fn register(
+        &mut self,
+        _registry: &Registry,
+        _token: Token,
+        _interests: mio::Interest,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    fn reregister(
+        &mut self,
+        _registry: &Registry,
+        _token: Token,
+        _interests: mio::Interest,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    fn deregister(&mut self, _registry: &Registry) -> Result<()> {
+        Ok(())
+    }
+}
+
+impl EventProducer for X11Producer {
+    fn notify(&mut self, _: ClientEvent) { }
+
+    fn read_events(&mut self) -> Drain<(ClientHandle, Event)> {
+        self.pending_events.drain(..)
+    }
+
+    fn release(&mut self) {}
 }

src/client.rs

@@ -1,106 +1,190 @@
-use std::{net::SocketAddr, error::Error, fmt::Display, sync::{Arc, atomic::{AtomicBool, Ordering, AtomicU32}, RwLock}};
+use std::{net::SocketAddr, collections::{HashSet, hash_set::Iter}, fmt::Display, time::{Instant, Duration}, iter::Cloned};
 
-use crate::{config::{self, DEFAULT_PORT}, dns};
+use serde::{Serialize, Deserialize};
 
-#[derive(Eq, Hash, PartialEq, Clone, Copy)]
+#[derive(Debug, Eq, Hash, PartialEq, Clone, Copy, Serialize, Deserialize)]
 pub enum Position {
     Left,
     Right,
     Top,
     Bottom,
 }
-
-#[derive(Clone, Copy)]
-pub struct Client {
-    pub addr: SocketAddr,
-    pub pos: Position,
-    pub handle: ClientHandle,
-}
-
-impl Client {
-    pub fn handle(&self) -> ClientHandle {
-        return self.handle;
+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",
+        })
     }
 }
 
-pub enum ClientEvent {
-    Create(Client),
-    Destroy(Client),
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
+pub struct Client {
+    /// handle to refer to the client.
+    /// This way any event consumer / producer backend does not
+    /// need to know anything about a client other than its handle.
+    pub handle: ClientHandle,
+    /// `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>,
+    /// all socket 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 addrs: HashSet<SocketAddr>,
+    /// position of a client on screen
+    pub pos: Position,
 }
 
-pub struct ClientManager {
-    next_id: AtomicU32,
-    clients: RwLock<Vec<Client>>,
-    subscribers: RwLock<Vec<Arc<AtomicBool>>>,
+pub enum ClientEvent {
+    Create(ClientHandle, Position),
+    Destroy(ClientHandle),
+    UpdatePos(ClientHandle, Position),
+    AddAddr(ClientHandle, SocketAddr),
+    RemoveAddr(ClientHandle, SocketAddr),
 }
 
 pub type ClientHandle = u32;
 
-#[derive(Debug)]
-struct ClientConfigError;
-
-impl Display for ClientConfigError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "neither ip nor hostname specified")
-    }
+pub struct ClientManager {
+    /// probably not beneficial to use a hashmap here
+    clients: Vec<Client>,
+    last_ping: Vec<(ClientHandle, Option<Instant>)>,
+    last_seen: Vec<(ClientHandle, Option<Instant>)>,
+    last_replied:  Vec<(ClientHandle, Option<Instant>)>,
+    next_client_id: u32,
 }
 
-impl Error for ClientConfigError {}
-
 impl ClientManager {
-    fn add_client(&self, client: &config::Client, pos: Position) -> Result<(), Box<dyn Error>> {
-        let ip = match client.ip {
-            Some(ip) => ip,
-            None => match &client.host_name {
-                Some(host_name) => dns::resolve(host_name)?,
-                None => return Err(Box::new(ClientConfigError{})),
-            },
-        };
-        let addr = SocketAddr::new(ip, client.port.unwrap_or(DEFAULT_PORT));
-        self.register_client(addr, pos);
-        Ok(())
-    }
-
-    fn notify(&self) {
-        for subscriber in self.subscribers.read().unwrap().iter() {
-            subscriber.store(true, Ordering::SeqCst);
+    pub fn new() -> Self {
+        Self {
+            clients: vec![],
+            next_client_id: 0,
+            last_ping: vec![],
+            last_seen: vec![],
+            last_replied: vec![],
         }
     }
 
-    fn new_id(&self) -> ClientHandle {
-        let id = self.next_id.load(Ordering::Acquire);
-        self.next_id.store(id + 1, Ordering::Release);
-        id as ClientHandle
+    /// add a new client to this manager
+    pub fn add_client(&mut self, addrs: HashSet<SocketAddr>, pos: Position) -> ClientHandle {
+        let handle = self.next_id();
+        // we dont know, which IP is initially active
+        let active_addr = None;
+
+        // store the client
+        let client = Client { handle, active_addr, addrs, pos };
+        self.clients.push(client);
+        self.last_ping.push((handle, None));
+        self.last_seen.push((handle, None));
+        self.last_replied.push((handle, None));
+        handle
     }
 
-    pub fn new(config: &config::Config) -> Result<Self, Box<dyn Error>> {
-
-        let client_manager = ClientManager {
-            next_id: AtomicU32::new(0),
-            clients: RwLock::new(Vec::new()),
-            subscribers: RwLock::new(vec![]),
-        };
-
-        // add clients from config
-        for (client, pos) in config.clients.iter() {
-            client_manager.add_client(&client, *pos)?;
+    /// add a socket address to the given client
+    pub fn add_addr(&mut self, client: ClientHandle, addr: SocketAddr) {
+        if let Some(client) = self.get_mut(client) {
+            client.addrs.insert(addr);
         }
-
-        Ok(client_manager)
     }
 
-    pub fn register_client(&self, addr: SocketAddr, pos: Position) {
-        let handle = self.new_id();
-        let client = Client { addr, pos, handle };
-        self.clients.write().unwrap().push(client);
-        self.notify();
+    /// remove socket address from the given client
+    pub fn remove_addr(&mut self, client: ClientHandle, addr: SocketAddr) {
+        if let Some(client) = self.get_mut(client) {
+            client.addrs.remove(&addr);
+        }
     }
 
-    pub fn get_clients(&self) -> Vec<Client> {
-        self.clients.read().unwrap().clone()
+    pub fn set_default_addr(&mut self, client: ClientHandle, addr: SocketAddr) {
+        if let Some(client) = self.get_mut(client) {
+            client.active_addr = Some(addr)
+        }
     }
 
-    pub fn subscribe(&self, subscriber: Arc<AtomicBool>) {
-        self.subscribers.write().unwrap().push(subscriber);
+    /// update the position of a client
+    pub fn update_pos(&mut self, client: ClientHandle, pos: Position) {
+        if let Some(client) = self.get_mut(client) {
+            client.pos = pos;
+        }
+    }
+
+    pub fn get_active_addr(&self, client: ClientHandle) -> Option<SocketAddr> {
+        self.get(client)?.active_addr
+    }
+
+    pub fn get_addrs(&self, client: ClientHandle) -> Option<Cloned<Iter<'_, SocketAddr>>> {
+        Some(self.get(client)?.addrs.iter().cloned())
+    }
+
+    pub fn last_ping(&self, client: ClientHandle) -> Option<Duration> {
+        let last_ping = self.last_ping
+            .iter()
+            .find(|(c,_)| *c == client)
+            .unwrap().1;
+        last_ping.map(|p| p.elapsed())
+    }
+
+    pub fn last_seen(&self, client: ClientHandle) -> Option<Duration> {
+        let last_seen = self.last_seen
+            .iter()
+            .find(|(c, _)| *c == client)
+            .unwrap().1;
+        last_seen.map(|t| t.elapsed())
+    }
+
+    pub fn last_replied(&self, client: ClientHandle) -> Option<Duration> {
+        let last_replied = self.last_replied
+            .iter()
+            .find(|(c, _)| *c == client)
+            .unwrap().1;
+        last_replied.map(|t| t.elapsed())
+    }
+
+    pub fn reset_last_ping(&mut self, client: ClientHandle) {
+        self.last_ping
+            .iter_mut()
+            .find(|(c, _)| *c == client)
+            .unwrap().1 = Some(Instant::now());
+    }
+
+    pub fn reset_last_seen(&mut self, client: ClientHandle) {
+        self.last_seen
+            .iter_mut()
+            .find(|(c, _)| *c == client)
+            .unwrap().1 = Some(Instant::now());
+    }
+
+    pub fn reset_last_replied(&mut self, client: ClientHandle) {
+        self.last_replied
+            .iter_mut()
+            .find(|(c, _)| *c == client)
+            .unwrap().1 = Some(Instant::now());
+    }
+
+    pub fn get_client(&self, addr: SocketAddr) -> Option<ClientHandle> {
+        self.clients
+            .iter()
+            .find(|c| c.addrs.contains(&addr))
+            .map(|c| c.handle)
+    }
+
+    fn next_id(&mut self) -> ClientHandle {
+        let handle = self.next_client_id;
+        self.next_client_id += 1;
+        handle
+    }
+
+    fn get<'a>(&'a self, client: ClientHandle) -> Option<&'a Client> {
+        self.clients
+            .iter()
+            .find(|c| c.handle == client)
+    }
+
+    fn get_mut<'a>(&'a mut self, client: ClientHandle) -> Option<&'a mut Client> {
+        self.clients
+            .iter_mut()
+            .find(|c| c.handle == client)
     }
 }

src/config.rs

@@ -1,19 +1,21 @@
-use serde_derive::{Deserialize, Serialize};
+use serde::{Deserialize, Serialize};
 use core::fmt;
-use std::net::IpAddr;
+use std::collections::HashSet;
+use std::net::{IpAddr, SocketAddr};
 use std::{error::Error, fs};
 
 use std::env;
 use toml;
 
 use crate::client::Position;
+use crate::dns;
 
 pub const DEFAULT_PORT: u16 = 4242;
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ConfigToml {
     pub port: Option<u16>,
-    pub backend: Option<String>,
+    pub frontend: Option<String>,
     pub left: Option<Client>,
     pub right: Option<Client>,
     pub top: Option<Client>,
@@ -23,7 +25,7 @@ pub struct ConfigToml {
 #[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
 pub struct Client {
     pub host_name: Option<String>,
-    pub ip: Option<IpAddr>,
+    pub ips: Option<Vec<IpAddr>>,
     pub port: Option<u16>,
 }
 
@@ -62,8 +64,13 @@ fn find_arg(key: &'static str) -> Result<Option<String>, MissingParameter> {
     Ok(None)
 }
 
+pub enum Frontend {
+    Gtk,
+    Cli,
+}
+
 pub struct Config {
-    pub backend: Option<String>,
+    pub frontend: Frontend,
     pub port: u16,
     pub clients: Vec<(Client, Position)>,
 }
@@ -73,19 +80,28 @@ impl Config {
         let config_path = "config.toml";
         let config_toml = match ConfigToml::new(config_path) {
             Err(e) => {
-                eprintln!("config.toml: {e}");
-                eprintln!("Continuing without config file ...");
+                log::error!("config.toml: {e}");
+                log::warn!("Continuing without config file ...");
                 None
             },
             Ok(c) => Some(c),
         };
 
-        let backend = match find_arg("--backend")? {
+        let frontend = match find_arg("--frontend")? {
             None => match &config_toml {
-                Some(c) => c.backend.clone(),
+                Some(c) => c.frontend.clone(),
                 None => None,
             },
-            backend => backend,
+            frontend => frontend,
+        };
+
+        let frontend = match frontend {
+            None => Frontend::Cli,
+            Some(s) => match s.as_str() {
+                "cli" => Frontend::Cli,
+                "gtk" => Frontend::Gtk,
+                    _ => Frontend::Cli,
+            }
         };
 
         let port = match find_arg("--port")? {
@@ -113,6 +129,43 @@ impl Config {
             }
         }
 
-        Ok(Config { backend, clients, port })
+        Ok(Config { frontend, clients, port })
+    }
+
+    pub fn get_clients(&self) -> Vec<(HashSet<SocketAddr>, Option<String>,  Position)> {
+        self.clients.iter().map(|(c,p)| {
+            let port = c.port.unwrap_or(DEFAULT_PORT);
+            // add ips from config
+            let config_ips: Vec<IpAddr> = if let Some(ips) = c.ips.as_ref() {
+                ips.iter().cloned().collect()
+            } else {
+                vec![]
+            };
+            let host_name = c.host_name.clone();
+            // add ips from dns lookup
+            let dns_ips = match host_name.as_ref() {
+                None => vec![],
+                Some(host_name) => match dns::resolve(host_name) {
+                    Err(e) => {
+                        log::warn!("{host_name}: could not resolve host: {e}");
+                        vec![]
+                    }
+                    Ok(l) if l.is_empty() => {
+                        log::warn!("{host_name}: could not resolve host");
+                        vec![]
+                    }
+                    Ok(l) => l,
+                }
+            };
+            if config_ips.is_empty() && dns_ips.is_empty() {
+                log::error!("no ips found for client {p:?}, ignoring!");
+                log::error!("You can manually specify ip addresses via the `ips` config option");
+            }
+            let ips = config_ips.into_iter().chain(dns_ips.into_iter());
+
+            // map ip addresses to socket addresses
+            let addrs: HashSet<SocketAddr> = ips.map(|ip| SocketAddr::new(ip, port)).collect();
+            (addrs, host_name, *p)
+        }).filter(|(a, _, _)| !a.is_empty()).collect()
     }
 }

src/consumer.rs

@@ -1,9 +1,8 @@
-use std::{thread::{JoinHandle, self}, sync::mpsc::Receiver, error::Error};
-
 #[cfg(unix)]
 use std::env;
 
-use crate::{backend::consumer, client::{Client, ClientHandle}, event::Event};
+use anyhow::Result;
+use crate::{backend::consumer, client::{ClientHandle, ClientEvent}, event::Event};
 
 #[cfg(unix)]
 #[derive(Debug)]
@@ -14,63 +13,87 @@ enum Backend {
     Libei,
 }
 
-pub fn start(consume_rx: Receiver<(Event, ClientHandle)>, clients: Vec<Client>, backend: Option<String>) -> Result<JoinHandle<()>, Box<dyn Error>> {
+pub trait EventConsumer {
+    /// Event corresponding to an abstract `client_handle`
+    fn consume(&self, event: Event, client_handle: ClientHandle);
+
+    /// Event corresponding to a configuration change
+    fn notify(&mut self, client_event: ClientEvent);
+}
+
+pub fn create() -> Result<Box<dyn EventConsumer>> {
     #[cfg(windows)]
-    let _backend = backend;
+    return Ok(Box::new(consumer::windows::WindowsConsumer::new()));
 
-    Ok(thread::Builder::new()
-        .name("event consumer".into())
-        .spawn(move || {
-            #[cfg(windows)]
-            consumer::windows::run(consume_rx, clients);
-
-            #[cfg(unix)]
-            let backend = match env::var("XDG_SESSION_TYPE") {
-                Ok(session_type) => match session_type.as_str() {
-                    "x11" => Backend::X11,
-                    "wayland" => {
-                        match backend {
-                            Some(backend) => match backend.as_str() {
-                                "wlroots" => Backend::Wlroots,
-                                "libei" => Backend::Libei,
-                                "xdg_desktop_portal" => Backend::RemoteDesktopPortal,
-                                backend => panic!("invalid backend: {}", backend)
-                            }
-                            // default to wlroots backend for now
-                            _ => Backend::Wlroots,
+    #[cfg(unix)]
+    let backend = match env::var("XDG_SESSION_TYPE") {
+        Ok(session_type) => match session_type.as_str() {
+            "x11" => {
+                log::info!("XDG_SESSION_TYPE = x11 -> using x11 event consumer");
+                Backend::X11
+            }
+            "wayland" => {
+                log::info!("XDG_SESSION_TYPE = wayland -> using wayland event consumer");
+                match env::var("XDG_CURRENT_DESKTOP") {
+                    Ok(current_desktop) => match current_desktop.as_str() {
+                        "gnome" => {
+                            log::info!("XDG_CURRENT_DESKTOP = gnome -> using libei backend");
+                            Backend::Libei
+                        }
+                        "KDE" => {
+                            log::info!("XDG_CURRENT_DESKTOP = KDE -> using xdg_desktop_portal backend");
+                            Backend::RemoteDesktopPortal
+                        }
+                        "sway" => {
+                            log::info!("XDG_CURRENT_DESKTOP = sway -> using wlroots backend");
+                            Backend::Wlroots
+                        }
+                        "Hyprland" => {
+                            log::info!("XDG_CURRENT_DESKTOP = Hyprland -> using wlroots backend");
+                            Backend::Wlroots
+                        }
+                        _ => {
+                            log::warn!("unknown XDG_CURRENT_DESKTOP -> defaulting to wlroots backend");
+                            Backend::Wlroots
                         }
                     }
-                    _ => panic!("unknown XDG_SESSION_TYPE"),
-                },
-                Err(_) => panic!("could not detect session type: XDG_SESSION_TYPE environment variable not set!"),
-            };
-
-            #[cfg(unix)]
-            match backend {
-                Backend::Libei => {
-                    #[cfg(not(feature = "libei"))]
-                    panic!("feature libei not enabled");
-                    #[cfg(feature = "libei")]
-                    consumer::libei::run(consume_rx, clients);
-                },
-                Backend::RemoteDesktopPortal => {
-                    #[cfg(not(feature = "xdg_desktop_portal"))]
-                    panic!("feature xdg_desktop_portal not enabled");
-                    #[cfg(feature = "xdg_desktop_portal")]
-                    consumer::xdg_desktop_portal::run(consume_rx, clients);
-                },
-                Backend::Wlroots => {
-                    #[cfg(not(feature = "wayland"))]
-                    panic!("feature wayland not enabled");
-                    #[cfg(feature = "wayland")]
-                    consumer::wlroots::run(consume_rx, clients);
-                },
-                Backend::X11 => {
-                    #[cfg(not(feature = "x11"))]
-                    panic!("feature x11 not enabled");
-                    #[cfg(feature = "x11")]
-                    consumer::x11::run(consume_rx, clients);
-                },
+                    // default to wlroots backend for now
+                    _ => {
+                        log::warn!("unknown XDG_CURRENT_DESKTOP -> defaulting to wlroots backend");
+                        Backend::Wlroots
+                    }
+                }
             }
-        })?)
+            _ => panic!("unknown XDG_SESSION_TYPE"),
+        },
+        Err(_) => panic!("could not detect session type: XDG_SESSION_TYPE environment variable not set!"),
+    };
+
+    #[cfg(unix)]
+    match backend {
+        Backend::Libei => {
+            #[cfg(not(feature = "libei"))]
+            panic!("feature libei not enabled");
+            #[cfg(feature = "libei")]
+            Ok(Box::new(consumer::libei::LibeiConsumer::new()))
+        },
+        Backend::RemoteDesktopPortal => {
+            #[cfg(not(feature = "xdg_desktop_portal"))]
+            panic!("feature xdg_desktop_portal not enabled");
+            #[cfg(feature = "xdg_desktop_portal")]
+            Ok(Box::new(consumer::xdg_desktop_portal::DesktopPortalConsumer::new()))
+        },
+        Backend::Wlroots => {
+            #[cfg(not(feature = "wayland"))]
+            panic!("feature wayland not enabled");
+            #[cfg(feature = "wayland")]
+            Ok(Box::new(consumer::wlroots::WlrootsConsumer::new()?))
+        },
+        Backend::X11 => {
+            #[cfg(not(feature = "x11"))]
+            panic!("feature x11 not enabled");
+            #[cfg(feature = "x11")]
+            Ok(Box::new(consumer::x11::X11Consumer::new()))
+        },
+    }
 }

src/dns.rs

@@ -1,27 +1,9 @@
-use std::{error::Error, fmt::Display, net::IpAddr};
+use std::{error::Error, net::IpAddr};
 
 use trust_dns_resolver::Resolver;
 
-#[derive(Debug, Clone)]
-struct InvalidConfigError;
-
-#[derive(Debug, Clone)]
-struct DnsError {
-    host: String,
-}
-
-impl Error for DnsError {}
-
-impl Display for DnsError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "couldn't resolve host \"{}\"", self.host)
-    }
-}
-
-pub fn resolve(host: &String) -> Result<IpAddr, Box<dyn Error>> {
+pub fn resolve(host: &str) -> Result<Vec<IpAddr>, Box<dyn Error>> {
+    log::info!("resolving {host} ...");
     let response = Resolver::from_system_conf()?.lookup_ip(host)?;
-    match response.iter().next() {
-        Some(ip) => Ok(ip),
-        None => Err(DnsError { host: host.clone() }.into()),
-    }
+    Ok(response.iter().collect())
 }

src/event.rs

@@ -1,7 +1,8 @@
-use std::{error::Error, fmt};
+use std::{error::Error, fmt::{self, Display}};
 
 pub mod server;
 
+#[derive(Debug, Clone, Copy)]
 pub enum PointerEvent {
     Motion {
         time: u32,
@@ -21,6 +22,7 @@ pub enum PointerEvent {
     Frame {},
 }
 
+#[derive(Debug, Clone, Copy)]
 pub enum KeyboardEvent {
     Key {
         time: u32,
@@ -35,14 +37,46 @@ pub enum KeyboardEvent {
     },
 }
 
+#[derive(Debug, Clone, Copy)]
 pub enum Event {
     Pointer(PointerEvent),
     Keyboard(KeyboardEvent),
     Release(),
+    Ping(),
+    Pong(),
 }
 
-unsafe impl Send for Event {}
-unsafe impl Sync for Event {}
+impl Display for PointerEvent {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            PointerEvent::Motion { time: _ , relative_x, relative_y } => write!(f, "motion({relative_x},{relative_y})"),
+            PointerEvent::Button { time: _ , button, state } => write!(f, "button({button}, {state})"),
+            PointerEvent::Axis { time: _, axis, value } => write!(f, "scroll({axis}, {value})"),
+            PointerEvent::Frame { } => write!(f, "frame()"),
+        }
+    }
+}
+
+impl Display for KeyboardEvent {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            KeyboardEvent::Key { time: _, key, state } => write!(f, "key({key}, {state})"),
+            KeyboardEvent::Modifiers { mods_depressed, mods_latched, mods_locked, group } => write!(f, "modifiers({mods_depressed},{mods_latched},{mods_locked},{group})"),
+        }
+    }
+}
+
+impl Display for Event {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Event::Pointer(p) => write!(f, "{}", p),
+            Event::Keyboard(k) => write!(f, "{}", k),
+            Event::Release() => write!(f, "release"),
+            Event::Ping() => write!(f, "ping"),
+            Event::Pong() => write!(f, "pong"),
+        }
+    }
+}
 
 impl Event {
     fn event_type(&self) -> EventType {
@@ -50,6 +84,8 @@ impl Event {
             Self::Pointer(_) => EventType::POINTER,
             Self::Keyboard(_) => EventType::KEYBOARD,
             Self::Release() => EventType::RELEASE,
+            Self::Ping() => EventType::PING,
+            Self::Pong() => EventType::PONG,
         }
     }
 }
@@ -88,6 +124,8 @@ enum EventType {
     POINTER,
     KEYBOARD,
     RELEASE,
+    PING,
+    PONG,
 }
 
 impl TryFrom<u8> for PointerEventType {
@@ -127,6 +165,8 @@ impl Into<Vec<u8>> for &Event {
             Event::Pointer(p) => p.into(),
             Event::Keyboard(k) => k.into(),
             Event::Release() => vec![],
+            Event::Ping() => vec![],
+            Event::Pong() => vec![],
         };
         vec![event_id, event_data].concat()
     }
@@ -153,6 +193,8 @@ impl TryFrom<Vec<u8>> for Event {
             i if i == (EventType::POINTER as u8) => Ok(Event::Pointer(value.try_into()?)),
             i if i == (EventType::KEYBOARD as u8) => Ok(Event::Keyboard(value.try_into()?)),
             i if i == (EventType::RELEASE as u8) => Ok(Event::Release()),
+            i if i == (EventType::PING as u8) => Ok(Event::Ping()),
+            i if i == (EventType::PONG as u8) => Ok(Event::Pong()),
             _ => Err(Box::new(ProtocolError {
                 msg: format!("invalid event_id {}", event_id),
             })),

src/event/server.rs

@@ -1,164 +1,280 @@
-use anyhow::Result;
+use std::{error::Error, io::Result, collections::HashSet, time::Duration};
+use log;
+use mio::{Events, Poll, Interest, Token, net::UdpSocket};
+#[cfg(not(windows))]
+use mio_signals::{Signals, Signal, SignalSet};
 
-use std::{
-    collections::HashMap,
-    error::Error,
-    net::{SocketAddr, UdpSocket},
-    sync::{
-        atomic::{AtomicBool, Ordering},
-        mpsc::{Receiver, SyncSender},
-        Arc,
-    },
-    thread::{self, JoinHandle},
-};
-
-use crate::{client::{ClientHandle, ClientManager}, ioutils::{ask_confirmation, ask_position}};
+use std::{net::SocketAddr, io::ErrorKind};
 
+use crate::{client::{ClientEvent, ClientManager, Position}, consumer::EventConsumer, producer::EventProducer, frontend::{FrontendEvent, FrontendAdapter}};
 use super::Event;
 
-pub struct Server {
-    listen_addr: SocketAddr,
-    sending: Arc<AtomicBool>,
+/// keeps track of state to prevent a feedback loop
+/// of continuously sending and receiving the same event.
+#[derive(Eq, PartialEq)]
+enum State {
+    Sending,
+    Receiving,
 }
 
+pub struct Server {
+    poll: Poll,
+    socket: UdpSocket,
+    producer: Box<dyn EventProducer>,
+    consumer: Box<dyn EventConsumer>,
+    #[cfg(not(windows))]
+    signals: Signals,
+    frontend: FrontendAdapter,
+    client_manager: ClientManager,
+    state: State,
+}
+
+const UDP_RX: Token = Token(0);
+const FRONTEND_RX: Token = Token(1);
+const PRODUCER_RX: Token = Token(2);
+#[cfg(not(windows))]
+const SIGNAL: Token = Token(3);
+
 impl Server {
-    pub fn new(port: u16) -> Result<Self, Box<dyn Error>> {
-        let listen_addr = SocketAddr::new("0.0.0.0".parse()?, port);
-        let sending = Arc::new(AtomicBool::new(false));
+    pub fn new(
+        port: u16,
+        mut producer: Box<dyn EventProducer>,
+        consumer: Box<dyn EventConsumer>,
+        mut frontend: FrontendAdapter,
+    ) -> Result<Self> {
+        // bind the udp socket
+        let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), port);
+        let mut socket = UdpSocket::bind(listen_addr)?;
+
+        // register event sources
+        let poll = Poll::new()?;
+
+        // hand signal handling over to the event loop
+        #[cfg(not(windows))]
+        let mut signals = Signals::new(SignalSet::all())?;
+
+        #[cfg(not(windows))]
+        poll.registry().register(&mut signals, SIGNAL, Interest::READABLE)?;
+        poll.registry().register(&mut socket, UDP_RX, Interest::READABLE | Interest::WRITABLE)?;
+        poll.registry().register(&mut producer, PRODUCER_RX, Interest::READABLE)?;
+        poll.registry().register(&mut frontend, FRONTEND_RX, Interest::READABLE)?;
+
+        // create client manager
+        let client_manager = ClientManager::new();
         Ok(Server {
-            listen_addr,
-            sending,
+            poll, socket, consumer, producer,
+            #[cfg(not(windows))]
+            signals, frontend,
+            client_manager,
+            state: State::Receiving,
         })
     }
 
-    pub fn run(
-        &self,
-        client_manager: Arc<ClientManager>,
-        produce_rx: Receiver<(Event, ClientHandle)>,
-        consume_tx: SyncSender<(Event, ClientHandle)>,
-    ) -> Result<(JoinHandle<Result<()>>, JoinHandle<Result<()>>), Box<dyn Error>> {
-        let udp_socket = UdpSocket::bind(self.listen_addr)?;
-        let rx = udp_socket.try_clone()?;
-        let tx = udp_socket;
+    pub fn run(&mut self) -> Result<()> {
+        let mut events = Events::with_capacity(10);
+        loop {
+            match self.poll.poll(&mut events, None) {
+                Ok(()) => (),
+                Err(e) if e.kind() == ErrorKind::Interrupted => continue,
+                Err(e) => return Err(e),
+            }
+            for event in &events {
+                if !event.is_readable() { continue }
+                match event.token() {
+                    UDP_RX => self.handle_udp_rx(),
+                    PRODUCER_RX => self.handle_producer_rx(),
+                    FRONTEND_RX => if self.handle_frontend_rx() { return Ok(()) },
+                    #[cfg(not(windows))]
+                    SIGNAL => if self.handle_signal() { return Ok(()) },
+                    _ => panic!("what happened here?")
+                }
+            }
+        }
+    }
 
-        let sending = self.sending.clone();
-        let clients_updated = Arc::new(AtomicBool::new(true));
-        client_manager.subscribe(clients_updated.clone());
-        let client_manager_clone = client_manager.clone();
+    pub fn add_client(&mut self, addr: HashSet<SocketAddr>, pos: Position) {
+        let client = self.client_manager.add_client(addr, pos);
+        self.producer.notify(ClientEvent::Create(client, pos));
+        self.consumer.notify(ClientEvent::Create(client, pos));
+    }
 
-        let receiver = thread::Builder::new()
-            .name("event receiver".into())
-            .spawn(move || {
-                let mut client_for_socket = HashMap::new();
-
-                loop {
-                    let (event, addr) = match Server::receive_event(&rx) {
-                        Ok(e) => e,
-                        Err(e) => {
-                            eprintln!("{}", e);
-                            continue;
-                        }
-                    };
-
-                    if let Ok(_) = clients_updated.compare_exchange(
-                        true,
-                        false,
-                        Ordering::SeqCst,
-                        Ordering::SeqCst,
-                    ) {
-                        clients_updated.store(false, Ordering::SeqCst);
-                        client_for_socket.clear();
-                        println!("updating clients: ");
-                        for client in client_manager_clone.get_clients() {
-                            println!("{}: {}", client.handle, client.addr);
-                            client_for_socket.insert(client.addr, client.handle);
+    fn handle_udp_rx(&mut self) {
+        loop {
+            let (event, addr) = match self.receive_event() {
+                Ok(e) => e,
+                Err(e) => {
+                    if e.is::<std::io::Error>() {
+                        if let ErrorKind::WouldBlock = e.downcast_ref::<std::io::Error>()
+                            .unwrap()
+                            .kind() {
+                            return
                         }
                     }
+                    log::error!("{}", e);
+                    continue
+                }
+            };
+            log::trace!("{:20} <-<-<-<------ {addr}", event.to_string());
 
-                    let client_handle = match client_for_socket.get(&addr) {
-                        Some(c) => *c,
-                        None => {
-                            eprint!("Allow connection from {:?}? ", addr);
-                            if ask_confirmation(false)? {
-                                client_manager_clone.register_client(addr, ask_position()?);
-                            } else {
-                                eprintln!("rejecting client: {:?}?", addr);
+            // get handle for addr
+            let handle = match self.client_manager.get_client(addr) {
+                Some(a) => a,
+                None => {
+                    log::warn!("ignoring event from client {addr:?}");
+                    continue
+                }
+            };
+
+            // reset ttl for client and set addr as new default for this client
+            self.client_manager.reset_last_seen(handle);
+            self.client_manager.set_default_addr(handle, addr);
+            match (event, addr) {
+                (Event::Pong(), _) => {},
+                (Event::Ping(), addr) => {
+                    if let Err(e) = Self::send_event(&self.socket, Event::Pong(), addr) {
+                        log::error!("udp send: {}", e);
+                    }
+                }
+                (event, addr) => {
+                    match self.state {
+                        State::Sending => {
+                            // in sending state, we dont want to process
+                            // any events to avoid feedback loops,
+                            // therefore we tell the event producer
+                            // to release the pointer and move on
+                            // first event -> release pointer
+                            if let Event::Release() = event {
+                                log::debug!("releasing pointer ...");
+                                self.producer.release();
+                                self.state = State::Receiving;
                             }
-                            continue;
                         }
-                    };
+                        State::Receiving => {
+                            // consume event
+                            self.consumer.consume(event, handle);
 
-                    // There is a race condition between loading this
-                    // value and handling the event:
-                    // In the meantime a event could be produced, which
-                    // should theoretically disable receiving of events.
-                    //
-                    // This is however not a huge problem, as some
-                    // events that make it through are not a large problem
-                    if sending.load(Ordering::Acquire) {
-                        // ignore received events when in sending state
-                        // if release event is received, switch state to receiving
-                        if let Event::Release() = event {
-                            sending.store(false, Ordering::Release);
-                            consume_tx
-                                .send((event, client_handle))
-                                .expect("event consumer unavailable");
+                            // let the server know we are still alive once every second
+                            let last_replied = self.client_manager.last_replied(handle);
+                            if  last_replied.is_none() 
+                            || last_replied.is_some() && last_replied.unwrap() > Duration::from_secs(1) {
+                                self.client_manager.reset_last_replied(handle);
+                                if let Err(e) = Self::send_event(&self.socket, Event::Pong(), addr) {
+                                    log::error!("udp send: {}", e);
+                                }
+                            }
                         }
-                    } else {
-                        if let Event::Release() = event {
-                            sending.store(false, Ordering::Release);
-                        }
-                        // we retrieve all events
-                        consume_tx
-                            .send((event, client_handle))
-                            .expect("event consumer unavailable");
                     }
                 }
-            })?;
-
-        let sending = self.sending.clone();
-
-        let mut socket_for_client = HashMap::new();
-        for client in client_manager.get_clients() {
-            socket_for_client.insert(client.handle, client.addr);
+            }
         }
-        let sender = thread::Builder::new()
-            .name("event sender".into())
-            .spawn(move || {
-                loop {
-                    let (event, client_handle) =
-                        produce_rx.recv().expect("event producer unavailable");
-                    let addr = match socket_for_client.get(&client_handle) {
-                        Some(addr) => addr,
-                        None => continue,
-                    };
-
-                    if sending.load(Ordering::Acquire) {
-                        Server::send_event(&tx, event, *addr);
-                    } else {
-                        // only accept enter event
-                        if let Event::Release() = event {
-                            // set state to sending, to ignore incoming events
-                            // and enable sending of events
-                            sending.store(true, Ordering::Release);
-                            Server::send_event(&tx, event, *addr);
-                        }
-                    }
-                }
-            })?;
-        Ok((receiver, sender))
     }
 
-    fn send_event(tx: &UdpSocket, e: Event, addr: SocketAddr) {
+    fn handle_producer_rx(&mut self) {
+        let events = self.producer.read_events();
+        for (c, e) in events.into_iter() {
+            // in receiving state, only release events
+            // must be transmitted
+            if let Event::Release() = e {
+                self.state = State::Sending;
+            }
+            // otherwise we should have an address to send to
+            // transmit events to the corrensponding client
+            if let Some(addr) = self.client_manager.get_active_addr(c) {
+                log::trace!("{:20} ------>->->-> {addr}", e.to_string());
+                if let Err(e) = Self::send_event(&self.socket, e, addr) {
+                    log::error!("udp send: {}", e);
+                }
+            }
+
+            // if client last responded > 2 seconds ago
+            // and we have not sent a ping since 500 milliseconds,
+            // send a ping
+            let last_seen = self.client_manager.last_seen(c);
+            let last_ping = self.client_manager.last_ping(c);
+            if last_seen.is_some() && last_seen.unwrap() < Duration::from_secs(2) {
+                continue
+            }
+
+            // client last seen > 500ms ago
+            if last_ping.is_some() && last_ping.unwrap() < Duration::from_millis(500) {
+                continue
+            }
+
+            // last ping > 500ms ago -> ping all interfaces
+            self.client_manager.reset_last_ping(c);
+            if let Some(iter) = self.client_manager.get_addrs(c) {
+                for addr in iter {
+                    log::debug!("pinging {addr}");
+                    if let Err(e) = Self::send_event(&self.socket, Event::Ping(), addr) {
+                        if e.kind() != ErrorKind::WouldBlock {
+                            log::error!("udp send: {}", e);
+                        }
+                    }
+                }
+            } else {
+                // TODO should repeat dns lookup
+            }
+        }
+    }
+
+    fn handle_frontend_rx(&mut self) -> bool {
+        loop {
+            match self.frontend.read_event() {
+                Ok(event) => match event {
+                    FrontendEvent::RequestPortChange(_) => todo!(),
+                    FrontendEvent::RequestClientAdd(addr, pos) => {
+                        self.add_client(HashSet::from_iter(&mut [addr].into_iter()), pos);
+                    }
+                    FrontendEvent::RequestClientDelete(_) => todo!(),
+                    FrontendEvent::RequestClientUpdate(_) => todo!(),
+                    FrontendEvent::RequestShutdown() => {
+                        log::info!("terminating gracefully...");
+                        return true;
+                    },
+                }
+                Err(e) if e.kind() == ErrorKind::WouldBlock => return false,
+                Err(e) => {
+                    log::error!("frontend: {e}");
+                }
+            }
+        }
+    }
+
+    #[cfg(not(windows))]
+    fn handle_signal(&mut self) -> bool {
+        #[cfg(windows)]
+        return false;
+        #[cfg(not(windows))]
+        loop {
+            match self.signals.receive() {
+                Err(e) if e.kind() == ErrorKind::WouldBlock => return false,
+                Err(e) => {
+                    log::error!("error reading signal: {e}");
+                    return false;
+                }
+                Ok(Some(Signal::Interrupt) | Some(Signal::Terminate)) => {
+                    // terminate on SIG_INT or SIG_TERM
+                    log::info!("terminating gracefully...");
+                    return true;
+                },
+                Ok(Some(signal)) => {
+                    log::info!("ignoring signal {signal:?}");
+                },
+                Ok(None) => return false,
+            }
+        }
+    }
+
+    fn send_event(sock: &UdpSocket, e: Event, addr: SocketAddr) -> Result<usize> {
         let data: Vec<u8> = (&e).into();
-        if let Err(e) = tx.send_to(&data[..], addr) {
-            eprintln!("{}", e);
-        }
+        // We are currently abusing a blocking send to get the lowest possible latency.
+        // It may be better to set the socket to non-blocking and only send when ready.
+        sock.send_to(&data[..], addr)
     }
 
-    fn receive_event(rx: &UdpSocket) -> Result<(Event, SocketAddr), Box<dyn Error>> {
+    fn receive_event(&self) -> std::result::Result<(Event, SocketAddr), Box<dyn Error>> {
         let mut buf = vec![0u8; 22];
-        match rx.recv_from(&mut buf) {
+        match self.socket.recv_from(&mut buf) {
             Ok((_amt, src)) => Ok((Event::try_from(buf)?, src)),
             Err(e) => Err(Box::new(e)),
         }

src/frontend.rs

@@ -0,0 +1,115 @@
+use std::io::{Read, Result};
+use std::{str, net::SocketAddr};
+
+#[cfg(unix)]
+use std::{env, path::{Path, PathBuf}};
+
+use mio::{Registry, Token, event::Source};
+
+#[cfg(unix)]
+use mio::net::UnixListener;
+#[cfg(windows)]
+use mio::net::TcpListener;
+
+use serde::{Serialize, Deserialize};
+
+use crate::client::{Client, Position};
+
+/// cli frontend
+pub mod cli;
+
+/// gtk frontend
+#[cfg(all(unix, feature = "gtk"))]
+pub mod gtk;
+
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
+pub enum FrontendEvent {
+    RequestPortChange(u16),
+    RequestClientAdd(SocketAddr, Position),
+    RequestClientDelete(Client),
+    RequestClientUpdate(Client),
+    RequestShutdown(),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum FrontendNotify {
+    NotifyClientCreate(Client),
+    NotifyError(String),
+}
+
+pub struct FrontendAdapter {
+    #[cfg(windows)]
+    listener: TcpListener,
+    #[cfg(unix)]
+    listener: UnixListener,
+    #[cfg(unix)]
+    socket_path: PathBuf,
+}
+
+impl FrontendAdapter {
+    pub fn new() -> std::result::Result<Self, Box<dyn std::error::Error>> {
+        #[cfg(unix)]
+        let socket_path = Path::new(env::var("XDG_RUNTIME_DIR")?.as_str()).join("lan-mouse-socket.sock");
+        #[cfg(unix)]
+        let listener = UnixListener::bind(&socket_path)?;
+
+        #[cfg(windows)]
+        let listener = TcpListener::bind("127.0.0.1:5252".parse().unwrap())?; // abuse tcp
+
+        let adapter = Self {
+            listener,
+            #[cfg(unix)]
+            socket_path,
+        };
+
+        Ok(adapter)
+    }
+
+    pub fn read_event(&mut self) -> Result<FrontendEvent>{
+        let (mut stream, _) = self.listener.accept()?;
+        let mut buf = [0u8; 128]; // FIXME
+        stream.read(&mut buf)?;
+        let json = str::from_utf8(&buf)
+            .unwrap()
+            .trim_end_matches(char::from(0)); // remove trailing 0-bytes
+        let event = serde_json::from_str(json).unwrap();
+        log::debug!("{:?}", event);
+        Ok(event)
+    }
+
+    pub fn notify(&self, _event: FrontendNotify) { }
+}
+
+impl Source for FrontendAdapter {
+    fn register(
+        &mut self,
+        registry: &Registry,
+        token: Token,
+        interests: mio::Interest,
+    ) -> Result<()> {
+        self.listener.register(registry, token, interests)
+    }
+
+    fn reregister(
+        &mut self,
+        registry: &Registry,
+        token: Token,
+        interests: mio::Interest,
+    ) -> Result<()> {
+        self.listener.reregister(registry, token, interests)
+    }
+
+    fn deregister(&mut self, registry: &Registry) -> Result<()> {
+        self.listener.deregister(registry)
+    }
+}
+
+#[cfg(unix)]
+impl Drop for FrontendAdapter {
+    fn drop(&mut self) {
+        log::debug!("remove socket: {:?}", self.socket_path);
+        std::fs::remove_file(&self.socket_path).unwrap();
+    }
+}
+
+pub trait Frontend { }

src/frontend/cli.rs

@@ -0,0 +1,85 @@
+use anyhow::Result;
+use std::{thread, io::Write, net::SocketAddr};
+#[cfg(windows)]
+use std::net::SocketAddrV4;
+
+#[cfg(unix)]
+use std::{os::unix::net::UnixStream, path::Path, env};
+#[cfg(windows)]
+use std::net::TcpStream;
+
+use crate::client::Position;
+
+use super::{FrontendEvent, Frontend};
+
+pub struct CliFrontend;
+
+impl Frontend for CliFrontend {}
+
+impl CliFrontend {
+    pub fn new() -> Result<CliFrontend> {
+        #[cfg(unix)]
+        let socket_path = Path::new(env::var("XDG_RUNTIME_DIR")?.as_str()).join("lan-mouse-socket.sock");
+        thread::Builder::new()
+            .name("cli-frontend".to_string())
+            .spawn(move || {
+            loop {
+                eprint!("lan-mouse > ");
+                std::io::stderr().flush().unwrap();
+                let mut buf = String::new();
+                match std::io::stdin().read_line(&mut buf) {
+                    Ok(len) => {
+                        if let Some(event) = parse_event(buf, len) {
+                            #[cfg(unix)]
+                            let Ok(mut stream) = UnixStream::connect(&socket_path) else {
+                                log::error!("Could not connect to lan-mouse-socket");
+                                continue;
+                            };
+                            #[cfg(windows)]
+                            let Ok(mut stream) = TcpStream::connect("127.0.0.1:5252".parse::<SocketAddrV4>().unwrap()) else {
+                                log::error!("Could not connect to lan-mouse-server");
+                                continue;
+                            };
+                            let json = serde_json::to_string(&event).unwrap();
+                            if let Err(e) = stream.write(json.as_bytes()) {
+                                log::error!("error sending message: {e}");
+                            };
+                            if event == FrontendEvent::RequestShutdown() {
+                                break;
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        log::error!("{e:?}");
+                        break
+                    }
+                }
+            }
+        }).unwrap();
+        Ok(Self {})
+    }
+}
+
+fn parse_event(s: String, len: usize) -> Option<FrontendEvent> {
+    if len == 0 {
+        return Some(FrontendEvent::RequestShutdown())
+    }
+    let mut l = s.split_whitespace();
+    let cmd = l.next()?;
+    match cmd {
+        "connect" => {
+            let addr = match l.next()?.parse() {
+                Ok(addr) => SocketAddr::V4(addr),
+                Err(e) => {
+                    log::error!("parse error: {e}");
+                    return None;
+                }
+            };
+            Some(FrontendEvent::RequestClientAdd(addr, Position::Left ))
+        }
+        _ => {
+            log::error!("unknown command: {s}");
+            None
+        }
+    }
+}

src/lib.rs

@@ -2,10 +2,10 @@ pub mod client;
 pub mod config;
 pub mod dns;
 pub mod event;
-pub mod request;
 
 pub mod consumer;
 pub mod producer;
 
 pub mod backend;
+pub mod frontend;
 pub mod ioutils;

src/main.rs

@@ -1,111 +1,65 @@
-use std::{sync::{mpsc, Arc}, process, env};
+use std::{process, error::Error};
 
+use env_logger::Env;
 use lan_mouse::{
-    client::ClientManager,
     consumer, producer,
-    config, event, request,
+    config::{Config, Frontend::{Gtk, Cli}}, event::server::Server,
+    frontend::{FrontendAdapter, cli::CliFrontend},
 };
 
-fn usage() {
-    eprintln!("usage: {} [--backend <backend>] [--port <port>]",
-        env::args().next().unwrap_or("lan-mouse".into()));
+pub fn main() {
+
+    // init logging
+    let env = Env::default().filter_or("LAN_MOUSE_LOG_LEVEL", "info");
+    env_logger::init_from_env(env);
+
+    if let Err(e) = run() {
+        log::error!("{e}");
+        process::exit(1);
+    }
 }
 
-pub fn main() {
+pub fn run() -> Result<(), Box<dyn Error>> {
     // parse config file
-    let config = match config::Config::new() {
-        Err(e) => {
-            eprintln!("{e}");
-            usage();
-            process::exit(1);
-        }
-        Ok(config) => config,
-    };
-
-    // port or default
-    let port = config.port;
-
-    // event channel for producing events
-    let (produce_tx, produce_rx) = mpsc::sync_channel(128);
-
-    // event channel for consuming events
-    let (consume_tx, consume_rx) = mpsc::sync_channel(128);
-
-    // create client manager
-    let client_manager = match ClientManager::new(&config) {
-        Err(e) => {
-            eprintln!("{e}");
-            process::exit(1);
-        }
-        Ok(m) => m,
-    };
-
-    // start receiving client connection requests
-    let (request_server, request_thread) = match request::Server::listen(port) {
-        Err(e) => {
-            eprintln!("Could not bind to port {port}: {e}");
-            process::exit(1);
-        }
-        Ok(r) => r,
-    };
-
-    println!("Press Ctrl+Alt+Shift+Super to release the mouse");
+    let config = Config::new()?;
 
     // start producing and consuming events
-    let event_producer = match producer::start(produce_tx, client_manager.get_clients(), request_server) {
-        Err(e) => {
-            eprintln!("Could not start event producer: {e}");
-            None
-        },
-        Ok(p) => Some(p),
-    };
-    let event_consumer = match consumer::start(consume_rx, client_manager.get_clients(), config.backend) {
-        Err(e) => {
-            eprintln!("Could not start event consumer: {e}");
-            None
-        },
-        Ok(p) => Some(p),
-    };
+    let producer = producer::create()?;
+    let consumer = consumer::create()?;
 
-    if event_consumer.is_none() && event_producer.is_none() {
-        process::exit(1);
-    }
+    // create frontend communication adapter
+    let frontend_adapter = FrontendAdapter::new()?;
 
     // start sending and receiving events
-    let event_server = match event::server::Server::new(port) {
-        Ok(s) => s,
-        Err(e) => {
-            eprintln!("{e}");
-            process::exit(1);
-        }
-    };
-    let (receiver, sender) = match event_server.run(Arc::new(client_manager), produce_rx, consume_tx) {
-        Ok((r,s)) => (r,s),
-        Err(e) => {
-            eprintln!("{e}");
-            process::exit(1);
+    let mut event_server = Server::new(config.port, producer, consumer, frontend_adapter)?;
+
+    // add clients form config
+    config.get_clients().into_iter().for_each(|(c, h, p)| {
+        let host_name = match h {
+            Some(h) => format!(" '{}'", h),
+            None => "".to_owned(),
+        };
+        if c.len() == 0 {
+            log::warn!("ignoring client{} with 0 assigned ips!", host_name);
         }
+        log::info!("adding client [{}]{} @ {:?}", p, host_name, c);
+        event_server.add_client(c, p);
+    });
+
+    // any threads need to be started after event_server sets up signal handling
+    match config.frontend {
+        Gtk => {
+            #[cfg(all(unix, feature = "gtk"))]
+            frontend::gtk::create();
+            #[cfg(not(feature = "gtk"))]
+            panic!("gtk frontend requested but feature not enabled!");
+        },
+        Cli => Box::new(CliFrontend::new()?),
     };
 
-    request_thread.join().unwrap();
+    log::info!("Press Ctrl+Alt+Shift+Super to release the mouse");
+    // run event loop
+    event_server.run()?;
 
-    // stop receiving events and terminate event-consumer
-    if let Err(e) = receiver.join().unwrap() {
-        eprint!("{e}");
-        process::exit(1);
-    }
-
-    if let Some(thread) = event_consumer {
-        thread.join().unwrap();
-    }
-
-    // stop producing events and terminate event-sender
-    if let Some(thread) = event_producer {
-        thread.join().unwrap();
-    }
-
-    if let Err(e) = sender.join().unwrap() {
-        eprint!("{e}");
-        process::exit(1);
-    }
+    Ok(())
 }

src/producer.rs

@@ -1,10 +1,10 @@
+use mio::event::Source;
+use std::{error::Error, vec::Drain};
+use crate::{client::{ClientHandle, ClientEvent}, event::Event};
+use crate::backend::producer;
+
 #[cfg(unix)]
 use std::env;
-use std::{thread::{JoinHandle, self}, sync::mpsc::SyncSender, error::Error};
-
-use crate::{client::{Client, ClientHandle}, event::Event, request::Server};
-
-use crate::backend::producer;
 
 #[cfg(unix)]
 enum Backend {
@@ -12,41 +12,52 @@ enum Backend {
     X11,
 }
 
-pub fn start(
-    produce_tx: SyncSender<(Event, ClientHandle)>,
-    clients: Vec<Client>,
-    request_server: Server,
-) -> Result<JoinHandle<()>, Box<dyn Error>> {
-    Ok(thread::Builder::new()
-        .name("event producer".into())
-        .spawn(move || {
-            #[cfg(windows)]
-            producer::windows::run(produce_tx, request_server, clients);
+pub fn create() -> Result<Box<dyn EventProducer>, Box<dyn Error>> {
+    #[cfg(windows)]
+    return Ok(Box::new(producer::windows::WindowsProducer::new()));
 
-            #[cfg(unix)]
-            let backend = match env::var("XDG_SESSION_TYPE") {
-                Ok(session_type) => match session_type.as_str() {
-                    "x11" => Backend::X11,
-                    "wayland" => Backend::Wayland,
-                    _ => panic!("unknown XDG_SESSION_TYPE"),
-                },
-                Err(_) => panic!("could not detect session type: XDG_SESSION_TYPE environment variable not set!"),
-            };
-
-            #[cfg(unix)]
-            match backend {
-                Backend::X11 => {
-                    #[cfg(not(feature = "x11"))]
-                    panic!("feature x11 not enabled");
-                    #[cfg(feature = "x11")]
-                    producer::x11::run(produce_tx, request_server, clients);
-                }
-                Backend::Wayland => {
-                    #[cfg(not(feature = "wayland"))]
-                    panic!("feature wayland not enabled");
-                    #[cfg(feature = "wayland")]
-                    producer::wayland::run(produce_tx, request_server, clients);
-                }
+    #[cfg(unix)]
+    let backend = match env::var("XDG_SESSION_TYPE") {
+        Ok(session_type) => match session_type.as_str() {
+            "x11" => {
+                log::info!("XDG_SESSION_TYPE = x11 -> using X11 event producer");
+                Backend::X11
+            },
+            "wayland" => {
+                log::info!("XDG_SESSION_TYPE = wayland -> using wayland event producer");
+                Backend::Wayland
             }
-        })?)
+            _ => panic!("unknown XDG_SESSION_TYPE"),
+        },
+        Err(_) => panic!("could not detect session type: XDG_SESSION_TYPE environment variable not set!"),
+    };
+
+    #[cfg(unix)]
+    match backend {
+        Backend::X11 => {
+            #[cfg(not(feature = "x11"))]
+            panic!("feature x11 not enabled");
+            #[cfg(feature = "x11")]
+            Ok(Box::new(producer::x11::X11Producer::new()))
+        }
+        Backend::Wayland => {
+            #[cfg(not(feature = "wayland"))]
+            panic!("feature wayland not enabled");
+            #[cfg(feature = "wayland")]
+            Ok(Box::new(producer::wayland::WaylandEventProducer::new()?))
+        }
+    }
+}
+
+pub trait EventProducer: Source {
+    /// notify event producer of configuration changes
+    fn notify(&mut self, event: ClientEvent);
+
+    /// read an event
+    /// this function must be invoked to retrieve an Event after
+    /// the eventfd indicates a pending Event
+    fn read_events(&mut self) -> Drain<(ClientHandle, Event)>;
+
+    /// release mouse
+    fn release(&mut self);
 }

src/request.rs

@@ -1,139 +0,0 @@
-use std::{
-    collections::HashMap,
-    error::Error,
-    fmt::Display,
-    io::prelude::*,
-    net::{SocketAddr, TcpListener, TcpStream},
-    sync::{Arc, RwLock},
-    thread::{self, JoinHandle},
-};
-
-use memmap::MmapMut;
-
-#[derive(Copy, Clone, PartialEq, Eq, Hash)]
-pub enum Request {
-    KeyMap,
-    Connect,
-}
-
-impl TryFrom<[u8; 4]> for Request {
-    fn try_from(buf: [u8; 4]) -> Result<Self, Self::Error> {
-        let val = u32::from_ne_bytes(buf);
-        match val {
-            x if x == Request::KeyMap as u32 => Ok(Self::KeyMap),
-            x if x == Request::Connect as u32 => Ok(Self::Connect),
-            _ => Err("Bad Request"),
-        }
-    }
-
-    type Error = &'static str;
-}
-
-#[derive(Clone)]
-pub struct Server {
-    data: Arc<RwLock<HashMap<Request, MmapMut>>>,
-}
-
-impl Server {
-    fn handle_request(&self, mut stream: TcpStream) -> Result<(), Box<dyn Error>> {
-        let mut buf = [0u8; 4];
-        stream.read_exact(&mut buf)?;
-        match Request::try_from(buf) {
-            Ok(Request::KeyMap) => {
-                let data = self.data.read().unwrap();
-                let buf = data.get(&Request::KeyMap);
-                match buf {
-                    None => {
-                        stream.write(&0u32.to_ne_bytes())?;
-                    }
-                    Some(buf) => {
-                        stream.write(&buf[..].len().to_ne_bytes())?;
-                        stream.write(&buf[..])?;
-                    }
-                }
-                stream.flush()?;
-            }
-            Ok(Request::Connect) => todo!(),
-            Err(msg) => eprintln!("{}", msg),
-        }
-        Ok(())
-    }
-
-    pub fn listen(port: u16) -> Result<(Server, JoinHandle<()>), Box<dyn Error>> {
-        let data: Arc<RwLock<HashMap<Request, MmapMut>>> = Arc::new(RwLock::new(HashMap::new()));
-        let listen_addr = SocketAddr::new("0.0.0.0".parse()?, port);
-        let server = Server { data };
-        let server_copy = server.clone();
-        let listen_socket = TcpListener::bind(listen_addr)?;
-        let thread = thread::Builder::new()
-            .name("tcp server".into())
-            .spawn(move || {
-            for stream in listen_socket.incoming() {
-                match stream {
-                    Ok(stream) => {
-                        if let Err(e) = server.handle_request(stream) {
-                            eprintln!("{}", e);
-                        }
-                    }
-                    Err(e) => {
-                        eprintln!("{}", e);
-                    }
-                }
-            }
-        })?;
-        Ok((server_copy, thread))
-    }
-
-    pub fn offer_data(&self, req: Request, d: MmapMut) {
-        self.data.write().unwrap().insert(req, d);
-    }
-}
-
-#[derive(Debug)]
-pub struct BadRequest;
-
-impl Display for BadRequest {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "BadRequest")
-    }
-}
-
-impl Error for BadRequest {}
-
-pub fn request_data(addr: SocketAddr, req: Request) -> Result<Vec<u8>, Box<dyn Error>> {
-    // connect to server
-    let mut sock = match TcpStream::connect(addr) {
-        Ok(sock) => sock,
-        Err(e) => return Err(Box::new(e)),
-    };
-
-    // write the request to the socket
-    // convert to u32
-    let req: u32 = req as u32;
-    if let Err(e) = sock.write(&req.to_ne_bytes()) {
-        return Err(Box::new(e));
-    }
-    if let Err(e) = sock.flush() {
-        return Err(Box::new(e));
-    }
-
-    // read the response = (len, data) - len 0 means no data / bad request
-    // read len
-    let mut buf = [0u8; 8];
-    if let Err(e) = sock.read_exact(&mut buf[..]) {
-        return Err(Box::new(e));
-    }
-    let len = usize::from_ne_bytes(buf);
-
-    // check for bad request
-    if len == 0 {
-        return Err(Box::new(BadRequest {}));
-    }
-
-    // read the data
-    let mut data: Vec<u8> = vec![0u8; len];
-    if let Err(e) = sock.read_exact(&mut data[..]) {
-        return Err(Box::new(e));
-    }
-    Ok(data)
-}