wip use calloop event loop

Cargo.lock

@@ -390,20 +390,23 @@ name = "cosmic-applet-workspaces"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "calloop",
  "cascade",
  "cosmic-panel-config",
  "gio",
  "gtk4",
  "i18n-embed",
  "i18n-embed-fl",
+ "log",
+ "nix 0.22.3",
  "once_cell",
  "pretty_env_logger",
  "rust-embed",
  "tokio",
  "wayland-backend",
- "wayland-client 0.30.0-beta.4",
+ "wayland-client 0.30.0-beta.5",
  "wayland-commons",
- "wayland-scanner 0.30.0-beta.4",
+ "wayland-scanner 0.30.0-beta.5",
 ]
 
 [[package]]
@@ -425,7 +428,7 @@ dependencies = [
 [[package]]
 name = "cosmic-panel-config"
 version = "0.1.0"
-source = "git+https://github.com/pop-os/cosmic-panel/#231dc1ec0656840458d9f0d3468d9c7ea5c2a98c"
+source = "git+https://github.com/pop-os/cosmic-panel#231dc1ec0656840458d9f0d3468d9c7ea5c2a98c"
 dependencies = [
  "anyhow",
  "gtk4",
@@ -2317,8 +2320,8 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
 
 [[package]]
 name = "wayland-backend"
-version = "0.1.0-beta.4"
-source = "git+https://github.com/smithay/wayland-rs.git#797f697a66f8541327cbaac319692ec8d2395a38"
+version = "0.1.0-beta.5"
+source = "git+https://github.com/smithay/wayland-rs.git#70b0389becc49edde19ec16779eec3c5b0ad0a5f"
 dependencies = [
  "cc",
  "downcast-rs",
@@ -2326,7 +2329,7 @@ dependencies = [
  "nix 0.24.1",
  "scoped-tls",
  "smallvec",
- "wayland-sys 0.30.0-beta.4",
+ "wayland-sys 0.30.0-beta.5",
 ]
 
 [[package]]
@@ -2347,8 +2350,8 @@ dependencies = [
 
 [[package]]
 name = "wayland-client"
-version = "0.30.0-beta.4"
-source = "git+https://github.com/smithay/wayland-rs.git#797f697a66f8541327cbaac319692ec8d2395a38"
+version = "0.30.0-beta.5"
+source = "git+https://github.com/smithay/wayland-rs.git#70b0389becc49edde19ec16779eec3c5b0ad0a5f"
 dependencies = [
  "bitflags",
  "futures-channel",
@@ -2357,7 +2360,7 @@ dependencies = [
  "nix 0.24.1",
  "thiserror",
  "wayland-backend",
- "wayland-scanner 0.30.0-beta.4",
+ "wayland-scanner 0.30.0-beta.5",
 ]
 
 [[package]]
@@ -2419,8 +2422,8 @@ dependencies = [
 
 [[package]]
 name = "wayland-scanner"
-version = "0.30.0-beta.4"
-source = "git+https://github.com/smithay/wayland-rs.git#797f697a66f8541327cbaac319692ec8d2395a38"
+version = "0.30.0-beta.5"
+source = "git+https://github.com/smithay/wayland-rs.git#70b0389becc49edde19ec16779eec3c5b0ad0a5f"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2460,8 +2463,8 @@ dependencies = [
 
 [[package]]
 name = "wayland-sys"
-version = "0.30.0-beta.4"
-source = "git+https://github.com/smithay/wayland-rs.git#797f697a66f8541327cbaac319692ec8d2395a38"
+version = "0.30.0-beta.5"
+source = "git+https://github.com/smithay/wayland-rs.git#70b0389becc49edde19ec16779eec3c5b0ad0a5f"
 dependencies = [
  "dlib",
  "log",

applets/cosmic-applet-workspaces/Cargo.toml

@@ -21,6 +21,9 @@ wayland-commons = "0.29.4"
 wayland-scanner = { git = "https://github.com/smithay/wayland-rs.git", version = "0.30.0-beta.4"}
 wayland-backend = { version = "0.1.0-beta.4", git = "https://github.com/smithay/wayland-rs.git" }
 wayland-client = { version = "0.30.0-beta.4", git = "https://github.com/smithay/wayland-rs.git" }
+calloop = "*"
+nix = "*"
+log = "0.4"
 
 [build-dependencies]
 gio = "0.15.10"

applets/cosmic-applet-workspaces/src/main.rs

@@ -17,6 +17,7 @@ use window::CosmicWorkspacesWindow;
 mod localize;
 mod utils;
 mod wayland;
+mod wayland_source;
 mod window;
 mod workspace_button;
 mod workspace_list;

applets/cosmic-applet-workspaces/src/wayland.rs

@@ -1,4 +1,4 @@
-use crate::{utils::{Activate}, wayland::generated::client::zext_workspace_manager_v1::ZextWorkspaceManagerV1};
+use crate::{utils::{Activate}, wayland::generated::client::zext_workspace_manager_v1::ZextWorkspaceManagerV1, wayland_source::WaylandSource};
 use std::{env, os::unix::net::UnixStream, path::PathBuf, sync::Arc, mem, time::Duration};
 use gtk4::glib;
 use tokio::sync::mpsc;
@@ -53,9 +53,15 @@ pub fn spawn_workspaces(tx: glib::Sender<State>) -> mpsc::Sender<Activate> {
         .and_then(|s| s.map(|s| Connection::from_socket(s).map_err(anyhow::Error::msg)))
     {
         std::thread::spawn(move || {
-            let mut event_queue = conn.new_event_queue::<State>();
+            let mut event_loop = calloop::EventLoop::<State>::try_new().unwrap();
+            let loop_handle = event_loop.handle();
+            let event_queue = conn.new_event_queue::<State>();
             let qhandle = event_queue.handle();
 
+            WaylandSource::new(event_queue).expect("Failed to create wayland source")
+            .insert(loop_handle)
+            .unwrap();
+
             let display = conn.display();
             display.get_registry(&qhandle, ()).unwrap();
 
@@ -81,7 +87,7 @@ pub fn spawn_workspaces(tx: glib::Sender<State>) -> mpsc::Sender<Activate> {
                 if changed {
                     state.workspace_manager.as_ref().unwrap().commit();
                 }
-                event_queue.sync_roundtrip(&mut state).unwrap();
+                event_loop.dispatch(Duration::from_millis(16), &mut state).unwrap();
                 std::thread::sleep(Duration::from_millis(16));
             }
         });

applets/cosmic-applet-workspaces/src/wayland_source.rs

@@ -0,0 +1,215 @@
+//! Utilities for using an [`EventQueue`] from wayland-client with an event loop that performs polling with
+//! [`calloop`](https://crates.io/crates/calloop).
+
+use std::{io, os::unix::prelude::RawFd};
+
+use calloop::{
+    generic::Generic, EventSource, InsertError, Interest, LoopHandle, Mode, Poll, PostAction,
+    Readiness, RegistrationToken, Token, TokenFactory,
+};
+use nix::errno::Errno;
+use wayland_backend::client::{ReadEventsGuard, WaylandError};
+use wayland_client::{DispatchError, EventQueue};
+
+/// An adapter to insert an [`EventQueue`] into a calloop [`EventLoop`](calloop::EventLoop).
+///
+/// This type implements [`EventSource`] which generates an event whenever events on the display need to be
+/// dispatched. The event queue available in the callback calloop registers may be used to dispatch pending
+/// events using [`EventQueue::dispatch_pending`].
+///
+/// [`WaylandSource::insert`] can be used to insert this source into an event loop and automatically dispatch
+/// pending events on the display.
+#[derive(Debug)]
+pub struct WaylandSource<D> {
+    queue: EventQueue<D>,
+    fd: Generic<RawFd>,
+    read_guard: Option<ReadEventsGuard>,
+}
+
+impl<D> WaylandSource<D> {
+    /// Wrap an [`EventQueue`] as a [`WaylandSource`].
+    pub fn new(queue: EventQueue<D>) -> Result<WaylandSource<D>, WaylandError> {
+        let guard = queue.prepare_read()?;
+        let fd = Generic::new(guard.connection_fd(), Interest::READ, Mode::Level);
+        drop(guard);
+
+        Ok(WaylandSource { queue, fd, read_guard: None })
+    }
+
+    /// Access the underlying event queue
+    ///
+    /// Note that you should be careful when interacting with it if you invoke methods that
+    /// interact with the wayland socket (such as `dispatch()` or `prepare_read()`). These may
+    /// interfere with the proper waking up of this event source in the event loop.
+    pub fn queue(&mut self) -> &mut EventQueue<D> {
+        &mut self.queue
+    }
+
+    /// Insert this source into the given event loop.
+    ///
+    /// This adapter will pass the event loop's shared data as the `D` type for the event loop.
+    pub fn insert(self, handle: LoopHandle<D>) -> Result<RegistrationToken, InsertError<Self>>
+    where
+        D: 'static,
+    {
+        handle.insert_source(self, |_, queue, data| queue.dispatch_pending(data))
+    }
+}
+
+impl<D> EventSource for WaylandSource<D> {
+    type Event = ();
+
+    /// The underlying event queue.
+    ///
+    /// You should call [`EventQueue::dispatch_pending`] inside your callback using this queue.
+    type Metadata = EventQueue<D>;
+    type Ret = Result<usize, DispatchError>;
+    type Error = calloop::Error;
+
+    fn process_events<F>(
+        &mut self,
+        readiness: Readiness,
+        token: Token,
+        mut callback: F,
+    ) -> Result<PostAction, Self::Error>
+    where
+        F: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
+    {
+        let queue = &mut self.queue;
+        let read_guard = &mut self.read_guard;
+
+        let action = self.fd.process_events(readiness, token, |_, _| {
+            // 1. read events from the socket if any are available
+            if let Some(guard) = read_guard.take() {
+                // might be None if some other thread read events before us, concurrently
+                if let Err(WaylandError::Io(err)) = guard.read() {
+                    if err.kind() != io::ErrorKind::WouldBlock {
+                        return Err(err);
+                    }
+                }
+            }
+
+            // 2. dispatch any pending events in the queue
+            // This is done to ensure we are not waiting for messages that are already in the buffer.
+            Self::loop_callback_pending(queue, &mut callback)?;
+            *read_guard = Some(Self::prepare_read(queue)?);
+
+            // 3. Once dispatching is finished, flush the responses to the compositor
+            if let Err(WaylandError::Io(e)) = queue.flush() {
+                if e.kind() != io::ErrorKind::WouldBlock {
+                    // in case of error, forward it and fast-exit
+                    return Err(e);
+                }
+                // WouldBlock error means the compositor could not process all our messages
+                // quickly. Either it is slowed down or we are a spammer.
+                // Should not really happen, if it does we do nothing and will flush again later
+            }
+
+            Ok(PostAction::Continue)
+        })?;
+
+        Ok(action)
+    }
+
+    fn register(
+        &mut self,
+        poll: &mut Poll,
+        token_factory: &mut TokenFactory,
+    ) -> calloop::Result<()> {
+        self.fd.register(poll, token_factory)
+    }
+
+    fn reregister(
+        &mut self,
+        poll: &mut Poll,
+        token_factory: &mut TokenFactory,
+    ) -> calloop::Result<()> {
+        self.fd.reregister(poll, token_factory)
+    }
+
+    fn unregister(&mut self, poll: &mut Poll) -> calloop::Result<()> {
+        self.fd.unregister(poll)
+    }
+
+    fn pre_run<F>(&mut self, mut callback: F) -> calloop::Result<()>
+    where
+        F: FnMut((), &mut Self::Metadata) -> Self::Ret,
+    {
+        debug_assert!(self.read_guard.is_none());
+
+        // flush the display before starting to poll
+        if let Err(WaylandError::Io(err)) = self.queue.flush() {
+            if err.kind() != io::ErrorKind::WouldBlock {
+                // in case of error, don't prepare a read, if the error is persistent, it'll trigger in other
+                // wayland methods anyway
+                log::error!("Error trying to flush the wayland display: {}", err);
+                return Err(err.into());
+            }
+        }
+
+        // ensure we are not waiting for messages that are already in the buffer.
+        Self::loop_callback_pending(&mut self.queue, &mut callback)?;
+        self.read_guard = Some(Self::prepare_read(&mut self.queue)?);
+
+        Ok(())
+    }
+
+    fn post_run<F>(&mut self, _: F) -> calloop::Result<()>
+    where
+        F: FnMut((), &mut Self::Metadata) -> Self::Ret,
+    {
+        // Drop implementation of ReadEventsGuard will do cleanup
+        self.read_guard.take();
+        Ok(())
+    }
+}
+
+impl<D> WaylandSource<D> {
+    /// Loop over the callback until all pending messages have been dispatched.
+    fn loop_callback_pending<F>(queue: &mut EventQueue<D>, callback: &mut F) -> io::Result<()>
+    where
+        F: FnMut((), &mut EventQueue<D>) -> Result<usize, DispatchError>,
+    {
+        // Loop on the callback until no pending events are left.
+        loop {
+            match callback((), queue) {
+                // No more pending events.
+                Ok(0) => break Ok(()),
+
+                Ok(_) => continue,
+
+                Err(DispatchError::Backend(WaylandError::Io(err))) => {
+                    return Err(err);
+                }
+
+                Err(DispatchError::Backend(WaylandError::Protocol(err))) => {
+                    log::error!("Protocol error received on display: {}", err);
+
+                    break Err(Errno::EPROTO.into());
+                }
+
+                Err(DispatchError::BadMessage { msg, interface }) => {
+                    log::error!(
+                        "Bad message on interface \"{}\": (opcode: {}, args: {:?})",
+                        interface,
+                        msg.opcode,
+                        msg.args,
+                    );
+
+                    break Err(Errno::EPROTO.into());
+                }
+            }
+        }
+    }
+
+    fn prepare_read(queue: &mut EventQueue<D>) -> io::Result<ReadEventsGuard> {
+        queue.prepare_read().map_err(|err| match err {
+            WaylandError::Io(err) => err,
+
+            WaylandError::Protocol(err) => {
+                log::error!("Protocol error received on display: {}", err);
+                Errno::EPROTO.into()
+            }
+        })
+    }
+}

applets/cosmic-applet-workspaces/src/window/mod.rs

@@ -35,7 +35,7 @@ impl CosmicWorkspacesWindow {
             ..set_title(Some(&fl!("cosmic-applet-workspaces")));
             ..add_css_class("transparent");
         };
-        let config = CosmicPanelConfig::load_from_env().unwrap();
+        let config = CosmicPanelConfig::load_from_env().unwrap_or_default();
 
         let app_list = WorkspaceList::new(config);
         self_.set_child(Some(&app_list));