openlogi_device/channel/
registry.rs

1//! Registry of HID++ channels owned by the persistent inventory enumerator.
2
3use std::collections::HashSet;
4use std::hash::Hash;
5use std::sync::{Arc, PoisonError, RwLock};
6
7use crate::backend::NodeId;
8use hidpp::channel::HidppChannel;
9
10use crate::{DeviceRoute, SharedChannel};
11
12struct Publication<Node, Channel> {
13    node: Node,
14    sequence: u64,
15    routes: Vec<DeviceRoute>,
16    channel: Channel,
17}
18
19struct NodeRegistry<Node, Channel> {
20    publications: Vec<Publication<Node, Channel>>,
21    next_sequence: u64,
22}
23
24impl<Node, Channel> Default for NodeRegistry<Node, Channel> {
25    fn default() -> Self {
26        Self {
27            publications: Vec::new(),
28            next_sequence: 0,
29        }
30    }
31}
32
33impl<Node: Eq, Channel> NodeRegistry<Node, Channel> {
34    fn replace_node(
35        &mut self,
36        node: Node,
37        routes: impl IntoIterator<Item = DeviceRoute>,
38        channel: Channel,
39    ) {
40        let routes = routes.into_iter().collect();
41        if let Some(publication) = self
42            .publications
43            .iter_mut()
44            .find(|publication| publication.node == node)
45        {
46            publication.routes = routes;
47            publication.channel = channel;
48            return;
49        }
50
51        let sequence = self.next_sequence;
52        self.next_sequence = self.next_sequence.wrapping_add(1);
53        self.publications.push(Publication {
54            node,
55            sequence,
56            routes,
57            channel,
58        });
59    }
60
61    fn remove_node(&mut self, node: &Node) {
62        self.publications
63            .retain(|publication| publication.node != *node);
64    }
65
66    fn lookup(&self, route: &DeviceRoute) -> Option<&Channel> {
67        self.publications
68            .iter()
69            .filter(|publication| publication.routes.contains(route))
70            .min_by_key(|publication| publication.sequence)
71            .map(|publication| &publication.channel)
72    }
73
74    fn any_current(&self, mut predicate: impl FnMut(&DeviceRoute, &Channel) -> bool) -> bool {
75        self.publications.iter().any(|publication| {
76            publication.routes.iter().any(|route| {
77                predicate(route, &publication.channel)
78                    && self
79                        .lookup(route)
80                        .is_some_and(|winner| std::ptr::eq(winner, &raw const publication.channel))
81            })
82        })
83    }
84}
85
86impl<Node: Eq + Hash, Channel> NodeRegistry<Node, Channel> {
87    fn retain_nodes(&mut self, nodes: &HashSet<Node>) {
88        self.publications
89            .retain(|publication| nodes.contains(&publication.node));
90    }
91}
92
93struct Registry<Node, Channel> {
94    state: Arc<RwLock<NodeRegistry<Node, Channel>>>,
95}
96
97impl<Node, Channel> Clone for Registry<Node, Channel> {
98    fn clone(&self) -> Self {
99        Self {
100            state: Arc::clone(&self.state),
101        }
102    }
103}
104
105impl<Node, Channel> Default for Registry<Node, Channel> {
106    fn default() -> Self {
107        Self {
108            state: Arc::new(RwLock::new(NodeRegistry::default())),
109        }
110    }
111}
112
113impl<Node: Eq, Channel> Registry<Node, Channel> {
114    fn replace_node(
115        &self,
116        node: Node,
117        routes: impl IntoIterator<Item = DeviceRoute>,
118        channel: Channel,
119    ) {
120        self.state
121            .write()
122            .unwrap_or_else(PoisonError::into_inner)
123            .replace_node(node, routes, channel);
124    }
125
126    fn remove_node(&self, node: &Node) {
127        self.state
128            .write()
129            .unwrap_or_else(PoisonError::into_inner)
130            .remove_node(node);
131    }
132}
133
134impl<Node: Eq + Hash, Channel> Registry<Node, Channel> {
135    fn retain_nodes(&self, nodes: &HashSet<Node>) {
136        self.state
137            .write()
138            .unwrap_or_else(PoisonError::into_inner)
139            .retain_nodes(nodes);
140    }
141}
142
143impl<Node: Eq, Channel: Clone> Registry<Node, Channel> {
144    fn lookup(&self, route: &DeviceRoute) -> Option<Channel> {
145        self.state.read().ok()?.lookup(route).cloned()
146    }
147}
148
149impl<Node: Eq, Channel> Registry<Node, Channel> {
150    fn any_current(&self, predicate: impl FnMut(&DeviceRoute, &Channel) -> bool) -> bool {
151        self.state
152            .read()
153            .is_ok_and(|state| state.any_current(predicate))
154    }
155}
156
157/// Channels already opened and owned by the persistent inventory enumerator.
158///
159/// Publications are keyed by OS HID node internally and selected by exact
160/// [`DeviceRoute`]. When identical direct devices publish the same route, the
161/// oldest live node wins until it is removed.
162#[derive(Clone, Default)]
163pub struct ChannelRegistry {
164    inner: Registry<NodeId, Arc<HidppChannel>>,
165}
166
167impl ChannelRegistry {
168    /// Replace every route published by `node`, preserving that node's original
169    /// collision priority when it was already present.
170    pub(crate) fn replace_node(
171        &self,
172        node: NodeId,
173        routes: impl IntoIterator<Item = DeviceRoute>,
174        channel: Arc<HidppChannel>,
175    ) {
176        self.inner.replace_node(node, routes, channel);
177    }
178
179    /// Remove every route and channel reference owned by `node`.
180    pub(crate) fn remove_node(&self, node: &NodeId) {
181        self.inner.remove_node(node);
182    }
183
184    /// Remove publications for nodes absent from the current OS enumeration.
185    pub(crate) fn retain_nodes(&self, nodes: &HashSet<NodeId>) {
186        self.inner.retain_nodes(nodes);
187    }
188
189    /// Clone the current exact-route winner.
190    #[must_use]
191    pub fn lookup(&self, route: &DeviceRoute) -> Option<SharedChannel> {
192        self.inner
193            .lookup(route)
194            .map(|channel| SharedChannel::new(channel, route.clone()))
195    }
196
197    /// Whether `shared` is still the winning publication for its exact route
198    /// and points to the same underlying connection.
199    #[must_use]
200    pub fn is_current(&self, shared: &SharedChannel) -> bool {
201        self.inner.any_current(|route, channel| {
202            shared.matches(route) && Arc::ptr_eq(channel, shared.channel())
203        })
204    }
205}
206
207#[cfg(test)]
208mod tests {
209    use std::collections::HashSet;
210    use std::panic::{AssertUnwindSafe, catch_unwind};
211    use std::sync::Arc;
212
213    use crate::DeviceRoute;
214
215    use super::{PoisonError, Registry};
216
217    impl<Node, Channel> Registry<Node, Channel> {
218        fn poison_for_test(&self) {
219            let _ = catch_unwind(AssertUnwindSafe(|| {
220                let _guard = self.state.write().unwrap_or_else(PoisonError::into_inner);
221                panic!("poison registry for test");
222            }));
223        }
224    }
225
226    impl<Node: Eq, Channel: Clone> Registry<Node, Channel> {
227        fn publisher_lookup_for_test(&self, route: &DeviceRoute) -> Option<Channel> {
228            self.state
229                .read()
230                .unwrap_or_else(PoisonError::into_inner)
231                .lookup(route)
232                .cloned()
233        }
234    }
235
236    fn direct(product_id: u16) -> DeviceRoute {
237        DeviceRoute::Direct {
238            vendor_id: 0x046d,
239            product_id,
240        }
241    }
242
243    fn bolt(uid: &str, slot: u8) -> DeviceRoute {
244        DeviceRoute::Bolt {
245            receiver_uid: uid.into(),
246            slot,
247        }
248    }
249
250    #[test]
251    fn lookup_rejects_every_non_exact_route_field() {
252        let registry = Registry::<u8, &'static str>::default();
253        registry.replace_node(1, [bolt("AABB", 2)], "channel-a");
254
255        assert_eq!(registry.lookup(&bolt("AABB", 2)), Some("channel-a"));
256        assert_eq!(registry.lookup(&bolt("AABB", 3)), None);
257        assert_eq!(registry.lookup(&bolt("CCDD", 2)), None);
258        assert_eq!(registry.lookup(&direct(0xb35b)), None);
259    }
260
261    #[test]
262    fn one_node_can_publish_multiple_receiver_slots() {
263        let registry = Registry::<u8, &'static str>::default();
264        registry.replace_node(1, [bolt("AABB", 1), bolt("AABB", 4)], "receiver-channel");
265
266        assert_eq!(registry.lookup(&bolt("AABB", 1)), Some("receiver-channel"));
267        assert_eq!(registry.lookup(&bolt("AABB", 4)), Some("receiver-channel"));
268    }
269
270    #[test]
271    fn current_check_uses_only_the_exact_winning_publication() {
272        let route = direct(0xb35b);
273        let registry = Registry::<u8, &'static str>::default();
274        registry.replace_node(1, [route.clone()], "a");
275        registry.replace_node(2, [route.clone()], "b");
276
277        assert!(
278            registry.any_current(|candidate, channel| { candidate == &route && *channel == "a" })
279        );
280        assert!(
281            !registry.any_current(|candidate, channel| { candidate == &route && *channel == "b" })
282        );
283    }
284
285    #[test]
286    fn same_route_with_a_different_arc_is_not_current() {
287        let route = direct(0xb35b);
288        let published = Arc::new(());
289        let stale = Arc::new(());
290        let registry = Registry::<u8, Arc<()>>::default();
291        registry.replace_node(1, [route.clone()], Arc::clone(&published));
292
293        assert!(registry.any_current(|candidate, channel| {
294            candidate == &route && Arc::ptr_eq(channel, &published)
295        }));
296        assert!(!registry.any_current(|candidate, channel| {
297            candidate == &route && Arc::ptr_eq(channel, &stale)
298        }));
299    }
300
301    #[test]
302    fn replacing_winner_preserves_priority_then_removal_promotes_next_owner() {
303        let route = direct(0xb35b);
304        let registry = Registry::<u8, &'static str>::default();
305        registry.replace_node(1, [route.clone()], "a-v1");
306        registry.replace_node(2, [route.clone()], "b");
307
308        assert_eq!(registry.lookup(&route), Some("a-v1"));
309
310        registry.replace_node(1, [route.clone()], "a-v2");
311        assert_eq!(registry.lookup(&route), Some("a-v2"));
312
313        registry.remove_node(&1);
314        assert_eq!(registry.lookup(&route), Some("b"));
315    }
316
317    #[test]
318    fn replacing_one_node_is_atomic_and_does_not_touch_another() {
319        let registry = Registry::<u8, &'static str>::default();
320        registry.replace_node(1, [bolt("A", 1), bolt("A", 2)], "a");
321        registry.replace_node(2, [bolt("B", 1)], "b");
322
323        registry.replace_node(1, [bolt("A", 3)], "a-new");
324
325        assert_eq!(registry.lookup(&bolt("A", 1)), None);
326        assert_eq!(registry.lookup(&bolt("A", 2)), None);
327        assert_eq!(registry.lookup(&bolt("A", 3)), Some("a-new"));
328        assert_eq!(registry.lookup(&bolt("B", 1)), Some("b"));
329    }
330
331    #[test]
332    fn retaining_nodes_removes_only_absent_owners() {
333        let registry = Registry::<u8, &'static str>::default();
334        registry.replace_node(1, [direct(0xb35b)], "a");
335        registry.replace_node(2, [direct(0xb36b)], "b");
336
337        registry.retain_nodes(&HashSet::from([2]));
338
339        assert_eq!(registry.lookup(&direct(0xb35b)), None);
340        assert_eq!(registry.lookup(&direct(0xb36b)), Some("b"));
341    }
342
343    #[test]
344    fn poisoned_read_fails_closed_but_publishers_can_clean_up() {
345        let registry = Registry::<u8, &'static str>::default();
346        registry.replace_node(1, [direct(0xb35b)], "a");
347        registry.poison_for_test();
348
349        assert_eq!(registry.lookup(&direct(0xb35b)), None);
350        assert!(!registry.any_current(|_, _| true));
351
352        registry.remove_node(&1);
353        registry.replace_node(2, [direct(0xb36b)], "b");
354        registry.retain_nodes(&HashSet::from([2]));
355
356        assert_eq!(registry.publisher_lookup_for_test(&direct(0xb35b)), None);
357        assert_eq!(
358            registry.publisher_lookup_for_test(&direct(0xb36b)),
359            Some("b")
360        );
361    }
362}
openlogi_device/channel/registry.rs

openlogi_device/channel/
registry.rs
