compio_term/event/
stream.rs

1use std::{
2    io,
3    pin::Pin,
4    sync::atomic::{AtomicBool, Ordering},
5    task::{Context, Poll, ready},
6};
7
8use futures_core::Stream;
9
10use super::{Event, InternalEvent, sys::EventSource};
11
12static ACTIVE_READER: AtomicBool = AtomicBool::new(false);
13
14/// A local asynchronous stream of terminal events.
15///
16/// The stream must be created while a Compio runtime is active. It does not
17/// enable raw mode or any optional terminal event modes. Use Crossterm's
18/// terminal and event commands to configure those modes before polling.
19///
20/// Terminal input is a process-wide resource. Only one `EventStream` can be
21/// active at a time. A second call to [`EventStream::new`] returns
22/// [`io::ErrorKind::AlreadyExists`]. Dropping the stream cancels its pending
23/// Compio operation and permits a new stream.
24#[must_use = "streams do nothing unless polled"]
25pub struct EventStream {
26    source: EventSource,
27    _lease: ReaderLease,
28}
29
30impl EventStream {
31    /// Creates an event stream for the process terminal.
32    ///
33    /// # Errors
34    ///
35    /// Returns an error when there is no active Compio runtime, another event
36    /// stream is active, or the process terminal cannot be opened.
37    pub fn new() -> io::Result<Self> {
38        let lease = ReaderLease::acquire()?;
39        let source = EventSource::new()?;
40        Ok(Self {
41            source,
42            _lease: lease,
43        })
44    }
45}
46
47impl Stream for EventStream {
48    type Item = io::Result<Event>;
49
50    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
51        loop {
52            match ready!(Pin::new(&mut self.source).poll_next(cx)) {
53                Some(Ok(InternalEvent::Event(event))) => {
54                    return Poll::Ready(Some(Ok(event)));
55                }
56                Some(Ok(InternalEvent::Ignored)) => {}
57                Some(Err(error)) => return Poll::Ready(Some(Err(error))),
58                None => return Poll::Ready(None),
59            }
60        }
61    }
62}
63
64struct ReaderLease;
65
66impl ReaderLease {
67    fn acquire() -> io::Result<Self> {
68        ACTIVE_READER
69            .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
70            .map_err(|_| {
71                io::Error::new(
72                    io::ErrorKind::AlreadyExists,
73                    "a terminal event stream is already active",
74                )
75            })?;
76        Ok(Self)
77    }
78}
79
80impl Drop for ReaderLease {
81    fn drop(&mut self) {
82        ACTIVE_READER.store(false, Ordering::Release);
83    }
84}
compio_term/event/stream.rs

compio_term/event/
stream.rs
