<NME> x11rb.rs
<BEF> use std::{
    io::Result as IOResult,
    sync::Arc,
};

use x11rb::{
    connection::Connection as _,
    protocol::{
        xproto::{Atom, ClientMessageEvent, ConnectionExt as _, EventMask, Window, CLIENT_MESSAGE_EVENT},
        Event,
    },
    rust_connection::RustConnection,
    rust_connection::RustConnection,
};

use smithay::reexports::calloop::{
    channel::{sync_channel, Channel, Event as ChannelEvent, SyncSender},
    EventSource, Poll, PostAction, Readiness, Token, TokenFactory,
};
/// Integration of an x11rb X11 connection with calloop.
///
/// This is a thin wrapper around `Channel`. It works by spawning an extra thread reads events from
/// the X11 connection and then sends them across the channel.
///
/// See [1] for why this extra thread is necessary. The single-thread solution proposed on that
/// page does not work with calloop, since it requires checking something on every main loop
/// iteration. Calloop only allows "when an FD becomes readable".
/// iteration. Calloop only allows "when an FD becomes readable".
///
/// [1]: https://docs.rs/x11rb/0.8.1/x11rb/event_loop_integration/index.html#threads-and-races
pub struct X11Source {
    connection: Arc<RustConnection>,
    channel: Channel<Event>,
    event_thread: Option<JoinHandle<()>>,
    close_window: Window,
    close_type: Atom,
}

