touchpad-toggle-observer

Cargo.toml

[package]
name = "touchpad-toggle-observer"
version = "0.1.0"
edition = "2021"

[dependencies.xcb]
version = "1"
features = [
    "xlib_xcb",
    "xinput",
    "x11",
    "debug_atom_names",
]

main.rs

mod atom_name {
    // #include <X11/extensions/XI.h>
    pub const XI_TOUCHPAD: &'static [u8] = b"TOUCHPAD";

    // #include <xorg/xserver-properties.h>
    pub const XI_PROP_ENABLED: &'static [u8] = b"Device Enabled";

    // #include <xorg/libinput-properties.h>
    pub const LIBINPUT_PROP_SENDEVENTS_ENABLED: &'static [u8] =
        b"libinput Send Events Mode Enabled";
}

#[derive(Debug)]
#[allow(non_snake_case)]
struct Atoms {
    pub XI_TOUCHPAD: xcb::x::Atom,
    pub XI_PROP_ENABLED: xcb::x::Atom,
    pub LIBINPUT_PROP_SENDEVENTS_ENABLED: xcb::x::Atom,
}

impl Atoms {
    pub fn new(conn: &xcb::Connection) -> xcb::Result<Self> {
        Ok(Self {
            XI_TOUCHPAD: intern_atom(conn, atom_name::XI_TOUCHPAD)?,
            XI_PROP_ENABLED: intern_atom(conn, atom_name::XI_PROP_ENABLED)?,
            LIBINPUT_PROP_SENDEVENTS_ENABLED: intern_atom(
                conn,
                atom_name::LIBINPUT_PROP_SENDEVENTS_ENABLED,
            )?,
        })
    }
}

fn intern_atom(conn: &xcb::Connection, name: &[u8]) -> xcb::Result<xcb::x::Atom> {
    let request = xcb::x::InternAtom {
        only_if_exists: false,
        name,
    };
    let reply = conn.wait_for_reply(conn.send_request(&request))?;
    Ok(reply.atom())
}

pub fn harness() -> xcb::Result<(xcb::Connection, xcb::x::Window)> {
    let (conn, screen_num) =
        xcb::Connection::connect_with_extensions(None, &[xcb::Extension::Input], &[]).unwrap();
    let setup = conn.get_setup();
    let screen = setup.roots().nth(screen_num as usize).unwrap();
    let window: xcb::x::Window = conn.generate_id();

    conn.send_request(&xcb::x::CreateWindow {
        depth: 0,
        wid: window,
        parent: screen.root(),
        x: 0,
        y: 0,
        width: 1,
        height: 1,
        border_width: 0,
        class: xcb::x::WindowClass::InputOutput,
        visual: screen.root_visual(),
        value_list: &[],
    });

    conn.flush()?;

    Ok((conn, window))
}

// Check whether a touchpad is enabled.
// If no touchpad is detected, defaults to true.
// If multiple touchpads are detected, check if all of them are enabled.
// In case of errors, defaults to true.
fn is_touchpad_enabled(conn: &xcb::Connection, atoms: &Atoms) -> bool {
    is_touchpad_enabled_result(conn, atoms).unwrap_or(true)
}

fn is_touchpad_enabled_result(conn: &xcb::Connection, atoms: &Atoms) -> xcb::Result<bool> {
    for device_id in list_touchpads(conn, atoms)? {
        let is_enabled = is_touchpad_enabled_single_result(conn, atoms, device_id).unwrap_or(true);
        if !is_enabled {
            return Ok(false);
        }
    }
    // default when no touchpads, or when all of them are enabled
    Ok(true)
}

