anybox/
event.rs

1use std::collections::VecDeque;
2use std::fmt;
3use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError, Weak};
4use std::time::{Duration, Instant};
5
6use crate::capability::{Action, Button};
7use crate::key::{Key, MediaKey};
8
9#[derive(Clone, Debug, PartialEq, Eq)]
10#[non_exhaustive]
11pub enum Event {
12    /// a physical button
13    #[non_exhaustive]
14    Button {
15        button: Button,
16        pressed: bool,
17        at: Stamp,
18    },
19    /// a physical key on the keyboard behind the box. just know that you see every key typed there, passwords
20    /// included
21    #[non_exhaustive]
22    Key {
23        key: Key,
24        pressed: bool,
25        at: Stamp,
26    },
27    /// a physical media key, like volume or play
28    #[non_exhaustive]
29    Media {
30        key: MediaKey,
31        pressed: bool,
32        at: Stamp,
33    },
34    /// physical motion. the wheel is positive going up and pan (horizontal scroll) is positive going right
35    #[non_exhaustive]
36    Motion {
37        dx: i32,
38        dy: i32,
39        wheel: i32,
40        pan: i32,
41        at: Stamp,
42    },
43    /// how many events never reached you since the last `Dropped` from the same source. one from the box can mean
44    /// it restarted a stream, so what you knew about which physical buttons and keys were held may be out of date.
45    /// the one for your own queue (`DropSource::Host`) sits at the front of the queue, not where the loss happened,
46    /// and on Medius the one from the box comes from a box wide counter read about once a second, so it can also
47    /// count what your own subscriptions through `medius()` lost
48    #[non_exhaustive]
49    Dropped {
50        count: u64,
51        source: DropSource,
52    },
53    /// the box lost its state, so the buttons and keys you were holding got released (on a box with
54    /// `Quirk::KeepsHeld` they stay held). what you knew about which physical buttons and keys were held is out of
55    /// date too, because one held through the reset can show a second press or never show its release
56    SessionReset,
57    /// the box refused something after the call already returned ok, MAKCU does that when its command queue is
58    /// full. slow down and send it again if it mattered. `count` is how many of that action got refused since you
59    /// last read one, so a flood of refusals never pushes your input out of the queue
60    #[non_exhaustive]
61    Rejected {
62        action: Action,
63        count: u64,
64    },
65    Disconnected,
66    /// the box is back, the buttons and keys you were holding got released (on a box with `Quirk::KeepsHeld`
67    /// they stay held), and the snapshot may have changed. you also get this when a mouse or keyboard gets plugged
68    /// into the box or pulled out. what you knew about which physical buttons and keys were held is out of date,
69    /// same as after `SessionReset`
70    Reconnected,
71}
72
73/// for tests and for code that builds events itself. the variants stay `#[non_exhaustive]` so a field added
74/// later doesn't break anyone, and these are how you make them from outside
75impl Event {
76    pub fn button(button: Button, pressed: bool, at: Stamp) -> Self {
77        Self::Button {
78            button,
79            pressed,
80            at,
81        }
82    }
83
84    pub fn key(key: Key, pressed: bool, at: Stamp) -> Self {
85        Self::Key { key, pressed, at }
86    }
87
88    pub fn media(key: MediaKey, pressed: bool, at: Stamp) -> Self {
89        Self::Media { key, pressed, at }
90    }
91
92    pub fn motion(dx: i32, dy: i32, wheel: i32, pan: i32, at: Stamp) -> Self {
93        Self::Motion {
94            dx,
95            dy,
96            wheel,
97            pan,
98            at,
99        }
100    }
101
102    pub fn dropped(count: u64, source: DropSource) -> Self {
103        Self::Dropped { count, source }
104    }
105
106    pub fn rejected(action: Action, count: u64) -> Self {
107        Self::Rejected { action, count }
108    }
109}
110
111/// which input a receiver wants, combined with `|`. a receiver only gets what it asked for, the box streams what
112/// any receiver wants, and connection events and drops reach every receiver
113#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
114pub struct EventKinds {
115    buttons: bool,
116    motion: bool,
117    keys: bool,
118    media: bool,
119}
120
121impl EventKinds {
122    pub const NONE: Self = Self {
123        buttons: false,
124        motion: false,
125        keys: false,
126        media: false,
127    };
128    pub const BUTTONS: Self = Self {
129        buttons: true,
130        ..Self::NONE
131    };
132    pub const MOTION: Self = Self {
133        motion: true,
134        ..Self::NONE
135    };
136    pub const KEYS: Self = Self {
137        keys: true,
138        ..Self::NONE
139    };
140    pub const MEDIA: Self = Self {
141        media: true,
142        ..Self::NONE
143    };
144    /// everything the box can stream, what `events()` gives you
145    pub const ALL: Self = Self {
146        buttons: true,
147        motion: true,
148        keys: true,
149        media: true,
150    };
151
152    pub fn contains(self, other: Self) -> bool {
153        self & other == other
154    }
155
156    pub fn is_empty(self) -> bool {
157        self == Self::NONE
158    }
159
160    /// the kind an event belongs to, none for the ones every receiver gets
161    pub(crate) fn of(event: &Event) -> Option<Self> {
162        match event {
163            Event::Button { .. } => Some(Self::BUTTONS),
164            Event::Motion { .. } => Some(Self::MOTION),
165            Event::Key { .. } => Some(Self::KEYS),
166            Event::Media { .. } => Some(Self::MEDIA),
167            Event::Rejected { action, .. } => Self::of_action(*action),
168            _ => None,
169        }
170    }
171
172    /// the kind a stream's action is about, so its refusal only goes to the receivers that wanted it
173    pub(crate) fn of_action(action: Action) -> Option<Self> {
174        match action {
175            Action::ButtonEvents => Some(Self::BUTTONS),
176            Action::MotionEvents => Some(Self::MOTION),
177            Action::KeyEvents => Some(Self::KEYS),
178            Action::MediaEvents => Some(Self::MEDIA),
179            _ => None,
180        }
181    }
182
183    pub(crate) fn to_input(self) -> crate::backend::InputKinds {
184        crate::backend::InputKinds {
185            buttons: self.buttons,
186            motion: self.motion,
187            keys: self.keys,
188            media: self.media,
189        }
190    }
191
192    pub(crate) fn from_input(kinds: crate::backend::InputKinds) -> Self {
193        Self {
194            buttons: kinds.buttons,
195            motion: kinds.motion,
196            keys: kinds.keys,
197            media: kinds.media,
198        }
199    }
200
201    /// each single kind in this set with the action that streams it
202    pub(crate) fn each(self) -> impl Iterator<Item = (Self, Action)> {
203        [
204            (Self::BUTTONS, Action::ButtonEvents),
205            (Self::MOTION, Action::MotionEvents),
206            (Self::KEYS, Action::KeyEvents),
207            (Self::MEDIA, Action::MediaEvents),
208        ]
209        .into_iter()
210        .filter(move |(kind, _)| self.contains(*kind))
211    }
212}
213
214impl std::ops::BitOr for EventKinds {
215    type Output = Self;
216
217    fn bitor(self, other: Self) -> Self {
218        Self {
219            buttons: self.buttons || other.buttons,
220            motion: self.motion || other.motion,
221            keys: self.keys || other.keys,
222            media: self.media || other.media,
223        }
224    }
225}
226
227impl std::ops::BitAnd for EventKinds {
228    type Output = Self;
229
230    fn bitand(self, other: Self) -> Self {
231        Self {
232            buttons: self.buttons && other.buttons,
233            motion: self.motion && other.motion,
234            keys: self.keys && other.keys,
235            media: self.media && other.media,
236        }
237    }
238}
239
240#[derive(Clone, Copy, Debug, PartialEq, Eq)]
241#[non_exhaustive]
242pub enum DropSource {
243    /// the box's own counters
244    Device,
245    /// your receiver fell behind and its queue overflowed
246    Host,
247}
248
249#[derive(Clone, Copy, Debug, PartialEq, Eq)]
250#[non_exhaustive]
251pub struct Stamp {
252    /// the best guess at when the event happened, on the PC's clock. on Medius it's the box clock mapped onto the
253    /// PC's through the medius crate, which keeps it from ever going backwards and smooths it after bursts, so it's
254    /// an estimate there and NOT when the PC got the event, use `received` for that
255    pub arrived: Instant,
256    /// the same moment as the time since the device was opened, for code that can't use `Instant`
257    pub since_open: Duration,
258    pub box_time: Option<BoxTime>,
259    /// when anybox got the event. on MAKCU that's right after the port read returned with the report's last byte,
260    /// on Medius it's when anybox took it from the medius crate, which reads the port on its own thread, so it's a
261    /// bit after the read there. either way it's the read and not the USB arrival, driver buffering and a backlog
262    /// make it late, and reports that came in one read share it
263    pub received: Instant,
264    /// the same moment as `received`, as the time since the device was opened
265    pub received_since_open: Duration,
266    /// the raw QueryPerformanceCounter value at `received`, on Windows only, since an `Instant` can't be turned
267    /// back into one and a USB capture stamps its packets in QPC
268    pub received_qpc: Option<i64>,
269}
270
271impl Stamp {
272    /// for tests and for code that builds events itself. `received` starts out as `arrived`, and `since_open`
273    /// isn't checked against `arrived`
274    pub fn new(arrived: Instant, since_open: Duration) -> Self {
275        Self {
276            arrived,
277            since_open,
278            box_time: None,
279            received: arrived,
280            received_since_open: since_open,
281            received_qpc: None,
282        }
283    }
284
285    pub fn with_box_time(mut self, box_time: BoxTime) -> Self {
286        self.box_time = Some(box_time);
287        self
288    }
289
290    /// `received_since_open` comes out of `received` and `since_open`, so set those right first
291    pub fn with_received(mut self, received: Instant, qpc: Option<i64>) -> Self {
292        self.received = received;
293        self.received_since_open = if received >= self.arrived {
294            self.since_open + received.duration_since(self.arrived)
295        } else {
296            self.since_open
297                .saturating_sub(self.arrived.duration_since(received))
298        };
299        self.received_qpc = qpc;
300        self
301    }
302}
303
304#[derive(Clone, Copy, Debug, PartialEq, Eq)]
305#[non_exhaustive]
306pub struct BoxTime {
307    pub micros: u32,
308    pub clock: Clock,
309}
310
311impl BoxTime {
312    pub fn new(micros: u32, clock: Clock) -> Self {
313        Self { micros, clock }
314    }
315}
316
317/// each box clock is its own counter, so only compare stamps that come from the same one
318#[derive(Clone, Copy, Debug, PartialEq, Eq)]
319#[non_exhaustive]
320pub enum Clock {
321    MediusHostChip,
322    MediusDeviceChip,
323}
324
325impl Clock {
326    /// a u32 microsecond counter rolls over every 71.6 minutes
327    pub fn wraps_after(self) -> Duration {
328        Duration::from_micros(u64::from(u32::MAX) + 1)
329    }
330}
331
332#[derive(Clone, Copy, Debug, PartialEq, Eq)]
333pub struct Closed;
334
335impl fmt::Display for Closed {
336    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
337        f.write_str("the device is gone, no more events will come")
338    }
339}
340
341impl std::error::Error for Closed {}
342
343#[derive(Clone, Copy, Debug, PartialEq, Eq)]
344pub enum RecvTimeoutError {
345    Timeout,
346    Closed,
347}
348
349impl fmt::Display for RecvTimeoutError {
350    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
351        match self {
352            Self::Timeout => f.write_str("no event came in time"),
353            Self::Closed => Closed.fmt(f),
354        }
355    }
356}
357
358impl std::error::Error for RecvTimeoutError {}
359
360#[derive(Clone, Copy, Debug, PartialEq, Eq)]
361pub enum TryRecvError {
362    Empty,
363    Closed,
364}
365
366impl fmt::Display for TryRecvError {
367    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
368        match self {
369            Self::Empty => f.write_str("no event is waiting"),
370            Self::Closed => Closed.fmt(f),
371        }
372    }
373}
374
375impl std::error::Error for TryRecvError {}
376
377pub(crate) const QUEUE_CAPACITY: usize = 1024;
378
379#[derive(Debug)]
380struct State {
381    events: VecDeque<Event>,
382    closed: bool,
383}
384
385/// one receiver's queue. when it's full the oldest events get thrown out and one `Dropped` at the front counts
386/// them, so a busy producer can never push that report out. only input events get thrown out, never a `Dropped`
387/// from the box or a disconnect or reconnect, so a slow reader still finds out that what it held got let go
388#[derive(Debug)]
389pub(crate) struct Queue {
390    state: Mutex<State>,
391    ready: Condvar,
392    capacity: usize,
393}
394
395impl Queue {
396    pub(crate) fn new(capacity: usize) -> Self {
397        Self {
398            state: Mutex::new(State {
399                events: VecDeque::new(),
400                closed: false,
401            }),
402            ready: Condvar::new(),
403            capacity: capacity.max(2),
404        }
405    }
406
407    fn lock(&self) -> MutexGuard<'_, State> {
408        self.state.lock().unwrap_or_else(PoisonError::into_inner)
409    }
410
411    pub(crate) fn push(&self, event: Event) {
412        let mut state = self.lock();
413        if state.closed {
414            return;
415        }
416        let events = &mut state.events;
417        if let Event::Rejected { action, count } = event {
418            // an unread refusal of the same action just gets its count bumped, so refusals take a few slots at most
419            let unread = events.iter_mut().find_map(|queued| match queued {
420                Event::Rejected {
421                    action: queued_action,
422                    count: queued_count,
423                } if *queued_action == action => Some(queued_count),
424                _ => None,
425            });
426            if let Some(queued_count) = unread {
427                *queued_count += count;
428                return;
429            }
430        }
431        if events.len() >= self.capacity {
432            let counted = matches!(
433                events.front(),
434                Some(Event::Dropped {
435                    source: DropSource::Host,
436                    ..
437                })
438            );
439            if !counted {
440                events.push_front(Event::Dropped {
441                    count: 0,
442                    source: DropSource::Host,
443                });
444            }
445            let motion_queued = events.iter().any(|e| matches!(e, Event::Motion { .. }));
446            if matches!(event, Event::Motion { .. }) && !motion_queued {
447                // a queue full of presses and releases keeps them and the new motion gets thrown out instead.
448                // the `Dropped` at the front still needs a slot though, so the oldest press or release makes room
449                // for it once
450                let mut lost = 1;
451                if events.len() > self.capacity
452                    && let Some(oldest) = events.iter().skip(1).position(|e| {
453                        matches!(
454                            e,
455                            Event::Button { .. } | Event::Key { .. } | Event::Media { .. }
456                        )
457                    })
458                {
459                    events.remove(oldest + 1);
460                    lost += 1;
461                }
462                if let Some(Event::Dropped { count, .. }) = events.front_mut() {
463                    *count += lost;
464                }
465                drop(state);
466                self.ready.notify_all();
467                return;
468            }
469            while events.len() >= self.capacity {
470                // motion gets thrown out first, otherwise a flood of it could push out the release of a key that's
471                // still held
472                let motion = events
473                    .iter()
474                    .skip(1)
475                    .position(|e| matches!(e, Event::Motion { .. }));
476                let edge = || {
477                    events.iter().skip(1).position(|e| {
478                        matches!(
479                            e,
480                            Event::Button { .. } | Event::Key { .. } | Event::Media { .. }
481                        )
482                    })
483                };
484                let Some(oldest) = motion.or_else(edge) else {
485                    break;
486                };
487                events.remove(oldest + 1);
488                if let Some(Event::Dropped { count, .. }) = events.front_mut() {
489                    *count += 1;
490                }
491            }
492        }
493        events.push_back(event);
494        drop(state);
495        self.ready.notify_all();
496    }
497
498    pub(crate) fn close(&self) {
499        self.lock().closed = true;
500        self.ready.notify_all();
501    }
502
503    fn try_pop(state: &mut State) -> Result<Event, TryRecvError> {
504        match state.events.pop_front() {
505            Some(event) => Ok(event),
506            None if state.closed => Err(TryRecvError::Closed),
507            None => Err(TryRecvError::Empty),
508        }
509    }
510}
511
512/// lets a receiver tell the device when it's dropped, so the box only streams input while someone listens
513pub(crate) trait Listeners: Send + Sync {
514    fn receiver_dropped(&self, id: u64);
515}
516
517/// a receiver of events from one device. it ends once the device is gone and you've read everything queued
518pub struct Events {
519    queue: Arc<Queue>,
520    listeners: Weak<dyn Listeners>,
521    id: u64,
522}
523
524impl Events {
525    pub(crate) fn new(queue: Arc<Queue>, listeners: Weak<dyn Listeners>, id: u64) -> Self {
526        Self {
527            queue,
528            listeners,
529            id,
530        }
531    }
532
533    pub fn recv(&self) -> Result<Event, Closed> {
534        let mut state = self.queue.lock();
535        loop {
536            match Queue::try_pop(&mut state) {
537                Ok(event) => return Ok(event),
538                Err(TryRecvError::Closed) => return Err(Closed),
539                Err(TryRecvError::Empty) => {
540                    state = self
541                        .queue
542                        .ready
543                        .wait(state)
544                        .unwrap_or_else(PoisonError::into_inner);
545                }
546            }
547        }
548    }
549
550    pub fn recv_timeout(&self, timeout: Duration) -> Result<Event, RecvTimeoutError> {
551        let deadline = Instant::now() + timeout;
552        let mut state = self.queue.lock();
553        loop {
554            match Queue::try_pop(&mut state) {
555                Ok(event) => return Ok(event),
556                Err(TryRecvError::Closed) => return Err(RecvTimeoutError::Closed),
557                Err(TryRecvError::Empty) => {
558                    let left = deadline.saturating_duration_since(Instant::now());
559                    if left.is_zero() {
560                        return Err(RecvTimeoutError::Timeout);
561                    }
562                    state = self
563                        .queue
564                        .ready
565                        .wait_timeout(state, left)
566                        .unwrap_or_else(PoisonError::into_inner)
567                        .0;
568                }
569            }
570        }
571    }
572
573    pub fn try_recv(&self) -> Result<Event, TryRecvError> {
574        Queue::try_pop(&mut self.queue.lock())
575    }
576}
577
578impl Iterator for Events {
579    type Item = Event;
580
581    fn next(&mut self) -> Option<Event> {
582        self.recv().ok()
583    }
584}
585
586impl Drop for Events {
587    fn drop(&mut self) {
588        if let Some(listeners) = self.listeners.upgrade() {
589            listeners.receiver_dropped(self.id);
590        }
591    }
592}
593
594impl fmt::Debug for Events {
595    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
596        f.debug_struct("Events").finish_non_exhaustive()
597    }
598}
599
600impl std::ops::Not for EventKinds {
601    type Output = Self;
602
603    fn not(self) -> Self {
604        Self {
605            buttons: !self.buttons,
606            motion: !self.motion,
607            keys: !self.keys,
608            media: !self.media,
609        }
610    }
611}
612
613#[cfg(test)]
614mod tests;
anybox/event.rs

anybox/
event.rs
