fix: make all subscriptions resistant to being restarted

many of the errors we've been seeing the last few days are because of subscriptions which are restarting

Cargo.toml

@@ -44,10 +44,10 @@ tracing-log = "0.2.0"
 lto = "thin"
 # lto = "fat"
 
-[patch."https://github.com/pop-os/cosmic-time"]
-cosmic-time = { path = "../cosmic-time" }
-[patch."https://github.com/pop-os/libcosmic"]
-libcosmic = { path = "../libcosmic" }
-cosmic-config = { path = "../libcosmic/cosmic-config" }
+# [patch."https://github.com/pop-os/cosmic-time"]
+# cosmic-time = { path = "../cosmic-time" }
+# [patch."https://github.com/pop-os/libcosmic"]
+# libcosmic = { git = "https://github.com/pop-os/libcosmic//", branch = "refactor-config-watch" }
+# cosmic-config = { git = "https://github.com/pop-os/libcosmic//", branch = "refactor-config-watch" }
 [patch."https://github.com/Smithay/client-toolkit"]
 sctk = { git = "https://github.com/smithay/client-toolkit//", package = "smithay-client-toolkit", rev = "e63ab5f" }

cosmic-app-list/Cargo.toml

@@ -16,6 +16,7 @@ once_cell = "1.9"
 xdg = "2.4"
 tracing-subscriber.workspace = true
 tracing-log.workspace = true
+tracing.workspace = true
 nix = "0.26"
 shlex = "1.1.0"
 anyhow = "1.0"

cosmic-app-list/src/app.rs

@@ -1122,7 +1122,7 @@ impl cosmic::Application for CosmicAppList {
 
     fn subscription(&self) -> Subscription<Message> {
         Subscription::batch(vec![
-            wayland_subscription(self.subscription_ctr).map(Message::Wayland),
+            wayland_subscription().map(Message::Wayland),
             listen_with(|e, _| match e {
                 cosmic::iced_runtime::core::Event::PlatformSpecific(
                     event::PlatformSpecific::Wayland(event::wayland::Event::Seat(e, seat)),

cosmic-app-list/src/wayland_subscription.rs

@@ -11,29 +11,31 @@ use futures::{
     channel::mpsc::{unbounded, UnboundedReceiver},
     SinkExt, StreamExt,
 };
+use once_cell::sync::Lazy;
 use std::{fmt::Debug, hash::Hash, thread::JoinHandle};
+use tokio::sync::Mutex;
 
 use crate::wayland_handler::wayland_handler;
 
-pub fn wayland_subscription<I: 'static + Hash + Copy + Send + Sync + Debug>(
-    id: I,
-) -> iced::Subscription<WaylandUpdate> {
-    subscription::channel(id, 50, move |mut output| async move {
-        let mut state = State::Ready;
+pub static WAYLAND_RX: Lazy<Mutex<Option<UnboundedReceiver<WaylandUpdate>>>> =
+    Lazy::new(|| Mutex::new(None));
 
-        loop {
-            state = start_listening(state, &mut output).await;
-        }
-    })
+pub fn wayland_subscription() -> iced::Subscription<WaylandUpdate> {
+    subscription::channel(
+        std::any::TypeId::of::<WaylandUpdate>(),
+        50,
+        move |mut output| async move {
+            let mut state = State::Waiting;
+
+            loop {
+                state = start_listening(state, &mut output).await;
+            }
+        },
+    )
 }
 
 pub enum State {
-    Ready,
-    Waiting(
-        UnboundedReceiver<WaylandUpdate>,
-        calloop::channel::Sender<WaylandRequest>,
-        JoinHandle<()>,
-    ),
+    Waiting,
     Finished,
 }
 
@@ -42,28 +44,28 @@ async fn start_listening(
     output: &mut futures::channel::mpsc::Sender<WaylandUpdate>,
 ) -> State {
     match state {
-        State::Ready => {
-            let (calloop_tx, calloop_rx) = calloop::channel::channel();
-            let (toplevel_tx, toplevel_rx) = unbounded();
-            let handle = std::thread::spawn(move || {
-                wayland_handler(toplevel_tx, calloop_rx);
-            });
-            let tx = calloop_tx.clone();
-            _ = output.send(WaylandUpdate::Init(tx)).await;
-            State::Waiting(toplevel_rx, calloop_tx, handle)
-        }
-        State::Waiting(mut rx, tx, handle) => {
-            if handle.is_finished() {
-                _ = output.send(WaylandUpdate::Finished).await;
-                return State::Finished;
-            }
+        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(rx, tx, handle)
+                    State::Waiting
                 }
                 None => {
                     _ = output.send(WaylandUpdate::Finished).await;
+                    tracing::error!("Wayland handler thread died");
                     State::Finished
                 }
             }

cosmic-applet-audio/src/main.rs

@@ -43,8 +43,7 @@ mod pulse;
 
 const VERSION: &str = env!("CARGO_PKG_VERSION");
 
-#[tokio::main(flavor = "current_thread")]
-pub async fn main() -> cosmic::iced::Result {
+pub fn main() -> cosmic::iced::Result {
     tracing_subscriber::fmt::init();
     let _ = tracing_log::LogTracer::init();