// Check if either "Device Enabled" or special libinput atoms indicate that the touchpad is disabled. Defaults to enabled otherwise.
fn is_touchpad_enabled_single_result(
    conn: &xcb::Connection,
    atoms: &Atoms,
    device_id: u8,
) -> xcb::Result<bool> {
    let property = conn.wait_for_reply(conn.send_request(&xcb::xinput::GetDeviceProperty {
        property: atoms.XI_PROP_ENABLED,
        r#type: xcb::x::ATOM_INTEGER,
        offset: 0,
        len: 1,
        device_id,
        delete: false,
    }))?;
    if property.length() == 0 {
        // rust-xcb may panic if format is 0 due to missing property on the device
        return Ok(true);
    }
    match property.items() {
        xcb::xinput::GetDevicePropertyReplyItems::N8Bits(items) => {
            if items.len() != 1 {
                // Expected exactly one item
                return Ok(true);
            }
            if items[0] == 0 {
                // Zero means disabled
                return Ok(false);
            }
        }
        // Format error
        _ => return Ok(true),
    }

    let property = conn.wait_for_reply(conn.send_request(&xcb::xinput::GetDeviceProperty {
        property: atoms.LIBINPUT_PROP_SENDEVENTS_ENABLED,
        r#type: xcb::x::ATOM_INTEGER,
        offset: 0,
        len: 1,
        device_id,
        delete: false,
    }))?;
    if property.length() == 0 {
        // rust-xcb may panic if format is 0 due to missing property on the device
        return Ok(true);
    }
    match property.items() {
        xcb::xinput::GetDevicePropertyReplyItems::N8Bits(items) => {
            // For the libinput's special atom, items[0] == 1 means disabled. No idea what the second item is for.
            if items[0] == 1u8 {
                return Ok(false);
            }
        }
        // Format error
        _ => return Ok(true),
    }

    Ok(true)
}

// Returns list of Device IDs for connected touchpads.
fn list_touchpads(conn: &xcb::Connection, atoms: &Atoms) -> xcb::Result<Vec<u8>> {
    let devices = conn.wait_for_reply(conn.send_request(&xcb::xinput::ListInputDevices {}))?;

    let mut touchpads = Vec::<u8>::new();

    for device_info in devices.devices() {
        if device_info.device_type() == atoms.XI_TOUCHPAD {
            touchpads.push(device_info.device_id());
        }
    }

    Ok(touchpads)
}

fn is_touchpad_enabled_sender(sender: std::sync::mpsc::Sender<bool>) -> xcb::Result<()> {
    let (conn, window) = harness()?;

    let atoms = Atoms::new(&conn)?;

    let is_enabled = is_touchpad_enabled(&conn, &atoms);
    if let Err(_) = sender.send(is_enabled) {
        return Ok(());
    }
    let mut last_enabled = is_enabled;

    // Subscribe to relevant events
    let mask = xcb::xinput::EventMaskBuf::new(
        xcb::xinput::Device::All,
        &[
            // Toggling XI_PROP_ENABLED atom causes hierarchy change event
            xcb::xinput::XiEventMask::HIERARCHY |
            // Toggling LIBINPUT_PROP_SENDEVENTS_ENABLED atom only causes a property change event
            xcb::xinput::XiEventMask::PROPERTY,
        ],
    );
    conn.check_request(conn.send_request_checked(&xcb::xinput::XiSelectEvents {
        window,
        masks: &[mask],
    }))?;

    loop {
        conn.flush()?;

        let event = conn.wait_for_event()?;

        match event {
            xcb::Event::Input(
                xcb::xinput::Event::Hierarchy(_) | xcb::xinput::Event::Property(_),
            ) => {
                // Do not read any properties of the Hierarchy event, it may crash rust-xcb
                let is_enabled = is_touchpad_enabled(&conn, &atoms);
                // Deduplicate the events, because both events may be
                // triggered dozens of times in a row, with the same end
                // result for us
                if last_enabled != is_enabled {
                    last_enabled = is_enabled;
                    if let Err(_) = sender.send(is_enabled) {
                        return Ok(());
                    }
                }
            }
            _ => {}
        }
    }
}

fn is_touchpad_enabled_channel() -> std::sync::mpsc::Receiver<bool> {
    let (sender, receiver) = std::sync::mpsc::channel::<bool>();

    std::thread::spawn(|| {
        is_touchpad_enabled_sender(sender)
    });

    receiver
}

fn main() {
    let stream = is_touchpad_enabled_channel();
    loop {
        match stream.recv() {
            Ok(is_enabled) => println!("is_enabled: {:?}", is_enabled),
            Err(e) => eprint!("Error: {:?}", e),
        }
    }
}