openlogi_device/
device_io.rs

1//! Process activity gate for host HID access.
2//!
3//! The gate carries policy from a host lifecycle observer down to code that
4//! may open a HID node or send a request. It contains no host integration of
5//! its own, so the device layer remains portable and testable.
6
7use thiserror::Error;
8use tokio::sync::watch;
9
10use crate::backend::BackendError;
11
12/// The gate's refusal while host device I/O is suspended.
13///
14/// Only [`DeviceIoGate::ensure_allowed`] constructs one, so the check and its
15/// wording live there; callers convert it into their own error with `?`.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
17#[error("host device I/O is suspended")]
18pub struct IoSuspended(());
19
20impl From<IoSuspended> for BackendError {
21    fn from(error: IoSuspended) -> Self {
22        Self::Backend(error.to_string())
23    }
24}
25
26/// Non-blocking producer owned by the host lifecycle observer.
27#[derive(Clone)]
28pub struct DeviceIoSignal {
29    sender: watch::Sender<DeviceIoState>,
30}
31
32/// Cheaply cloneable read capability for device-I/O producers.
33#[derive(Clone)]
34pub struct DeviceIoGate {
35    receiver: watch::Receiver<DeviceIoState>,
36}
37
38#[derive(Clone, Copy, Debug, PartialEq, Eq)]
39enum DeviceIoState {
40    Allowed,
41    Suspended,
42}
43
44/// Create an initially-open device-I/O gate.
45#[must_use]
46pub fn device_io_channel() -> (DeviceIoSignal, DeviceIoGate) {
47    let (sender, receiver) = watch::channel(DeviceIoState::Allowed);
48    (DeviceIoSignal { sender }, DeviceIoGate { receiver })
49}
50
51impl DeviceIoSignal {
52    /// Close the gate without blocking the native lifecycle callback.
53    ///
54    /// Returns whether this call changed the published state.
55    #[must_use]
56    pub fn suspend(&self) -> bool {
57        self.sender.send_if_modified(|state| {
58            if *state == DeviceIoState::Suspended {
59                return false;
60            }
61            *state = DeviceIoState::Suspended;
62            true
63        })
64    }
65
66    /// Reopen the gate after a user-visible resume.
67    ///
68    /// Returns whether this call changed the published state.
69    #[must_use]
70    pub fn resume(&self) -> bool {
71        self.sender.send_if_modified(|state| {
72            if *state == DeviceIoState::Allowed {
73                return false;
74            }
75            *state = DeviceIoState::Allowed;
76            true
77        })
78    }
79}
80
81impl DeviceIoGate {
82    /// Whether opening a node or sending a request is currently allowed.
83    #[must_use]
84    pub fn allows_io(&self) -> bool {
85        *self.receiver.borrow() == DeviceIoState::Allowed
86    }
87
88    /// Refuse an open or a request while host device I/O is suspended.
89    ///
90    /// # Errors
91    ///
92    /// Returns [`IoSuspended`] while the gate is closed.
93    pub fn ensure_allowed(&self) -> Result<(), IoSuspended> {
94        if self.allows_io() {
95            Ok(())
96        } else {
97            Err(IoSuspended(()))
98        }
99    }
100
101    /// Wait for the next distinct gate transition.
102    ///
103    /// Returns the new allow-state, or `None` if the lifecycle producer was
104    /// dropped.
105    pub async fn changed(&mut self) -> Option<bool> {
106        self.receiver.changed().await.ok()?;
107        Some(self.allows_io())
108    }
109
110    /// Wait until the gate is open. Returns `false` if its producer disappears
111    /// while it is closed.
112    pub async fn wait_until_allowed(&mut self) -> bool {
113        while !self.allows_io() {
114            if self.changed().await.is_none() {
115                return false;
116            }
117        }
118        true
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    use std::time::Duration;
125
126    use super::{BackendError, device_io_channel};
127
128    #[tokio::test]
129    async fn latest_transition_is_retained_and_duplicates_coalesce() {
130        let (signal, mut gate) = device_io_channel();
131        assert!(gate.allows_io());
132
133        assert!(signal.suspend());
134        assert!(!signal.suspend());
135        gate.changed()
136            .await
137            .expect("the suspend transition should remain available");
138        assert!(!gate.allows_io());
139        tokio::time::timeout(Duration::from_millis(10), gate.changed())
140            .await
141            .expect_err("duplicate suspend publications must coalesce");
142
143        assert!(signal.resume());
144        assert!(gate.wait_until_allowed().await);
145        assert!(gate.allows_io());
146    }
147
148    #[test]
149    fn a_closed_gate_refuses_until_it_reopens() {
150        let (signal, gate) = device_io_channel();
151        gate.ensure_allowed()
152            .expect("an open gate allows device I/O");
153
154        assert!(signal.suspend());
155        let refusal = gate
156            .ensure_allowed()
157            .expect_err("a suspended gate must refuse device I/O");
158        assert!(matches!(
159            BackendError::from(refusal),
160            BackendError::Backend(message) if message == refusal.to_string()
161        ));
162
163        assert!(signal.resume());
164        gate.ensure_allowed()
165            .expect("a resumed gate allows device I/O again");
166    }
167}
openlogi_device/device_io.rs

openlogi_device/
device_io.rs
