refactor: use Task::stream instead of a channel subscription

cosmic-settings/src/app.rs

@@ -25,8 +25,7 @@ use cosmic::app::context_drawer::ContextDrawer;
 use cosmic::app::DbusActivationMessage;
 #[cfg(feature = "wayland")]
 use cosmic::cctk::{sctk::output::OutputInfo, wayland_client::protocol::wl_output::WlOutput};
-use cosmic::iced::futures::SinkExt;
+use cosmic::iced::Subscription;
-use cosmic::iced::{stream, Subscription};
 use cosmic::widget::{self, button, row, text_input};
 use cosmic::{
     app::{Core, Task},
@@ -65,7 +64,6 @@ pub struct SettingsApp {
     config: Config,
     core: Core,
     nav_model: nav_bar::Model,
-    page_sender: Option<tokio::sync::mpsc::Sender<crate::pages::Message>>,
     pages: page::Binder<crate::pages::Message>,
     search_active: bool,
     search_id: cosmic::widget::Id,
@@ -144,7 +142,6 @@ impl SettingsApp {
 #[derive(Clone, Debug)]
 pub enum Message {
     CloseContextDrawer,
-    DelayedInit(page::Entity),
     #[cfg(feature = "wayland")]
     DesktopInfo,
     Error(String),
@@ -158,7 +155,6 @@ pub enum Message {
     PageMessage(crate::pages::Message),
     #[cfg(feature = "wayland")]
     PanelConfig(CosmicPanelConfig),
-    RegisterSubscriptionSender(tokio::sync::mpsc::Sender<pages::Message>),
     SearchActivate,
     SearchChanged(String),
     SearchClear,
@@ -193,7 +189,6 @@ impl cosmic::Application for SettingsApp {
             config,
             core,
             nav_model: nav_bar::Model::default(),
-            page_sender: None,
             pages: page::Binder::default(),
             search_active: false,
             search_id: cosmic::widget::Id::unique(),
@@ -228,7 +223,8 @@ impl cosmic::Application for SettingsApp {
         }
         .unwrap_or(desktop_id);
 
-        (app, cosmic::task::message(Message::DelayedInit(active_id)))
+        let task = app.activate_page(active_id);
+        (app, task)
     }
 
     fn nav_model(&self) -> Option<&nav_bar::Model> {
@@ -286,22 +282,6 @@ impl cosmic::Application for SettingsApp {
 
     fn subscription(&self) -> Subscription<Message> {
         Subscription::batch(vec![
-            // Creates a channel that listens to messages from pages.
-            // The sender is given back to the application so that it may pass it on.
-            Subscription::run_with_id(
-                std::any::TypeId::of::<Self>(),
-                stream::channel(4, move |mut output| async move {
-                    let (tx, mut rx) = tokio::sync::mpsc::channel::<pages::Message>(4);
-
-                    let _res = output.send(Message::RegisterSubscriptionSender(tx)).await;
-
-                    while let Some(event) = rx.recv().await {
-                        let _res = output.send(Message::PageMessage(event)).await;
-                    }
-
-                    futures::future::pending::<()>().await;
-                }),
-            ),
             #[cfg(feature = "ashpd")]
             crate::subscription::daytime().map(|daytime| {
                 Message::PageMessage(pages::Message::Appearance(appearance::Message::Daytime(
@@ -754,19 +734,6 @@ impl cosmic::Application for SettingsApp {
             Message::Error(error) => {
                 tracing::error!(error, "error occurred");
             }
-
-            Message::RegisterSubscriptionSender(sender) => {
-                self.page_sender = Some(sender);
-            }
-
-            // It is necessary to delay init to allow time for the page sender to be initialized
-            Message::DelayedInit(active_id) => {
-                if self.page_sender.is_none() {
-                    return cosmic::task::message(Message::DelayedInit(active_id));
-                }
-
-                return self.activate_page(active_id);
-            }
         }
 
         Task::none()
@@ -860,17 +827,11 @@ impl SettingsApp {
         self.search_clear();
         self.search_active = false;
         self.activate_navbar(page);
-
-        let sender = self
-            .page_sender
-            .clone()
-            .expect("sender should be available");
-
         self.loaded_pages.insert(page);
 
         let page_task = self
             .pages
-            .on_enter(page, sender)
+            .on_enter(page)
             .map(Message::PageMessage)
             .map(Into::into);
 
@@ -1032,11 +993,9 @@ impl SettingsApp {
                     }
                 }
 
-                if let Some(ref sender) = self.page_sender {
+                for page in load {
-                    for page in load {
+                    self.loaded_pages.insert(page);
-                        self.loaded_pages.insert(page);
+                    tasks.push(self.pages.on_enter(page));
-                        tasks.push(self.pages.on_enter(page, sender.clone()));
-                    }
                 }
 
                 for page in unload {

cosmic-settings/src/pages/accessibility/magnifier.rs

@@ -14,7 +14,6 @@ use cosmic_settings_page::{
     Entity,
 };
 use slotmap::SlotMap;
-use tokio::sync::mpsc;
 use tracing::error;
 
 use super::{wayland, AccessibilityEvent, AccessibilityRequest};
@@ -114,28 +113,29 @@ impl page::Page<crate::pages::Message> for Page {
         ])
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        sender: mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         if self.wayland_thread.is_none() {
             match wayland::spawn_wayland_connection() {
                 Ok((tx, mut rx)) => {
                     self.wayland_thread = Some(tx);
-                    tokio::task::spawn(async move {
+
-                        while let Some(event) = rx.recv().await {
+                    return cosmic::Task::stream(async_fn_stream::fn_stream(
-                            let _ = sender
+                        |emitter| async move {
-                                .send(crate::pages::Message::AccessibilityMagnifier(
+                            while let Some(event) = rx.recv().await {
-                                    Message::Event(event),
+                                let _ = emitter
+                                    .emit(crate::pages::Message::AccessibilityMagnifier(
+                                        Message::Event(event),
+                                    ))
+                                    .await;
+                            }
+
+                            let _ = emitter
+                                .emit(crate::pages::Message::AccessibilityMagnifier(
+                                    Message::ProtocolUnavailable,
                                 ))
                                 .await;
-                        }
+                        },
-                        let _ = sender
+                    ));
-                            .send(crate::pages::Message::AccessibilityMagnifier(
-                                Message::ProtocolUnavailable,
-                            ))
-                            .await;
-                    });
                 }
                 Err(err) => {
                     tracing::warn!(

cosmic-settings/src/pages/accessibility/mod.rs

@@ -11,7 +11,6 @@ use cosmic_settings_page::{
     Insert,
 };
 use slotmap::SlotMap;
-use tokio::sync::mpsc;
 
 pub mod magnifier;
 mod wayland;
@@ -53,27 +52,30 @@ impl page::Page<crate::pages::Message> for Page {
         Some(vec![sections.insert(vision())])
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        sender: mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         if self.wayland_thread.is_none() {
             match wayland::spawn_wayland_connection() {
                 Ok((tx, mut rx)) => {
                     self.wayland_available = true;
                     self.wayland_thread = Some(tx);
-                    tokio::task::spawn(async move {
+
-                        while let Some(event) = rx.recv().await {
+                    return cosmic::Task::stream(async_fn_stream::fn_stream(
-                            let _ = sender
+                        |emitter| async move {
-                                .send(crate::pages::Message::Accessibility(Message::Event(event)))
+                            while let Some(event) = rx.recv().await {
+                                let _ = emitter
+                                    .emit(crate::pages::Message::Accessibility(Message::Event(
+                                        event,
+                                    )))
+                                    .await;
+                            }
+
+                            let _ = emitter
+                                .emit(crate::pages::Message::Accessibility(
+                                    Message::ProtocolUnavailable,
+                                ))
                                 .await;
-                        }
+                        },
-                        let _ = sender
+                    ));
-                            .send(crate::pages::Message::Accessibility(
-                                Message::ProtocolUnavailable,
-                            ))
-                            .await;
-                    });
                 }
                 Err(err) => {
                     tracing::warn!(

cosmic-settings/src/pages/bluetooth/mod.rs

@@ -8,6 +8,7 @@ use cosmic::{theme, Apply, Element, Task};
 use cosmic_settings_page::{self as page, section, Section};
 use cosmic_settings_subscriptions::bluetooth::*;
 use futures::channel::oneshot;
+use futures::StreamExt;
 use slab::Slab;
 use slotmap::SlotMap;
 use std::collections::{HashMap, HashSet};
@@ -69,14 +70,11 @@ impl page::Page<crate::pages::Message> for Page {
         ])
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         // TODO start stream for new device
         cosmic::task::future(async move {
             match zbus::Connection::system().await {
-                Ok(connection) => Message::DBusConnect(connection, sender),
+                Ok(connection) => Message::DBusConnect(connection),
                 Err(why) => Message::DBusError(why.to_string()),
             }
         })
@@ -148,10 +146,7 @@ impl page::Page<crate::pages::Message> for Page {
 pub enum Message {
     BluetoothEvent(Event),
     ConnectDevice(OwnedObjectPath),
-    DBusConnect(
+    DBusConnect(zbus::Connection),
-        zbus::Connection,
-        tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ),
     DBusError(String),
     DisconnectDevice(OwnedObjectPath),
     ForgetDevice(OwnedObjectPath),
@@ -378,26 +373,47 @@ impl Page {
                 tracing::warn!("No DBus connection ready");
             }
 
-            Message::DBusConnect(connection, sender) => {
+            Message::DBusConnect(connection) => {
                 self.connection = Some(connection.clone());
 
+                let get_adapters_fut = get_adapters(connection.clone());
+
                 if self.subscription.is_none() {
                     let connection = connection.clone();
-                    self.subscription = Some(crate::utils::forward_event_loop(
-                        sender,
-                        |response| {
-                            crate::pages::Message::Bluetooth(Message::BluetoothEvent(response))
-                        },
-                        move |tx| async move {
-                            _ = futures::join!(
-                                subscription::watch(connection.clone(), tx.clone()),
-                                agent::watch(connection, tx),
-                            );
-                        },
-                    ));
-                }
 
-                return cosmic::task::future(get_adapters(connection.clone()));
+                    let (cancellation, task) = crate::utils::forward_event_loop(move |emitter| {
+                        let connection = connection.clone();
+
+                        async move {
+                            let (tx, mut rx) = futures::channel::mpsc::channel(1);
+
+                            let watchers = std::pin::pin!(async move {
+                                _ = futures::future::join(
+                                    subscription::watch(connection.clone(), tx.clone()),
+                                    agent::watch(connection, tx),
+                                )
+                                .await;
+                            });
+
+                            let forwarder = std::pin::pin!(async move {
+                                while let Some(message) = rx.next().await {
+                                    _ = emitter
+                                        .emit(crate::pages::Message::Bluetooth(
+                                            Message::BluetoothEvent(message),
+                                        ))
+                                        .await;
+                                }
+                            });
+
+                            futures::future::select(watchers, forwarder).await;
+                        }
+                    });
+
+                    self.subscription = Some(cancellation);
+                    return cosmic::task::batch(vec![cosmic::task::future(get_adapters_fut), task]);
+                } else {
+                    return cosmic::task::future(get_adapters_fut);
+                }
             }
             Message::PopupDevice(popup) => {
                 self.popup_device = popup;

cosmic-settings/src/pages/desktop/appearance/mod.rs

@@ -1562,10 +1562,7 @@ impl page::Page<crate::pages::Message> for Page {
             .description(fl!("appearance", "desc"))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         let (task, handle) = cosmic::task::batch(vec![
             // Load icon themes
             // cosmic::task::future(icon_themes::fetch()).map(crate::pages::Message::Appearance),

cosmic-settings/src/pages/desktop/panel/applets_inner.rs

@@ -143,10 +143,7 @@ impl page::Page<crate::pages::Message> for Page {
         })
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         Task::none()
     }
 }

cosmic-settings/src/pages/desktop/wallpaper/mod.rs

@@ -210,10 +210,7 @@ impl page::Page<crate::pages::Message> for Page {
             .description(fl!("wallpaper", "desc"))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         // Check if the page is already being loaded.
         if self.on_enter_handle.is_some() {
             return Task::none();

cosmic-settings/src/pages/display/mod.rs

@@ -17,7 +17,6 @@ use cosmic_randr_shell::{
     AdaptiveSyncAvailability, AdaptiveSyncState, List, Output, OutputKey, Transform,
 };
 use cosmic_settings_page::{self as page, section, Section};
-use futures::pin_mut;
 use once_cell::sync::Lazy;
 use slab::Slab;
 use slotmap::{Key, SecondaryMap, SlotMap};
@@ -252,10 +251,7 @@ impl page::Page<crate::pages::Message> for Page {
     }
 
     #[cfg(not(feature = "test"))]
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         use std::time::Duration;
 
         self.cache.orientations = [
@@ -290,7 +286,7 @@ impl page::Page<crate::pages::Message> for Page {
             // Spawns a background service to monitor for display state changes.
             // This must be spawned onto its own thread because `*mut wayland_sys::client::wl_display` is not Send-able.
             tokio::task::spawn_blocking(move || {
-                let dispatcher = async move {
+                let dispatcher = std::pin::pin!(async move {
                     let Ok((mut context, mut event_queue)) = cosmic_randr::connect(tx) else {
                         return;
                     };
@@ -300,9 +296,8 @@ impl page::Page<crate::pages::Message> for Page {
                             return;
                         }
                     }
-                };
+                });
 
-                pin_mut!(dispatcher);
                 runtime.block_on(futures::future::select(cancelled, dispatcher));
             });
 
@@ -365,7 +360,7 @@ impl page::Page<crate::pages::Message> for Page {
                     return;
                 };
 
-                let emitter = async move {
+                let emitter = std::pin::pin!(async move {
                     loop {
                         // If any DRM events occur, a message updating the display list will be sent.
                         if let Ok(mut guard) = async_fd.writable().await {
@@ -383,14 +378,11 @@ impl page::Page<crate::pages::Message> for Page {
 
                         tokio::time::sleep(Duration::from_secs(3)).await;
                     }
-                };
+                });
 
-                let cancellation = async move {
+                let cancellation = std::pin::pin!(async move {
                     _ = hotplug_cancel_rx.await;
-                };
+                });
-
-                futures::pin_mut!(emitter);
-                futures::pin_mut!(cancellation);
 
                 futures::future::select(cancellation, emitter).await;
             });
@@ -426,10 +418,7 @@ impl page::Page<crate::pages::Message> for Page {
     }
 
     #[cfg(feature = "test")]
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         cosmic::task::future(async move {
             let mut randr = List::default();
 

cosmic-settings/src/pages/input/keyboard/mod.rs

@@ -300,10 +300,7 @@ impl page::Page<crate::pages::Message> for Page {
         }
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         self.xkb = super::get_config(&self.config, "xkb_config");
         match (
             xkb_data::keyboard_layouts(),

cosmic-settings/src/pages/input/keyboard/shortcuts/custom.rs

@@ -358,10 +358,7 @@ impl page::Page<crate::pages::Message> for Page {
         .map(|el| el.map(crate::pages::Message::CustomShortcuts))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         self.model.on_enter();
         Task::none()
     }

cosmic-settings/src/pages/input/keyboard/shortcuts/manage_windows.rs

@@ -59,10 +59,7 @@ impl page::Page<crate::pages::Message> for Page {
             .map(|el| el.map(crate::pages::Message::ManageWindowShortcuts))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         self.model.on_enter();
 
         Task::none()

cosmic-settings/src/pages/input/keyboard/shortcuts/mod.rs

@@ -142,10 +142,7 @@ impl page::Page<crate::pages::Message> for Page {
         }
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         if self.shortcuts_context.is_none() {
             self.shortcuts_context = cosmic_settings_config::shortcuts::context().ok();
         }

cosmic-settings/src/pages/input/keyboard/shortcuts/move_window.rs

@@ -59,10 +59,7 @@ impl page::Page<crate::pages::Message> for Page {
             .map(|el| el.map(crate::pages::Message::MoveWindowShortcuts))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         self.model.on_enter();
 
         Task::none()

cosmic-settings/src/pages/input/keyboard/shortcuts/nav.rs

@@ -59,10 +59,7 @@ impl page::Page<crate::pages::Message> for Page {
             .map(|el| el.map(crate::pages::Message::NavShortcuts))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         self.model.on_enter();
 
         Task::none()

cosmic-settings/src/pages/input/keyboard/shortcuts/system.rs

@@ -59,10 +59,7 @@ impl page::Page<crate::pages::Message> for Page {
             .map(|el| el.map(crate::pages::Message::SystemShortcuts))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         self.model.on_enter();
 
         Task::none()

cosmic-settings/src/pages/input/keyboard/shortcuts/tiling.rs

@@ -59,10 +59,7 @@ impl page::Page<crate::pages::Message> for Page {
             .map(|el| el.map(crate::pages::Message::TilingShortcuts))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         self.model.on_enter();
 
         Task::none()

cosmic-settings/src/pages/networking/mod.rs

@@ -15,7 +15,7 @@ use cosmic_dbus_networkmanager::{
 };
 use cosmic_settings_page::{self as page, section, Section};
 use cosmic_settings_subscriptions::network_manager;
-use futures::{SinkExt, StreamExt};
+use futures::StreamExt;
 use slotmap::SlotMap;
 
 static NM_CONNECTION_EDITOR: &str = "nm-connection-editor";
@@ -35,12 +35,7 @@ pub enum Message {
     /// An error occurred.
     Error(String),
     /// Successfully connected to the system dbus.
-    NetworkManagerConnect(
+    NetworkManagerConnect(zbus::Connection),
-        (
-            zbus::Connection,
-            tokio::sync::mpsc::Sender<crate::pages::Message>,
-        ),
-    ),
     /// Open the wifi settings page with the selected device.
     OpenPage {
         page: page::Entity,
@@ -224,10 +219,7 @@ impl page::Page<crate::pages::Message> for Page {
         Some(vec![sections.insert(device_list)])
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         if self.nm_task.is_none() {
             return cosmic::Task::future(async move {
                 zbus::Connection::system()
@@ -235,7 +227,7 @@ impl page::Page<crate::pages::Message> for Page {
                     .context("failed to create system dbus connection")
                     .map_or_else(
                         |why| Message::Error(why.to_string()),
-                        |conn| Message::NetworkManagerConnect((conn, sender.clone())),
+                        |conn| Message::NetworkManagerConnect(conn),
                     )
                     .apply(crate::pages::Message::Networking)
             });
@@ -278,8 +270,8 @@ impl Page {
         let _span = span.enter();
 
         match message {
-            Message::NetworkManagerConnect((conn, output)) => {
+            Message::NetworkManagerConnect(conn) => {
-                self.connect(conn.clone(), output);
+                return self.connect(conn.clone());
             }
 
             Message::Error(why) => {
@@ -315,46 +307,45 @@ impl Page {
         Task::none()
     }
 
-    fn connect(
+    fn connect(&mut self, conn: zbus::Connection) -> Task<crate::app::Message> {
-        &mut self,
-        conn: zbus::Connection,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) {
         if self.nm_task.is_none() {
-            self.nm_task = Some(crate::utils::forward_event_loop(
+            let (canceller, task) = crate::utils::forward_event_loop(move |emitter| async move {
-                sender,
+                let network_manager = match NetworkManager::new(&conn).await {
-                crate::pages::Message::Networking,
+                    Ok(n) => n,
-                move |mut tx| async move {
+                    Err(why) => {
-                    let network_manager = match NetworkManager::new(&conn).await {
+                        tracing::error!(
-                        Ok(n) => n,
+                            why = why.to_string(),
-                        Err(why) => {
+                            "failed to connect to network_manager"
-                            tracing::error!(
+                        );
-                                why = why.to_string(),
-                                "failed to connect to network_manager"
-                            );
 
-                            return futures::future::pending().await;
+                        return futures::future::pending().await;
-                        }
-                    };
-
-                    let mut devices_changed = std::pin::pin!(network_manager
-                        .receive_devices_changed()
-                        .await
-                        .then(|_| async {
-                            match network_manager::devices::list(&conn, |_| true).await {
-                                Ok(devices) => Message::UpdateDevices(
-                                    devices.into_iter().map(Arc::new).collect(),
-                                ),
-                                Err(why) => Message::Error(why.to_string()),
-                            }
-                        }));
-
-                    while let Some(message) = devices_changed.next().await {
-                        _ = tx.send(message).await;
                     }
-                },
+                };
-            ));
+
+                let mut devices_changed = std::pin::pin!(network_manager
+                    .receive_devices_changed()
+                    .await
+                    .then(|_| async {
+                        match network_manager::devices::list(&conn, |_| true).await {
+                            Ok(devices) => {
+                                Message::UpdateDevices(devices.into_iter().map(Arc::new).collect())
+                            }
+                            Err(why) => Message::Error(why.to_string()),
+                        }
+                    }));
+
+                while let Some(message) = devices_changed.next().await {
+                    _ = emitter
+                        .emit(crate::pages::Message::Networking(message))
+                        .await;
+                }
+            });
+
+            self.nm_task = Some(canceller);
+            return task.map(crate::app::Message::from);
         }
+
+        Task::none()
     }
 }
 

cosmic-settings/src/pages/networking/vpn/mod.rs

@@ -45,12 +45,7 @@ pub enum Message {
     /// An update from the network manager daemon
     NetworkManager(network_manager::Event),
     /// Successfully connected to the system dbus.
-    NetworkManagerConnect(
+    NetworkManagerConnect(zbus::Connection),
-        (
-            zbus::Connection,
-            tokio::sync::mpsc::Sender<crate::pages::Message>,
-        ),
-    ),
     /// Updates the password text input
     PasswordUpdate(SecureString),
     /// Refresh devices and their connection profiles
@@ -334,10 +329,7 @@ impl page::Page<crate::pages::Message> for Page {
         )
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         if self.nm_task.is_none() {
             return cosmic::task::future(async move {
                 zbus::Connection::system()
@@ -345,7 +337,7 @@ impl page::Page<crate::pages::Message> for Page {
                     .context("failed to create system dbus connection")
                     .map_or_else(
                         |why| Message::Error(ErrorKind::DbusConnection, why.to_string()),
-                        |conn| Message::NetworkManagerConnect((conn, sender.clone())),
+                        Message::NetworkManagerConnect,
                     )
             });
         }
@@ -625,8 +617,8 @@ impl Page {
                 self.dialog = Some(VpnDialog::Error(error_kind, why))
             }
 
-            Message::NetworkManagerConnect((conn, output)) => {
+            Message::NetworkManagerConnect(conn) => {
-                self.connect(conn.clone(), output);
+                return self.connect(conn.clone());
             }
         }
 
@@ -660,24 +652,36 @@ impl Page {
         })
     }
 
-    fn connect(
+    fn connect(&mut self, conn: zbus::Connection) -> Task<crate::app::Message> {
-        &mut self,
-        conn: zbus::Connection,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) {
         if self.nm_task.is_none() {
-            self.nm_task = Some(crate::utils::forward_event_loop(
+            let (canceller, task) = crate::utils::forward_event_loop(move |emitter| async move {
-                sender,
+                let (tx, mut rx) = futures::channel::mpsc::channel(1);
-                |event| crate::pages::Message::Vpn(Message::NetworkManager(event)),
+
-                move |tx| async move {
+                let watchers = std::pin::pin!(async move {
                     futures::join!(
                         network_manager::watch(conn.clone(), tx.clone()),
                         network_manager::active_conns::watch(conn.clone(), tx.clone()),
                         network_manager::devices::watch(conn, true, tx)
-                    );
+                    )
-                },
+                });
-            ));
+
+                let forwarder = std::pin::pin!(async move {
+                    while let Some(message) = rx.next().await {
+                        _ = emitter
+                            .emit(crate::pages::Message::Vpn(Message::NetworkManager(message)))
+                            .await;
+                    }
+                });
+
+                futures::future::select(watchers, forwarder).await;
+            });
+
+            self.nm_task = Some(canceller);
+
+            return task.map(crate::app::Message::from);
         }
+
+        Task::none()
     }
 
     /// Closes the view more popup and applies any withheld updates.

cosmic-settings/src/pages/networking/wifi.rs

@@ -43,12 +43,7 @@ pub enum Message {
     /// An update from the network manager daemon
     NetworkManager(network_manager::Event),
     /// Successfully connected to the system dbus.
-    NetworkManagerConnect(
+    NetworkManagerConnect(zbus::Connection),
-        (
-            zbus::Connection,
-            tokio::sync::mpsc::Sender<crate::pages::Message>,
-        ),
-    ),
     /// Request an auth dialog
     PasswordRequest(network_manager::SSID),
     /// Update the password from the dialog
@@ -201,10 +196,7 @@ impl page::Page<crate::pages::Message> for Page {
         )
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         if self.nm_task.is_none() {
             return cosmic::Task::future(async move {
                 zbus::Connection::system()
@@ -212,7 +204,7 @@ impl page::Page<crate::pages::Message> for Page {
                     .context("failed to create system dbus connection")
                     .map_or_else(
                         |why| Message::Error(why.to_string()),
-                        |conn| Message::NetworkManagerConnect((conn, sender.clone())),
+                        Message::NetworkManagerConnect,
                     )
                     .apply(crate::pages::Message::WiFi)
             });
@@ -444,35 +436,49 @@ impl Page {
                 self.active_device = Some(device);
             }
 
-            Message::NetworkManagerConnect((conn, output)) => {
+            Message::NetworkManagerConnect(conn) => {
-                self.connect(conn.clone(), output);
+                return cosmic::task::batch(vec![
-
+                    self.connect(conn.clone()),
-                return connection_settings(conn);
+                    connection_settings(conn),
+                ])
             }
         }
 
         Task::none()
     }
 
-    fn connect(
+    fn connect(&mut self, conn: zbus::Connection) -> Task<crate::app::Message> {
-        &mut self,
-        conn: zbus::Connection,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) {
         if self.nm_task.is_none() {
-            self.nm_task = Some(crate::utils::forward_event_loop(
+            let (canceller, task) = crate::utils::forward_event_loop(move |emitter| async move {
-                sender,
+                let (tx, mut rx) = futures::channel::mpsc::channel(1);
-                |event| crate::pages::Message::WiFi(Message::NetworkManager(event)),
+
-                move |tx| async move {
+                let watchers = std::pin::pin!(async move {
                     futures::join!(
                         network_manager::watch(conn.clone(), tx.clone()),
                         network_manager::active_conns::watch(conn.clone(), tx.clone()),
                         network_manager::wireless_enabled::watch(conn.clone(), tx.clone()),
                         network_manager::watch_connections_changed(conn, tx)
                     );
-                },
+                });
-            ));
+
+                let forwarder = std::pin::pin!(async move {
+                    while let Some(message) = rx.next().await {
+                        _ = emitter
+                            .emit(crate::pages::Message::WiFi(Message::NetworkManager(
+                                message,
+                            )))
+                            .await;
+                    }
+                });
+
+                futures::future::select(watchers, forwarder).await;
+            });
+
+            self.nm_task = Some(canceller);
+            return task.map(crate::app::Message::from);
         }
+
+        Task::none()
     }
 
     /// Closes the view more popup and applies any withheld updates.

cosmic-settings/src/pages/networking/wired.rs

@@ -15,6 +15,7 @@ use cosmic_settings_page::{self as page, section, Section};
 use cosmic_settings_subscriptions::network_manager::{
     self, current_networks::ActiveConnectionInfo, NetworkManagerState,
 };
+use futures::StreamExt;
 
 pub type ConnectionId = Arc<str>;
 
@@ -33,12 +34,7 @@ pub enum Message {
     /// An update from the network manager daemon
     NetworkManager(network_manager::Event),
     /// Successfully connected to the system dbus.
-    NetworkManagerConnect(
+    NetworkManagerConnect(zbus::Connection),
-        (
-            zbus::Connection,
-            tokio::sync::mpsc::Sender<crate::pages::Message>,
-        ),
-    ),
     /// Refresh devices and their connection profiles
     Refresh,
     /// Create a dialog to ask for confirmation of removal.
@@ -156,10 +152,7 @@ impl page::Page<crate::pages::Message> for Page {
         )
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         if self.nm_task.is_none() {
             return cosmic::task::future(async move {
                 zbus::Connection::system()
@@ -167,7 +160,7 @@ impl page::Page<crate::pages::Message> for Page {
                     .context("failed to create system dbus connection")
                     .map_or_else(
                         |why| Message::Error(why.to_string()),
-                        |conn| Message::NetworkManagerConnect((conn, sender.clone())),
+                        Message::NetworkManagerConnect,
                     )
                     .apply(crate::pages::Message::Wired)
             });
@@ -356,32 +349,45 @@ impl Page {
                 tracing::error!(why);
             }
 
-            Message::NetworkManagerConnect((conn, output)) => {
+            Message::NetworkManagerConnect(conn) => {
-                self.connect(conn.clone(), output);
+                return self.connect(conn);
             }
         }
 
         Task::none()
     }
 
-    fn connect(
+    fn connect(&mut self, conn: zbus::Connection) -> Task<crate::app::Message> {
-        &mut self,
-        conn: zbus::Connection,
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) {
         if self.nm_task.is_none() {
-            self.nm_task = Some(crate::utils::forward_event_loop(
+            let (canceller, task) = crate::utils::forward_event_loop(move |emitter| async move {
-                sender,
+                let (tx, mut rx) = futures::channel::mpsc::channel(1);
-                |event| crate::pages::Message::Wired(Message::NetworkManager(event)),
+
-                move |tx| async move {
+                let watchers = std::pin::pin!(async move {
                     futures::join!(
                         network_manager::watch(conn.clone(), tx.clone()),
                         network_manager::active_conns::watch(conn.clone(), tx.clone()),
                         network_manager::devices::watch(conn, true, tx)
-                    );
+                    )
-                },
+                });
-            ));
+
+                let forwarder = std::pin::pin!(async move {
+                    while let Some(message) = rx.next().await {
+                        _ = emitter
+                            .emit(crate::pages::Message::Wired(Message::NetworkManager(
+                                message,
+                            )))
+                            .await;
+                    }
+                });
+
+                futures::future::select(watchers, forwarder).await;
+            });
+
+            self.nm_task = Some(canceller);
+            return task.map(crate::app::Message::from);
         }
+
+        Task::none()
     }
 
     /// Closes the view more popup and applies any withheld updates.

cosmic-settings/src/pages/power/mod.rs

@@ -111,10 +111,7 @@ impl page::Page<crate::pages::Message> for Page {
         ])
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         let futures: Vec<Task<Message>> = vec![
             cosmic::Task::future(async move {
                 let battery = Battery::update_battery().await;
@@ -137,10 +134,10 @@ impl page::Page<crate::pages::Message> for Page {
                     let added_stream = ConnectedDevice::device_added_stream(&connection).await;
                     let removed_stream = ConnectedDevice::device_removed_stream(&connection).await;
 
-                    let added_future = async {
+                    let added_future = std::pin::pin!(async {
                         match added_stream {
                             Ok(stream) => {
-                                futures::pin_mut!(stream);
+                                let mut stream = std::pin::pin!(stream);
                                 while let Some(device) = stream.next().await {
                                     tracing::info!(device = device.model, "device added");
                                     emitter.emit(Message::DeviceConnect(device)).await;
@@ -148,12 +145,12 @@ impl page::Page<crate::pages::Message> for Page {
                             }
                             Err(err) => tracing::error!(?err, "cannot establish added stream"),
                         }
-                    };
+                    });
 
-                    let removed_future = async {
+                    let removed_future = std::pin::pin!(async {
                         match removed_stream {
                             Ok(stream) => {
-                                futures::pin_mut!(stream);
+                                let mut stream = std::pin::pin!(stream);
                                 while let Some(device_path) = stream.next().await {
                                     tracing::info!(device_path, "device removed");
                                     emitter.emit(Message::DeviceDisconnect(device_path)).await;
@@ -161,10 +158,7 @@ impl page::Page<crate::pages::Message> for Page {
                             }
                             Err(err) => tracing::error!(?err, "cannot establish removed stream"),
                         }
-                    };
+                    });
-
-                    futures::pin_mut!(added_future);
-                    futures::pin_mut!(removed_future);
 
                     futures::future::select(added_future, removed_future).await;
                 }),

cosmic-settings/src/pages/sound.rs

@@ -129,13 +129,9 @@ impl page::Page<crate::pages::Message> for Page {
             .description(fl!("sound", "desc"))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
+        let mut tasks = Vec::with_capacity(2);
-        sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         if self.pulse_thread.is_none() {
-            let sender = sender.clone();
-
             let (tx, mut rx) = futures::channel::mpsc::channel(1);
             let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>();
 
@@ -146,18 +142,18 @@ impl page::Page<crate::pages::Message> for Page {
 
             // Forward events from the pulse thread to the application until
             // the application requests to stop listening to the pulse thread.
-            tokio::task::spawn(async move {
+            tasks.push(Task::stream(async_fn_stream::fn_stream(
-                let forwarder = std::pin::pin!(async move {
+                |emitter| async move {
-                    while let Some(event) = rx.next().await {
+                    let forwarder = std::pin::pin!(async move {
-                        let event = crate::pages::Message::Sound(Message::Pulse(event));
+                        while let Some(event) = rx.next().await {
-                        if sender.send(event).await.is_err() {
+                            let event = crate::pages::Message::Sound(Message::Pulse(event));
-                            break;
+                            emitter.emit(event).await;
                         }
-                    }
+                    });
-                });
 
-                futures::future::select(std::pin::pin!(cancel_rx), forwarder).await;
+                    futures::future::select(std::pin::pin!(cancel_rx), forwarder).await;
-            });
+                },
+            )));
 
             self.pulse_thread = Some(cancel_tx);
         }
@@ -171,23 +167,23 @@ impl page::Page<crate::pages::Message> for Page {
 
             // Forward events from the pipewire thread to the application until
             // the application requests to stop listening to the pulse thread.
-            tokio::task::spawn(async move {
+            tasks.push(Task::stream(async_fn_stream::fn_stream(
-                let forwarder = std::pin::pin!(async move {
+                |emitter| async move {
-                    while let Some(event) = rx.next().await {
+                    let forwarder = std::pin::pin!(async move {
-                        let event = crate::pages::Message::Sound(Message::Pipewire(event));
+                        while let Some(event) = rx.next().await {
-                        if sender.send(event).await.is_err() {
+                            let event = crate::pages::Message::Sound(Message::Pipewire(event));
-                            break;
+                            emitter.emit(event).await;
                         }
-                    }
+                    });
-                });
 
-                futures::future::select(std::pin::pin!(cancel_rx), forwarder).await;
+                    futures::future::select(std::pin::pin!(cancel_rx), forwarder).await;
-            });
+                },
+            )));
 
             self.pipewire_thread = Some((cancel_tx, terminate));
         }
 
-        Task::none()
+        cosmic::task::batch(tasks)
     }
 
     fn on_leave(&mut self) -> Task<crate::pages::Message> {

cosmic-settings/src/pages/system/about.rs

@@ -49,10 +49,7 @@ impl page::Page<crate::pages::Message> for Page {
             .description(fl!("about", "desc"))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         let (task, handle) = Task::future(async move {
             crate::pages::Message::About(Message::Info(Box::new(Info::load())))
         })

cosmic-settings/src/pages/system/default_apps.rs

@@ -21,7 +21,6 @@ use freedesktop_desktop_entry::{
 };
 use mime_apps::App;
 use slotmap::SlotMap;
-use tokio::sync::mpsc;
 
 const DROPDOWN_WEB_BROWSER: usize = 0;
 const DROPDOWN_FILE_MANAGER: usize = 1;
@@ -105,10 +104,7 @@ impl page::Page<crate::pages::Message> for Page {
             .description(fl!("default-apps", "desc"))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         if let Some(handle) = self.on_enter_handle.take() {
             handle.abort();
         }

cosmic-settings/src/pages/system/users/mod.rs

@@ -222,10 +222,7 @@ impl page::Page<crate::pages::Message> for Page {
         dialog_element.map(crate::pages::Message::User).into()
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         if let Some(handle) = self.on_enter_handle.take() {
             handle.abort();
         }

cosmic-settings/src/pages/time/date.rs

@@ -133,10 +133,7 @@ impl page::Page<crate::pages::Message> for Page {
             .description(fl!("time-date", "desc"))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> Task<crate::pages::Message> {
-        &mut self,
-        _sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
-    ) -> Task<crate::pages::Message> {
         cosmic::Task::future(async move {
             let client = match zbus::Connection::system().await {
                 Ok(client) => client,

cosmic-settings/src/pages/time/region.rs

@@ -23,7 +23,6 @@ use icu::decimal::options::FixedDecimalFormatterOptions;
 use icu::decimal::FixedDecimalFormatter;
 use lichen_system::locale;
 use slotmap::{DefaultKey, SlotMap};
-use tokio::sync::mpsc;
 
 #[derive(Clone, Debug)]
 pub enum Message {
@@ -136,10 +135,7 @@ impl page::Page<crate::pages::Message> for Page {
             .description(fl!("time-region", "desc"))
     }
 
-    fn on_enter(
+    fn on_enter(&mut self) -> cosmic::Task<crate::pages::Message> {
-        &mut self,
-        _sender: mpsc::Sender<crate::pages::Message>,
-    ) -> cosmic::Task<crate::pages::Message> {
         cosmic::task::future(async move { Message::Refresh(Arc::new(page_reload().await)) })
     }
 

cosmic-settings/src/utils.rs

@@ -1,6 +1,6 @@
 use std::{future::Future, io, process};
 
-use futures::{future::select, StreamExt};
+use futures::future::select;
 
 /// Normalize the labeling of displays across settings pages.
 pub fn display_name(name: &str, physical: (u32, u32)) -> String {
@@ -22,29 +22,19 @@ pub fn display_name(name: &str, physical: (u32, u32)) -> String {
 
 /// Spawn a background tasks and forward its messages
 pub fn forward_event_loop<M: 'static + Send, T: Future<Output = ()> + Send + 'static>(
-    sender: tokio::sync::mpsc::Sender<crate::pages::Message>,
+    event_loop: impl FnOnce(async_fn_stream::StreamEmitter<M>) -> T + Send + 'static,
-    message_map: fn(M) -> crate::pages::Message,
+) -> (tokio::sync::oneshot::Sender<()>, cosmic::Task<M>) {
-    event_loop: impl FnOnce(futures::channel::mpsc::Sender<M>) -> T + Send + 'static,
-) -> tokio::sync::oneshot::Sender<()> {
     let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>();
 
-    tokio::task::spawn(async move {
+    let task = cosmic::Task::stream(async_fn_stream::fn_stream(|emitter| async move {
-        let (tx, mut rx) = futures::channel::mpsc::channel(1);
+        select(
+            std::pin::pin!(cancel_rx),
+            std::pin::pin!(event_loop(emitter)),
+        )
+        .await;
+    }));
 
-        let cancel = std::pin::pin!(cancel_rx);
+    (cancel_tx, task)
-
-        let forwarder = std::pin::pin!(async move {
-            while let Some(event) = rx.next().await {
-                if sender.send(message_map(event)).await.is_err() {
-                    break;
-                }
-            }
-        });
-
-        select(cancel, select(forwarder, std::pin::pin!(event_loop(tx)))).await;
-    });
-
-    cancel_tx
 }
 
 /// On process failure, return stderr as `String`.

page/src/binder.rs

@@ -74,7 +74,7 @@ impl<Message: 'static> Binder<Message> {
         if self.contains_item(id) {
             self.storage
                 .entry(TypeId::of::<Data>())
-                .or_insert_with(SecondaryMap::new)
+                .or_default()
                 .insert(id, Box::new(data));
         }
     }
@@ -168,13 +168,9 @@ impl<Message: 'static> Binder<Message> {
     }
 
     /// Calls a page's load function to refresh its data.
-    pub fn on_enter(
+    pub fn on_enter(&mut self, id: crate::Entity) -> Task<Message> {
-        &mut self,
-        id: crate::Entity,
-        sender: tokio::sync::mpsc::Sender<Message>,
-    ) -> Task<Message> {
         if let Some(page) = self.page.get_mut(id) {
-            return page.on_enter(sender);
+            return page.on_enter();
         }
 
         Task::none()

page/src/lib.rs

@@ -65,7 +65,7 @@ pub trait Page<Message: 'static>: Downcast {
 
     /// Reload page metadata via a Task.
     #[allow(unused)]
-    fn on_enter(&mut self, sender: tokio::sync::mpsc::Sender<Message>) -> Task<Message> {
+    fn on_enter(&mut self) -> Task<Message> {
         Task::none()
     }