improv(minimize): simplify wayland subscription

Removes the redundant unbounded channel and forwards the subscription's
sender directly to the wayland handler.

cosmic-applet-minimize/src/lib.rs

@@ -147,7 +147,10 @@ impl cosmic::Application for Minimize {
                             self.apps.push((handle, info, data, None));
                         }
                     }
-                    ToplevelUpdate::Remove(handle) => self.apps.retain(|a| a.0 != handle),
+                    ToplevelUpdate::Remove(handle) => {
+                        self.apps.retain(|a| a.0 != handle);
+                        self.apps.shrink_to_fit();
+                    }
                 },
                 WaylandUpdate::Image(handle, img) => {
                     if let Some(pos) = self.apps.iter().position(|a| a.0 == handle) {

cosmic-applet-minimize/src/wayland_handler.rs

@@ -54,7 +54,7 @@ use cosmic_protocols::{
     toplevel_info::v1::client::zcosmic_toplevel_handle_v1,
     toplevel_management::v1::client::zcosmic_toplevel_manager_v1,
 };
-use futures::channel::mpsc::UnboundedSender;
+use futures::{channel::mpsc, SinkExt};
 use sctk::registry::{ProvidesRegistryState, RegistryState};
 use wayland_client::{globals::registry_queue_init, Connection, QueueHandle};
 
@@ -115,7 +115,7 @@ impl ScreencopyFrameDataExt for FrameData {
 
 struct AppData {
     exit: bool,
-    tx: UnboundedSender<WaylandUpdate>,
+    tx: mpsc::Sender<WaylandUpdate>,
     queue_handle: QueueHandle<Self>,
     conn: Connection,
     screencopy_state: ScreencopyState,
@@ -301,7 +301,7 @@ impl ToplevelManagerHandler for AppData {
 }
 impl AppData {
     fn send_image(&self, handle: ZcosmicToplevelHandleV1) {
-        let tx = self.tx.clone();
+        let mut tx = self.tx.clone();
         let capure_data = CaptureData {
             qh: self.queue_handle.clone(),
             conn: self.conn.clone(),
@@ -325,7 +325,7 @@ impl AppData {
             // XXX is this going to use to much memory?
             let img = capure_data.capture_source_shm_fd(false, handle.clone(), fd, None);
             if let Some(img) = img {
-                let Ok(img) = img.image() else {
+                let Ok(mut img) = img.image() else {
                     tracing::error!("Failed to get RgbaImage");
                     return;
                 };
@@ -334,23 +334,21 @@ impl AppData {
                 let max = img.width().max(img.height());
                 let ratio = max as f32 / 128.0;
 
-                let img = if ratio > 1.0 {
+                if ratio > 1.0 {
                     let new_width = (img.width() as f32 / ratio).round();
                     let new_height = (img.height() as f32 / ratio).round();
 
-                    image::imageops::resize(
+                    img = image::imageops::resize(
                         &img,
                         new_width as u32,
                         new_height as u32,
-                        image::imageops::FilterType::Lanczos3,
+                        image::imageops::FilterType::Nearest,
-                    )
+                    );
-                } else {
+                }
-                    img
-                };
 
-                if let Err(err) =
+                if let Err(err) = futures::executor::block_on(
-                    tx.unbounded_send(WaylandUpdate::Image(handle, WaylandImage::new(img)))
+                    tx.send(WaylandUpdate::Image(handle, WaylandImage::new(img))),
-                {
+                ) {
                     tracing::error!("Failed to send image event to subscription {err:?}");
                 };
             } else {
@@ -378,18 +376,13 @@ impl ToplevelInfoHandler for AppData {
             {
                 // spawn thread for sending the image
                 self.send_image(toplevel.clone());
-                let _ = self
+                let _ = futures::executor::block_on(self.tx.send(WaylandUpdate::Toplevel(
-                    .tx
+                    ToplevelUpdate::Add(toplevel.clone(), info.clone()),
-                    .unbounded_send(WaylandUpdate::Toplevel(ToplevelUpdate::Add(
+                )));
-                        toplevel.clone(),
-                        info.clone(),
-                    )));
             } else {
-                let _ = self
+                let _ = futures::executor::block_on(self.tx.send(WaylandUpdate::Toplevel(
-                    .tx
+                    ToplevelUpdate::Remove(toplevel.clone()),
-                    .unbounded_send(WaylandUpdate::Toplevel(ToplevelUpdate::Remove(
+                )));
-                        toplevel.clone(),
-                    )));
             }
         }
     }
@@ -406,18 +399,13 @@ impl ToplevelInfoHandler for AppData {
                 .contains(&zcosmic_toplevel_handle_v1::State::Minimized)
             {
                 self.send_image(toplevel.clone());
-                let _ = self
+                let _ = futures::executor::block_on(self.tx.send(WaylandUpdate::Toplevel(
-                    .tx
+                    ToplevelUpdate::Update(toplevel.clone(), info.clone()),
-                    .unbounded_send(WaylandUpdate::Toplevel(ToplevelUpdate::Update(
+                )));
-                        toplevel.clone(),
-                        info.clone(),
-                    )));
             } else {
-                let _ = self
+                let _ = futures::executor::block_on(self.tx.send(WaylandUpdate::Toplevel(
-                    .tx
+                    ToplevelUpdate::Remove(toplevel.clone()),
-                    .unbounded_send(WaylandUpdate::Toplevel(ToplevelUpdate::Remove(
+                )));
-                        toplevel.clone(),
-                    )));
             }
         }
     }
@@ -428,16 +416,14 @@ impl ToplevelInfoHandler for AppData {
         _qh: &QueueHandle<Self>,
         toplevel: &zcosmic_toplevel_handle_v1::ZcosmicToplevelHandleV1,
     ) {
-        let _ = self
+        let _ = futures::executor::block_on(self.tx.send(WaylandUpdate::Toplevel(
-            .tx
+            ToplevelUpdate::Remove(toplevel.clone()),
-            .unbounded_send(WaylandUpdate::Toplevel(ToplevelUpdate::Remove(
+        )));
-                toplevel.clone(),
-            )));
     }
 }
 
 pub(crate) fn wayland_handler(
-    tx: UnboundedSender<WaylandUpdate>,
+    tx: mpsc::Sender<WaylandUpdate>,
     rx: calloop::channel::Channel<WaylandRequest>,
 ) {
     let socket = std::env::var("X_PRIVILEGED_WAYLAND_SOCKET")

cosmic-applet-minimize/src/wayland_subscription.rs

@@ -13,73 +13,33 @@ use cosmic::{
     iced_futures::{futures, stream},
 };
 use cosmic_protocols::toplevel_info::v1::client::zcosmic_toplevel_handle_v1::ZcosmicToplevelHandleV1;
-use futures::{
+use futures::SinkExt;
-    channel::mpsc::{unbounded, UnboundedReceiver},
-    SinkExt, StreamExt,
-};
 use image::EncodableLayout;
-use once_cell::sync::Lazy;
 use std::fmt::Debug;
-use tokio::sync::Mutex;
 
 use crate::wayland_handler::wayland_handler;
 
-pub static WAYLAND_RX: Lazy<Mutex<Option<UnboundedReceiver<WaylandUpdate>>>> =
-    Lazy::new(|| Mutex::new(None));
-
 pub fn wayland_subscription() -> iced::Subscription<WaylandUpdate> {
     Subscription::run_with_id(
         std::any::TypeId::of::<WaylandUpdate>(),
-        stream::channel(50, move |mut output| async move {
+        stream::channel(1, move |mut output| async move {
-            let mut state = State::Waiting;
+            let (calloop_tx, calloop_rx) = calloop::channel::channel();
+            let runtime = tokio::runtime::Handle::current();
 
-            loop {
+            let _ = std::thread::spawn(move || {
-                state = start_listening(state, &mut output).await;
+                runtime.block_on(async move {
-            }
+                    _ = output.send(WaylandUpdate::Init(calloop_tx)).await;
+                    wayland_handler(output.clone(), calloop_rx);
+                    tracing::error!("Wayland handler thread died");
+                    _ = output.send(WaylandUpdate::Finished).await;
+                });
+            });
+
+            futures::future::pending().await
         }),
     )
 }
 
-pub enum State {
-    Waiting,
-    Finished,
-}
-
-async fn start_listening(
-    state: State,
-    output: &mut futures::channel::mpsc::Sender<WaylandUpdate>,
-) -> State {
-    match state {
-        State::Waiting => {
-            let mut guard = WAYLAND_RX.lock().await;
-            let rx = {
-                if guard.is_none() {
-                    let (calloop_tx, calloop_rx) = calloop::channel::channel();
-                    let (toplevel_tx, toplevel_rx) = unbounded();
-                    let _ = std::thread::spawn(move || {
-                        wayland_handler(toplevel_tx, calloop_rx);
-                    });
-                    *guard = Some(toplevel_rx);
-                    _ = output.send(WaylandUpdate::Init(calloop_tx)).await;
-                }
-                guard.as_mut().unwrap()
-            };
-            match rx.next().await {
-                Some(u) => {
-                    _ = output.send(u).await;
-                    State::Waiting
-                }
-                None => {
-                    _ = output.send(WaylandUpdate::Finished).await;
-                    tracing::error!("Wayland handler thread died");
-                    State::Finished
-                }
-            }
-        }
-        State::Finished => iced::futures::future::pending().await,
-    }
-}
-
 #[derive(Clone, Debug)]
 pub enum WaylandUpdate {
     Init(calloop::channel::Sender<WaylandRequest>),