Feature: calloop EventSource implementation

Cargo.toml

@@ -50,6 +50,9 @@ derive-more = ["dep:derive_more"]
 ## Enables interacting with a host clipboard via [`clipboard`](clipboard/index.html)
 osc52 = ["dep:base64"]
 
+## Enables calloop [EventSource](calloop::EventSource) support.
+calloop = ["dep:calloop", "rustix/process"]
+
 [dependencies]
 base64 = { version = "0.22", optional = true }
 bitflags = { version = "2.9" }
@@ -74,6 +77,8 @@ mio = { version = "1.0", features = ["os-poll"], optional = true }
 rustix = { version = "1", default-features = false, features = ["std", "stdio", "termios"] }
 signal-hook = { version = "0.3.17", optional = true }
 signal-hook-mio = { version = "0.2.4", features = ["support-v1_0"], optional = true }
+# runtime
+calloop = { version = "0.14.2", optional = true }
 
 [dev-dependencies]
 async-std = "1.13"

examples/cl_demo.rs

@@ -0,0 +1,92 @@
+use std::io;
+
+use calloop::EventLoop;
+use crossterm::{
+    event::{
+        runtime::calloop::UnixEventSource, DisableBracketedPaste, DisableFocusChange,
+        DisableMouseCapture, EnableBracketedPaste, EnableFocusChange, EnableMouseCapture,
+        KeyboardEnhancementFlags, PopKeyboardEnhancementFlags, PushKeyboardEnhancementFlags,
+    },
+    execute, queue,
+    terminal::{disable_raw_mode, enable_raw_mode},
+};
+
+struct LoopData {
+    exit: bool,
+}
+
+fn event_loop() {
+    let mut el = EventLoop::try_new().unwrap();
+    el.handle()
+        .insert_source(
+            UnixEventSource::new().unwrap(),
+            |es, _, data: &mut LoopData| {
+                println!("{:?}\r", es);
+                es.iter().for_each(|e| {
+                    if let crossterm::event::Event::Key(key_event) = e {
+                        if key_event.code.is_esc() {
+                            data.exit = true;
+                        }
+                    }
+                });
+                Ok(())
+            },
+        )
+        .unwrap();
+
+    let mut a = LoopData { exit: false };
+    loop {
+        el.dispatch(None, &mut a).unwrap();
+        if a.exit {
+            break;
+        }
+    }
+}
+
+fn main() {
+    enable_raw_mode().unwrap();
+
+    let mut stdout = io::stdout();
+
+    let supports_keyboard_enhancement = matches!(
+        crossterm::terminal::supports_keyboard_enhancement(),
+        Ok(true)
+    );
+
+    if supports_keyboard_enhancement {
+        queue!(
+            stdout,
+            PushKeyboardEnhancementFlags(
+                KeyboardEnhancementFlags::DISAMBIGUATE_ESCAPE_CODES
+                    | KeyboardEnhancementFlags::REPORT_ALL_KEYS_AS_ESCAPE_CODES
+                    | KeyboardEnhancementFlags::REPORT_ALTERNATE_KEYS
+                    | KeyboardEnhancementFlags::REPORT_EVENT_TYPES
+            )
+        )
+        .unwrap();
+    }
+
+    execute!(
+        stdout,
+        EnableBracketedPaste,
+        EnableFocusChange,
+        EnableMouseCapture,
+    )
+    .unwrap();
+
+    event_loop();
+
+    if supports_keyboard_enhancement {
+        queue!(stdout, PopKeyboardEnhancementFlags).unwrap();
+    }
+
+    execute!(
+        stdout,
+        DisableBracketedPaste,
+        DisableFocusChange,
+        DisableMouseCapture
+    )
+    .unwrap();
+
+    disable_raw_mode().unwrap();
+}

src/event.rs

@@ -126,6 +126,8 @@
 pub(crate) mod sys;
 pub(crate) mod timeout;
 
+pub mod runtime;
+
 #[cfg(feature = "derive-more")]
 use derive_more::derive::IsVariant;
 #[cfg(feature = "event-stream")]
@@ -1446,11 +1448,11 @@
             KeyCode::Tab => write!(f, "Tab"),
             KeyCode::BackTab => write!(f, "Back Tab"),
             KeyCode::Insert => write!(f, "Insert"),
             KeyCode::F(n) => write!(f, "F{}", n),
             KeyCode::Char(c) => match c {
                 // special case for non-visible characters
                 ' ' => write!(f, "Space"),
                 c => write!(f, "{}", c),
             },
             KeyCode::Null => write!(f, "Null"),
             KeyCode::Esc => write!(f, "Esc"),
