colpetto 0.1.2

Async libinput wrapper
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use std::{
    os::fd::RawFd,
    pin::Pin,
    task::{self, Poll},
};

use futures_core::{ready, Stream};
use tokio::io::unix::AsyncFd;

use crate::{Error, Event, Libinput, Result};

/// An asynchronous stream of libinput events integrated with the tokio runtime.
///
/// `EventStream` provides an asynchronous interface to libinput's event system, allowing
/// you to receive input device events using tokio's async/await syntax. It implements
/// the [`Stream`] trait, making it compatible with tokio's async ecosystem.
///
/// Events include device additions/removals, as well as input events like key presses,
/// pointer movements, touch events, etc. The stream will yield these events as they
/// occur from the underlying libinput context.
///
/// # Initial Events
///
/// When first polling the stream, it will immediately yield any pending device creation
/// events before waiting for new events. This ensures you receive information about
/// all currently connected devices.
///
/// # Example usage
///
/// ```
/// use tokio_stream::StreamExt;
///
/// /* libinput initialization is omited for brevity */
///
/// let mut stream = libinput.event_stream()?;
///
/// while let Some(event) = stream.try_next().await? {
///     println!(
///         "Got \"{}\" event from \"{}\"",
///         event.event_type(),
///         event.device().name().to_string_lossy()
///     )
/// }
/// ```
///
/// # Implementation Note
///
/// The stream internally manages an [`AsyncFd`] wrapper around the libinput file descriptor,
/// ensuring efficient integration with tokio's event loop. It will only wake up when new
/// events are available to be read from the libinput context.
pub struct EventStream {
    libinput: Libinput,
    fd: AsyncFd<i32>,
    is_first: bool,
}

impl EventStream {
    pub(crate) fn new(libinput: Libinput, fd: RawFd) -> Result<Self> {
        Ok(Self {
            libinput,
            fd: AsyncFd::new(fd)?,
            is_first: true,
        })
    }
}

impl Stream for EventStream {
    type Item = Result<Event, Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
        loop {
            // The first time we poll there is already device created events available
            if self.is_first {
                self.libinput.dispatch()?;

                if let Some(event) = self.libinput.get_event() {
                    return Poll::Ready(Some(Ok(event)));
                } else {
                    self.is_first = false;
                    continue;
                }
            }

            let mut guard = ready!(self.fd.poll_read_ready(cx))?;
            self.libinput.dispatch()?;

            if let Some(event) = self.libinput.get_event() {
                return Poll::Ready(Some(Ok(event)));
            } else {
                guard.clear_ready();
                continue;
            }
        }
    }
}