impl X11Source {
    /// Create a new X11 source.
    ///
    /// The returned instance will use `SendRequest` to cause a `ClientMessageEvent` to be sent to
    /// the given window with the given type. The expectation is that this is a window that was
    /// created by us. Thus, the event reading thread will wake up and check an internal exit flag,
    /// then exit.
    pub fn new(
        connection: Arc<RustConnection>,
        close_window: Window,
        close_type: Atom,
        log: slog::Logger,
    ) -> Self {
        let (sender, channel) = sync_channel(5);
        let conn = Arc::clone(&connection);
        let log2 = log.clone();
        let event_thread = Some(spawn(move || {
        let event_thread = Some(spawn(move || {
            run_event_thread(conn, sender, log2);
        }));
        Self {
            connection,
            channel,
            event_thread,
            close_window,
            close_type,
        }
    }
}

impl Drop for X11Source {
impl Drop for X11Source {
    fn drop(&mut self) {
        // Signal the worker thread to exit by dropping the read end of the channel.
        // There is no easy and nice way to do this, so do it the ugly way: Replace it.
        let (_, channel) = sync_channel(1);
        self.channel = channel;

        // Send an event to wake up the worker so that it actually exits
        let event = ClientMessageEvent {
            format: 8,
            sequence: 0,
            window: self.close_window,
            type_: self.close_type,
            type_: self.close_type,
            data: [0; 20].into(),
        };
        let _ = self
            .connection
            .send_event(false, self.close_window, EventMask::NO_EVENT, event);

        // Wait for the worker thread to exit
        self.event_thread.take().map(|handle| handle.join());
    }
}

impl EventSource for X11Source {
    type Event = Event;
    type Metadata = ();
    type Ret = ();
    type Error = ChannelError;

    fn process_events<C>(
        &mut self,
        readiness: Readiness,
        token: Token,
        mut callback: C,
    ) -> Result<PostAction, ChannelError>
    where
        C: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
    {
        let log = self.log.clone();
        self.channel
            .process_events(readiness, token, move |event, meta| match event {
                ChannelEvent::Closed => slog::warn!(log, "Event thread exited"),
                ChannelEvent::Msg(event) => callback(event, meta),
            })
    }

    fn register(&mut self, poll: &mut Poll, factory: &mut TokenFactory) -> IOResult<()> {
        self.channel.register(poll, factory)
    }

    fn reregister(&mut self, poll: &mut Poll, factory: &mut TokenFactory) -> IOResult<()> {
        self.channel.reregister(poll, factory)
    }

    fn unregister(&mut self, poll: &mut Poll) -> IOResult<()> {
        self.channel.unregister(poll)
    }
}

/// This thread reads X11 events from the connection and sends them on the channel.
///
/// This is run in an extra thread since sending an X11 request or waiting for the reply to an X11
/// request can both read X11 events from the underlying socket which are then saved in the
/// RustConnection. Thus, readability of the underlying socket is not enough to guarantee we do not
/// miss wakeups.
///
/// This thread will call wait_for_event(). RustConnection then ensures internally to wake us up
/// when an event arrives. So far, this seems to be the only safe way to integrate x11rb with
/// calloop.
fn run_event_thread(connection: Arc<RustConnection>, sender: SyncSender<Event>, log: slog::Logger) {
    loop {
        let event = match connection.wait_for_event() {
            Ok(event) => event,
            Err(err) => {
                // Connection errors are most likely permanent. Thus, exit the thread.
                slog::crit!(log, "Event thread exiting due to connection error {}", err);
                break;
            }
        };
        match sender.send(event) {
            Ok(()) => {}
            Err(_) => {
                // The only possible error is that the other end of the channel was dropped.
                // This happens in X11Source's Drop impl.
                break;
            }
        }
    }
}

<MSG> Move X11Source to smithay under`x11rb_event_source` feature


<DFF> @@ -1,3 +1,8 @@
+//! Helper utilities for using x11rb as an event source in calloop.
+//!
+//! The primary use for this module is XWayland integration but is also widely useful for an X11
+//! backend in a compositor.
+
 use std::{
     io::Result as IOResult,
     sync::Arc,
@@ -13,7 +18,7 @@ use x11rb::{
     rust_connection::RustConnection,
 };
 
-use smithay::reexports::calloop::{
+use calloop::{
     channel::{sync_channel, Channel, Event as ChannelEvent, SyncSender},
     EventSource, Poll, PostAction, Readiness, Token, TokenFactory,
 };
@@ -28,9 +33,10 @@ use smithay::reexports::calloop::{
 /// iteration. Calloop only allows "when an FD becomes readable".
 ///
 /// [1]: https://docs.rs/x11rb/0.8.1/x11rb/event_loop_integration/index.html#threads-and-races
+#[derive(Debug)]
 pub struct X11Source {
     connection: Arc<RustConnection>,
-    channel: Channel<Event>,
+    channel: Option<Channel<Event>>,
     event_thread: Option<JoinHandle<()>>,
     close_window: Window,
     close_type: Atom,
@@ -56,9 +62,10 @@ impl X11Source {
         let event_thread = Some(spawn(move || {
             run_event_thread(conn, sender, log2);
         }));
+
         Self {
             connection,
-            channel,
+            channel: Some(channel),
             event_thread,
             close_window,
             close_type,
@@ -70,9 +77,7 @@ impl X11Source {
 impl Drop for X11Source {
     fn drop(&mut self) {
         // Signal the worker thread to exit by dropping the read end of the channel.
-        // There is no easy and nice way to do this, so do it the ugly way: Replace it.
-        let (_, channel) = sync_channel(1);
-        self.channel = channel;
+        self.channel.take();
 
         // Send an event to wake up the worker so that it actually exits
         let event = ClientMessageEvent {
@@ -83,6 +88,7 @@ impl Drop for X11Source {
             type_: self.close_type,
             data: [0; 20].into(),
         };
+
         let _ = self
             .connection
             .send_event(false, self.close_window, EventMask::NO_EVENT, event);
@@ -108,23 +114,39 @@ impl EventSource for X11Source {
         C: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
     {
         let log = self.log.clone();
-        self.channel
-            .process_events(readiness, token, move |event, meta| match event {
+
+        if let Some(channel) = &mut self.channel {
+            channel.process_events(readiness, token, move |event, meta| match event {
                 ChannelEvent::Closed => slog::warn!(log, "Event thread exited"),
                 ChannelEvent::Msg(event) => callback(event, meta),
             })
+        } else {
+            Ok(PostAction::Remove)
+        }
     }
 
     fn register(&mut self, poll: &mut Poll, factory: &mut TokenFactory) -> IOResult<()> {
-        self.channel.register(poll, factory)
+        if let Some(channel) = &mut self.channel {
+            channel.register(poll, factory)?;
+        }
+
+        Ok(())
     }
 
     fn reregister(&mut self, poll: &mut Poll, factory: &mut TokenFactory) -> IOResult<()> {
-        self.channel.reregister(poll, factory)
+        if let Some(channel) = &mut self.channel {
+            channel.reregister(poll, factory)?;
+        }
+
+        Ok(())
     }
 
     fn unregister(&mut self, poll: &mut Poll) -> IOResult<()> {
-        self.channel.unregister(poll)
+        if let Some(channel) = &mut self.channel {
+            channel.unregister(poll)?;
+        }
+
+        Ok(())
     }
 }
 

Move X11Source to smithay under`x11rb_event_source` feature