@@ -1461,8 +1463,8 @@
             KeyCode::Pause => write!(f, "Pause"),
             KeyCode::Menu => write!(f, "Menu"),
             KeyCode::KeypadBegin => write!(f, "Begin"),
             KeyCode::Media(media) => write!(f, "{}", media),
             KeyCode::Modifier(modifier) => write!(f, "{}", modifier),
         }
     }
 }

src/event/runtime/calloop.rs

@@ -0,0 +1,247 @@
+#[cfg(not(feature = "libc"))]
+use std::os::fd::AsFd;
+#[cfg(feature = "libc")]
+use std::os::fd::AsRawFd;
+use std::{collections::VecDeque, io, os::unix::net::UnixStream};
+
+use calloop::{generic::Generic, EventSource, Interest, Mode, Poll, PostAction, TokenFactory};
+use signal_hook::low_level::pipe;
+
+use crate::{
+    event::{sys::unix::parse::parse_event, Event, InternalEvent},
+    terminal::sys::file_descriptor::{tty_fd, FileDesc},
+};
+
+// I (@zrzka) wasn't able to read more than 1_022 bytes when testing
+// reading on macOS/Linux -> we don't need bigger buffer and 1k of bytes
+// is enough.
+const TTY_BUFFER_SIZE: usize = 1_024;
+
+fn nonblocking_unix_pair() -> io::Result<(UnixStream, UnixStream)> {
+    let (receiver, sender) = UnixStream::pair()?;
+    receiver.set_nonblocking(true)?;
+    sender.set_nonblocking(true)?;
+    Ok((receiver, sender))
+}
+
+pub struct UnixEventSource {
+    parser: Parser,
+    tty_buffer: [u8; TTY_BUFFER_SIZE],
+    #[cfg(not(feature = "libc"))]
+    tty_source: Generic<FileDesc<'static>>,
+    #[cfg(feature = "libc")]
+    tty_source: Generic<calloop::generic::FdWrapper<FileDesc<'static>>>,
+    sig_source: Generic<UnixStream>,
+}
+
+impl UnixEventSource {
+    pub fn new() -> io::Result<Self> {
+        Ok(UnixEventSource {
+            parser: Parser::default(),
+            tty_buffer: [0u8; TTY_BUFFER_SIZE],
+            tty_source: {
+                let fd = {
+                    #[cfg(feature = "libc")]
+                    unsafe {
+                        calloop::generic::FdWrapper::new(tty_fd()?)
+                    }
+                    #[cfg(not(feature = "libc"))]
+                    tty_fd()?
+                };
+                Generic::new(fd, Interest::READ, Mode::Edge)
+            },
+            sig_source: {
+                let (receiver, sender) = nonblocking_unix_pair()?;
+                #[cfg(feature = "libc")]
+                pipe::register(libc::SIGWINCH, sender)?;
+                #[cfg(not(feature = "libc"))]
+                pipe::register(rustix::process::Signal::WINCH.as_raw(), sender)?;
+                Generic::new(receiver, Interest::READ, Mode::Edge)
+            },
+        })
+    }
+}
+
+impl EventSource for UnixEventSource {
+    type Event = Vec<Event>;
+    type Metadata = ();
+    type Ret = io::Result<()>;
+    type Error = io::Error;
+
+    fn register(
+        &mut self,
+        poll: &mut Poll,
+        factory: &mut TokenFactory,
+    ) -> Result<(), calloop::Error> {
+        self.tty_source.register(poll, factory)?;
+        self.sig_source.register(poll, factory)?;
+        Ok(())
+    }
+
+    fn reregister(
+        &mut self,
+        poll: &mut Poll,
+        factory: &mut TokenFactory,
+    ) -> Result<(), calloop::Error> {
+        self.tty_source.reregister(poll, factory)?;
+        self.sig_source.reregister(poll, factory)?;
+        Ok(())
+    }
+
+    fn unregister(&mut self, poll: &mut Poll) -> Result<(), calloop::Error> {
+        self.tty_source.unregister(poll)?;
+        self.sig_source.unregister(poll)?;
+        Ok(())
+    }
+
+    fn process_events<F>(
+        &mut self,
+        readiness: calloop::Readiness,
+        token: calloop::Token,
+        mut callback: F,
+    ) -> Result<PostAction, Self::Error>
+    where
+        F: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
+    {
+        self.tty_source.process_events(readiness, token, |_, f| {
+            loop {
+                let read_count = read_complete(f, &mut self.tty_buffer)?;
+                if read_count > 0 {
+                    self.parser.advance(
+                        &self.tty_buffer[..read_count],
+                        read_count == TTY_BUFFER_SIZE,
+                    );
+                }
+
+                if !self.parser.internal_events.is_empty() {
+                    break;
+                }
+
+                if read_count == 0 {
+                    break;
+                }
+            }
+
+            Ok(calloop::PostAction::Continue)
+        })?;
+
+        self.sig_source.process_events(readiness, token, |_, f| {
+            #[cfg(feature = "libc")]
+            let fd = FileDesc::new(f.as_raw_fd(), false);
+            #[cfg(not(feature = "libc"))]
+            let fd = FileDesc::Borrowed(f.as_fd());
+
+            // drain the pipe
+            while read_complete(&fd, &mut [0; 1024])? != 0 {}
+
+            let new_size = crate::terminal::size()?;
+            self.parser
+                .internal_events
+                .push_back(InternalEvent::Event(Event::Resize(new_size.0, new_size.1)));
+
+            Ok(calloop::PostAction::Continue)
+        })?;
+
+        if !self.parser.internal_events.is_empty() {
+            let public_events: Self::Event = self
+                .parser
+                .take_events()
+                .into_iter()
+                .filter_map(|e| match e {
+                    InternalEvent::Event(event) => Some(event),
+                    _ => None,
+                })
+                .collect();
+            if !public_events.is_empty() {
+                callback(public_events, &mut ()).unwrap();
+            }
+        };
+
+        Ok(PostAction::Continue)
+    }
+}
+
+/// read_complete reads from a non-blocking file descriptor
+/// until the buffer is full or it would block.
+///
+/// Similar to `std::io::Read::read_to_end`, except this function
+/// only fills the given buffer and does not read beyond that.
+fn read_complete(fd: &FileDesc, buf: &mut [u8]) -> io::Result<usize> {
+    loop {
+        match fd.read(buf) {
+            Ok(x) => return Ok(x),
+            Err(e) => match e.kind() {
+                io::ErrorKind::WouldBlock => return Ok(0),
+                io::ErrorKind::Interrupted => continue,
+                _ => return Err(e),
+            },
+        }
+    }
+}
+
+//
+// Following `Parser` structure exists for two reasons:
+//
+//  * mimic anes Parser interface
+//  * move the advancing, parsing, ... stuff out of the `try_read` method
+//
+#[derive(Debug)]
+struct Parser {
+    buffer: Vec<u8>,
+    internal_events: VecDeque<InternalEvent>,
+}
+
+impl Default for Parser {
+    fn default() -> Self {
+        Parser {
+            // This buffer is used for -> 1 <- ANSI escape sequence. Are we
+            // aware of any ANSI escape sequence that is bigger? Can we make
+            // it smaller?
+            //
+            // Probably not worth spending more time on this as "there's a plan"
+            // to use the anes crate parser.
+            buffer: Vec::with_capacity(256),
+            // TTY_BUFFER_SIZE is 1_024 bytes. How many ANSI escape sequences can
+            // fit? What is an average sequence length? Let's guess here
+            // and say that the average ANSI escape sequence length is 8 bytes. Thus
+            // the buffer size should be 1024/8=128 to avoid additional allocations
+            // when processing large amounts of data.
+            //
+            // There's no need to make it bigger, because when you look at the `try_read`
+            // method implementation, all events are consumed before the next TTY_BUFFER
+            // is processed -> events pushed.
+            internal_events: VecDeque::with_capacity(128),
+        }
+    }
+}
+
+impl Parser {
+    fn advance(&mut self, buffer: &[u8], more: bool) {
+        for (idx, byte) in buffer.iter().enumerate() {
+            let more = idx + 1 < buffer.len() || more;
+
+            self.buffer.push(*byte);
+
+            match parse_event(&self.buffer, more) {
+                Ok(Some(ie)) => {
+                    self.internal_events.push_back(ie);
+                    self.buffer.clear();
+                }
+                Ok(None) => {
+                    // Event can't be parsed, because we don't have enough bytes for
+                    // the current sequence. Keep the buffer and process next bytes.
+                }
+                Err(_) => {
+                    // Event can't be parsed (not enough parameters, parameter is not a number, ...).
+                    // Clear the buffer and continue with another sequence.
+                    self.buffer.clear();
+                }
+            }
+        }
+    }
+    fn take_events(&mut self) -> VecDeque<InternalEvent> {
+        let mut es = std::mem::replace(&mut self.internal_events, VecDeque::with_capacity(128));
+        es.shrink_to_fit();
+        es
+    }
+}

src/event/runtime/mod.rs

@@ -0,0 +1,6 @@
+// here we do runtime specific implementations
+
+/// [calloop] implementation
+#[cfg(unix)]
+#[cfg(feature = "calloop")]
+pub mod calloop;