anybox/
event.rs

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

anybox/
event.rs
