From dda70e1441de8611eeb33e67432f11f2a5db7787 Mon Sep 17 00:00:00 2001 From: Michael Aaron Murphy Date: Fri, 28 Feb 2025 16:55:48 +0100 Subject: [PATCH] refactor: use Task::stream instead of a channel subscription --- cosmic-settings/src/app.rs | 55 ++--------- .../src/pages/accessibility/magnifier.rs | 34 +++---- .../src/pages/accessibility/mod.rs | 34 +++---- cosmic-settings/src/pages/bluetooth/mod.rs | 64 ++++++++----- .../src/pages/desktop/appearance/mod.rs | 5 +- .../src/pages/desktop/panel/applets_inner.rs | 5 +- .../src/pages/desktop/wallpaper/mod.rs | 5 +- cosmic-settings/src/pages/display/mod.rs | 27 ++---- .../src/pages/input/keyboard/mod.rs | 5 +- .../pages/input/keyboard/shortcuts/custom.rs | 5 +- .../keyboard/shortcuts/manage_windows.rs | 5 +- .../src/pages/input/keyboard/shortcuts/mod.rs | 5 +- .../input/keyboard/shortcuts/move_window.rs | 5 +- .../src/pages/input/keyboard/shortcuts/nav.rs | 5 +- .../pages/input/keyboard/shortcuts/system.rs | 5 +- .../pages/input/keyboard/shortcuts/tiling.rs | 5 +- cosmic-settings/src/pages/networking/mod.rs | 91 +++++++++---------- .../src/pages/networking/vpn/mod.rs | 54 ++++++----- cosmic-settings/src/pages/networking/wifi.rs | 58 ++++++------ cosmic-settings/src/pages/networking/wired.rs | 56 +++++++----- cosmic-settings/src/pages/power/mod.rs | 20 ++-- cosmic-settings/src/pages/sound.rs | 50 +++++----- cosmic-settings/src/pages/system/about.rs | 5 +- .../src/pages/system/default_apps.rs | 6 +- cosmic-settings/src/pages/system/users/mod.rs | 5 +- cosmic-settings/src/pages/time/date.rs | 5 +- cosmic-settings/src/pages/time/region.rs | 6 +- cosmic-settings/src/utils.rs | 32 +++---- page/src/binder.rs | 10 +- page/src/lib.rs | 2 +- 30 files changed, 284 insertions(+), 385 deletions(-) diff --git a/cosmic-settings/src/app.rs b/cosmic-settings/src/app.rs index e413af4..ce70ff6 100644 --- a/cosmic-settings/src/app.rs +++ b/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::{stream, Subscription}; +use cosmic::iced::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>, pages: page::Binder, 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), 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 { 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::(), - stream::channel(4, move |mut output| async move { - let (tx, mut rx) = tokio::sync::mpsc::channel::(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 { - self.loaded_pages.insert(page); - tasks.push(self.pages.on_enter(page, sender.clone())); - } + for page in load { + self.loaded_pages.insert(page); + tasks.push(self.pages.on_enter(page)); } for page in unload { diff --git a/cosmic-settings/src/pages/accessibility/magnifier.rs b/cosmic-settings/src/pages/accessibility/magnifier.rs index d609dba..049ebbd 100644 --- a/cosmic-settings/src/pages/accessibility/magnifier.rs +++ b/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 for Page { ]) } - fn on_enter( - &mut self, - sender: mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { 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 { - let _ = sender - .send(crate::pages::Message::AccessibilityMagnifier( - Message::Event(event), + + return cosmic::Task::stream(async_fn_stream::fn_stream( + |emitter| async move { + while let Some(event) = rx.recv().await { + 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!( diff --git a/cosmic-settings/src/pages/accessibility/mod.rs b/cosmic-settings/src/pages/accessibility/mod.rs index 718ee93..313a072 100644 --- a/cosmic-settings/src/pages/accessibility/mod.rs +++ b/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 for Page { Some(vec![sections.insert(vision())]) } - fn on_enter( - &mut self, - sender: mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { 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 { - let _ = sender - .send(crate::pages::Message::Accessibility(Message::Event(event))) + + return cosmic::Task::stream(async_fn_stream::fn_stream( + |emitter| async move { + 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!( diff --git a/cosmic-settings/src/pages/bluetooth/mod.rs b/cosmic-settings/src/pages/bluetooth/mod.rs index 4051f9e..c6f24f9 100644 --- a/cosmic-settings/src/pages/bluetooth/mod.rs +++ b/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 for Page { ]) } - fn on_enter( - &mut self, - sender: tokio::sync::mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { // 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 for Page { pub enum Message { BluetoothEvent(Event), ConnectDevice(OwnedObjectPath), - DBusConnect( - zbus::Connection, - tokio::sync::mpsc::Sender, - ), + DBusConnect(zbus::Connection), 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; diff --git a/cosmic-settings/src/pages/desktop/appearance/mod.rs b/cosmic-settings/src/pages/desktop/appearance/mod.rs index eb58009..a107038 100644 --- a/cosmic-settings/src/pages/desktop/appearance/mod.rs +++ b/cosmic-settings/src/pages/desktop/appearance/mod.rs @@ -1562,10 +1562,7 @@ impl page::Page for Page { .description(fl!("appearance", "desc")) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { let (task, handle) = cosmic::task::batch(vec![ // Load icon themes // cosmic::task::future(icon_themes::fetch()).map(crate::pages::Message::Appearance), diff --git a/cosmic-settings/src/pages/desktop/panel/applets_inner.rs b/cosmic-settings/src/pages/desktop/panel/applets_inner.rs index 085d6f4..6993e11 100644 --- a/cosmic-settings/src/pages/desktop/panel/applets_inner.rs +++ b/cosmic-settings/src/pages/desktop/panel/applets_inner.rs @@ -143,10 +143,7 @@ impl page::Page for Page { }) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { Task::none() } } diff --git a/cosmic-settings/src/pages/desktop/wallpaper/mod.rs b/cosmic-settings/src/pages/desktop/wallpaper/mod.rs index f2f0b4a..e1ff6d1 100644 --- a/cosmic-settings/src/pages/desktop/wallpaper/mod.rs +++ b/cosmic-settings/src/pages/desktop/wallpaper/mod.rs @@ -210,10 +210,7 @@ impl page::Page for Page { .description(fl!("wallpaper", "desc")) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { // Check if the page is already being loaded. if self.on_enter_handle.is_some() { return Task::none(); diff --git a/cosmic-settings/src/pages/display/mod.rs b/cosmic-settings/src/pages/display/mod.rs index a2286fc..099a7b2 100644 --- a/cosmic-settings/src/pages/display/mod.rs +++ b/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 for Page { } #[cfg(not(feature = "test"))] - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { use std::time::Duration; self.cache.orientations = [ @@ -290,7 +286,7 @@ impl page::Page 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 for Page { return; } } - }; + }); - pin_mut!(dispatcher); runtime.block_on(futures::future::select(cancelled, dispatcher)); }); @@ -365,7 +360,7 @@ impl page::Page 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 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 for Page { } #[cfg(feature = "test")] - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { cosmic::task::future(async move { let mut randr = List::default(); diff --git a/cosmic-settings/src/pages/input/keyboard/mod.rs b/cosmic-settings/src/pages/input/keyboard/mod.rs index 77dfa7f..c37f51d 100644 --- a/cosmic-settings/src/pages/input/keyboard/mod.rs +++ b/cosmic-settings/src/pages/input/keyboard/mod.rs @@ -300,10 +300,7 @@ impl page::Page for Page { } } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { self.xkb = super::get_config(&self.config, "xkb_config"); match ( xkb_data::keyboard_layouts(), diff --git a/cosmic-settings/src/pages/input/keyboard/shortcuts/custom.rs b/cosmic-settings/src/pages/input/keyboard/shortcuts/custom.rs index f13b591..d9b9e83 100644 --- a/cosmic-settings/src/pages/input/keyboard/shortcuts/custom.rs +++ b/cosmic-settings/src/pages/input/keyboard/shortcuts/custom.rs @@ -358,10 +358,7 @@ impl page::Page for Page { .map(|el| el.map(crate::pages::Message::CustomShortcuts)) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { self.model.on_enter(); Task::none() } diff --git a/cosmic-settings/src/pages/input/keyboard/shortcuts/manage_windows.rs b/cosmic-settings/src/pages/input/keyboard/shortcuts/manage_windows.rs index 403cb42..1a7550d 100644 --- a/cosmic-settings/src/pages/input/keyboard/shortcuts/manage_windows.rs +++ b/cosmic-settings/src/pages/input/keyboard/shortcuts/manage_windows.rs @@ -59,10 +59,7 @@ impl page::Page for Page { .map(|el| el.map(crate::pages::Message::ManageWindowShortcuts)) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { self.model.on_enter(); Task::none() diff --git a/cosmic-settings/src/pages/input/keyboard/shortcuts/mod.rs b/cosmic-settings/src/pages/input/keyboard/shortcuts/mod.rs index f3f09ef..674b35d 100644 --- a/cosmic-settings/src/pages/input/keyboard/shortcuts/mod.rs +++ b/cosmic-settings/src/pages/input/keyboard/shortcuts/mod.rs @@ -142,10 +142,7 @@ impl page::Page for Page { } } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { if self.shortcuts_context.is_none() { self.shortcuts_context = cosmic_settings_config::shortcuts::context().ok(); } diff --git a/cosmic-settings/src/pages/input/keyboard/shortcuts/move_window.rs b/cosmic-settings/src/pages/input/keyboard/shortcuts/move_window.rs index 2efa75d..4beb9e4 100644 --- a/cosmic-settings/src/pages/input/keyboard/shortcuts/move_window.rs +++ b/cosmic-settings/src/pages/input/keyboard/shortcuts/move_window.rs @@ -59,10 +59,7 @@ impl page::Page for Page { .map(|el| el.map(crate::pages::Message::MoveWindowShortcuts)) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { self.model.on_enter(); Task::none() diff --git a/cosmic-settings/src/pages/input/keyboard/shortcuts/nav.rs b/cosmic-settings/src/pages/input/keyboard/shortcuts/nav.rs index 5acdf6d..7d86d6f 100644 --- a/cosmic-settings/src/pages/input/keyboard/shortcuts/nav.rs +++ b/cosmic-settings/src/pages/input/keyboard/shortcuts/nav.rs @@ -59,10 +59,7 @@ impl page::Page for Page { .map(|el| el.map(crate::pages::Message::NavShortcuts)) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { self.model.on_enter(); Task::none() diff --git a/cosmic-settings/src/pages/input/keyboard/shortcuts/system.rs b/cosmic-settings/src/pages/input/keyboard/shortcuts/system.rs index b4b5163..0dbc890 100644 --- a/cosmic-settings/src/pages/input/keyboard/shortcuts/system.rs +++ b/cosmic-settings/src/pages/input/keyboard/shortcuts/system.rs @@ -59,10 +59,7 @@ impl page::Page for Page { .map(|el| el.map(crate::pages::Message::SystemShortcuts)) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { self.model.on_enter(); Task::none() diff --git a/cosmic-settings/src/pages/input/keyboard/shortcuts/tiling.rs b/cosmic-settings/src/pages/input/keyboard/shortcuts/tiling.rs index 7467327..b312ad8 100644 --- a/cosmic-settings/src/pages/input/keyboard/shortcuts/tiling.rs +++ b/cosmic-settings/src/pages/input/keyboard/shortcuts/tiling.rs @@ -59,10 +59,7 @@ impl page::Page for Page { .map(|el| el.map(crate::pages::Message::TilingShortcuts)) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { self.model.on_enter(); Task::none() diff --git a/cosmic-settings/src/pages/networking/mod.rs b/cosmic-settings/src/pages/networking/mod.rs index c4ca1ca..75e8b44 100644 --- a/cosmic-settings/src/pages/networking/mod.rs +++ b/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( - ( - zbus::Connection, - tokio::sync::mpsc::Sender, - ), - ), + NetworkManagerConnect(zbus::Connection), /// Open the wifi settings page with the selected device. OpenPage { page: page::Entity, @@ -224,10 +219,7 @@ impl page::Page for Page { Some(vec![sections.insert(device_list)]) } - fn on_enter( - &mut self, - sender: tokio::sync::mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { if self.nm_task.is_none() { return cosmic::Task::future(async move { zbus::Connection::system() @@ -235,7 +227,7 @@ impl page::Page 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)) => { - self.connect(conn.clone(), output); + Message::NetworkManagerConnect(conn) => { + return self.connect(conn.clone()); } Message::Error(why) => { @@ -315,46 +307,45 @@ impl Page { Task::none() } - fn connect( - &mut self, - conn: zbus::Connection, - sender: tokio::sync::mpsc::Sender, - ) { + fn connect(&mut self, conn: zbus::Connection) -> Task { if self.nm_task.is_none() { - self.nm_task = Some(crate::utils::forward_event_loop( - sender, - crate::pages::Message::Networking, - move |mut tx| async move { - let network_manager = match NetworkManager::new(&conn).await { - Ok(n) => n, - Err(why) => { - tracing::error!( - why = why.to_string(), - "failed to connect to network_manager" - ); + let (canceller, task) = crate::utils::forward_event_loop(move |emitter| async move { + let network_manager = match NetworkManager::new(&conn).await { + Ok(n) => n, + Err(why) => { + tracing::error!( + why = why.to_string(), + "failed to connect to network_manager" + ); - 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; + 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 { + _ = emitter + .emit(crate::pages::Message::Networking(message)) + .await; + } + }); + + self.nm_task = Some(canceller); + return task.map(crate::app::Message::from); } + + Task::none() } } diff --git a/cosmic-settings/src/pages/networking/vpn/mod.rs b/cosmic-settings/src/pages/networking/vpn/mod.rs index 8dea9b0..dc04812 100644 --- a/cosmic-settings/src/pages/networking/vpn/mod.rs +++ b/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( - ( - zbus::Connection, - tokio::sync::mpsc::Sender, - ), - ), + NetworkManagerConnect(zbus::Connection), /// Updates the password text input PasswordUpdate(SecureString), /// Refresh devices and their connection profiles @@ -334,10 +329,7 @@ impl page::Page for Page { ) } - fn on_enter( - &mut self, - sender: tokio::sync::mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { if self.nm_task.is_none() { return cosmic::task::future(async move { zbus::Connection::system() @@ -345,7 +337,7 @@ impl page::Page 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)) => { - self.connect(conn.clone(), output); + Message::NetworkManagerConnect(conn) => { + return self.connect(conn.clone()); } } @@ -660,24 +652,36 @@ impl Page { }) } - fn connect( - &mut self, - conn: zbus::Connection, - sender: tokio::sync::mpsc::Sender, - ) { + fn connect(&mut self, conn: zbus::Connection) -> Task { if self.nm_task.is_none() { - self.nm_task = Some(crate::utils::forward_event_loop( - sender, - |event| crate::pages::Message::Vpn(Message::NetworkManager(event)), - move |tx| async move { + let (canceller, task) = crate::utils::forward_event_loop(move |emitter| async move { + let (tx, mut rx) = futures::channel::mpsc::channel(1); + + 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. diff --git a/cosmic-settings/src/pages/networking/wifi.rs b/cosmic-settings/src/pages/networking/wifi.rs index 26b718d..b4020ee 100644 --- a/cosmic-settings/src/pages/networking/wifi.rs +++ b/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( - ( - zbus::Connection, - tokio::sync::mpsc::Sender, - ), - ), + NetworkManagerConnect(zbus::Connection), /// Request an auth dialog PasswordRequest(network_manager::SSID), /// Update the password from the dialog @@ -201,10 +196,7 @@ impl page::Page for Page { ) } - fn on_enter( - &mut self, - sender: tokio::sync::mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { if self.nm_task.is_none() { return cosmic::Task::future(async move { zbus::Connection::system() @@ -212,7 +204,7 @@ impl page::Page 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)) => { - self.connect(conn.clone(), output); - - return connection_settings(conn); + Message::NetworkManagerConnect(conn) => { + return cosmic::task::batch(vec![ + self.connect(conn.clone()), + connection_settings(conn), + ]) } } Task::none() } - fn connect( - &mut self, - conn: zbus::Connection, - sender: tokio::sync::mpsc::Sender, - ) { + fn connect(&mut self, conn: zbus::Connection) -> Task { if self.nm_task.is_none() { - self.nm_task = Some(crate::utils::forward_event_loop( - sender, - |event| crate::pages::Message::WiFi(Message::NetworkManager(event)), - move |tx| async move { + let (canceller, task) = crate::utils::forward_event_loop(move |emitter| async move { + let (tx, mut rx) = futures::channel::mpsc::channel(1); + + 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. diff --git a/cosmic-settings/src/pages/networking/wired.rs b/cosmic-settings/src/pages/networking/wired.rs index 4b06f8c..3c0f47e 100644 --- a/cosmic-settings/src/pages/networking/wired.rs +++ b/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; @@ -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( - ( - zbus::Connection, - tokio::sync::mpsc::Sender, - ), - ), + NetworkManagerConnect(zbus::Connection), /// Refresh devices and their connection profiles Refresh, /// Create a dialog to ask for confirmation of removal. @@ -156,10 +152,7 @@ impl page::Page for Page { ) } - fn on_enter( - &mut self, - sender: tokio::sync::mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { if self.nm_task.is_none() { return cosmic::task::future(async move { zbus::Connection::system() @@ -167,7 +160,7 @@ impl page::Page 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)) => { - self.connect(conn.clone(), output); + Message::NetworkManagerConnect(conn) => { + return self.connect(conn); } } Task::none() } - fn connect( - &mut self, - conn: zbus::Connection, - sender: tokio::sync::mpsc::Sender, - ) { + fn connect(&mut self, conn: zbus::Connection) -> Task { if self.nm_task.is_none() { - self.nm_task = Some(crate::utils::forward_event_loop( - sender, - |event| crate::pages::Message::Wired(Message::NetworkManager(event)), - move |tx| async move { + let (canceller, task) = crate::utils::forward_event_loop(move |emitter| async move { + let (tx, mut rx) = futures::channel::mpsc::channel(1); + + 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. diff --git a/cosmic-settings/src/pages/power/mod.rs b/cosmic-settings/src/pages/power/mod.rs index 41afe45..dc515e8 100644 --- a/cosmic-settings/src/pages/power/mod.rs +++ b/cosmic-settings/src/pages/power/mod.rs @@ -111,10 +111,7 @@ impl page::Page for Page { ]) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { let futures: Vec> = vec![ cosmic::Task::future(async move { let battery = Battery::update_battery().await; @@ -137,10 +134,10 @@ impl page::Page 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 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 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; }), diff --git a/cosmic-settings/src/pages/sound.rs b/cosmic-settings/src/pages/sound.rs index d36ea00..5b41a3b 100644 --- a/cosmic-settings/src/pages/sound.rs +++ b/cosmic-settings/src/pages/sound.rs @@ -129,13 +129,9 @@ impl page::Page for Page { .description(fl!("sound", "desc")) } - fn on_enter( - &mut self, - sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { + let mut tasks = Vec::with_capacity(2); 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 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 { - let forwarder = std::pin::pin!(async move { - while let Some(event) = rx.next().await { - let event = crate::pages::Message::Sound(Message::Pulse(event)); - if sender.send(event).await.is_err() { - break; + tasks.push(Task::stream(async_fn_stream::fn_stream( + |emitter| async move { + let forwarder = std::pin::pin!(async move { + while let Some(event) = rx.next().await { + let event = crate::pages::Message::Sound(Message::Pulse(event)); + 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 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 { - let forwarder = std::pin::pin!(async move { - while let Some(event) = rx.next().await { - let event = crate::pages::Message::Sound(Message::Pipewire(event)); - if sender.send(event).await.is_err() { - break; + tasks.push(Task::stream(async_fn_stream::fn_stream( + |emitter| async move { + let forwarder = std::pin::pin!(async move { + while let Some(event) = rx.next().await { + let event = crate::pages::Message::Sound(Message::Pipewire(event)); + 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 { diff --git a/cosmic-settings/src/pages/system/about.rs b/cosmic-settings/src/pages/system/about.rs index cc98d84..9833da5 100644 --- a/cosmic-settings/src/pages/system/about.rs +++ b/cosmic-settings/src/pages/system/about.rs @@ -49,10 +49,7 @@ impl page::Page for Page { .description(fl!("about", "desc")) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { let (task, handle) = Task::future(async move { crate::pages::Message::About(Message::Info(Box::new(Info::load()))) }) diff --git a/cosmic-settings/src/pages/system/default_apps.rs b/cosmic-settings/src/pages/system/default_apps.rs index 096838a..f73ac55 100644 --- a/cosmic-settings/src/pages/system/default_apps.rs +++ b/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 for Page { .description(fl!("default-apps", "desc")) } - fn on_enter( - &mut self, - _sender: mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { if let Some(handle) = self.on_enter_handle.take() { handle.abort(); } diff --git a/cosmic-settings/src/pages/system/users/mod.rs b/cosmic-settings/src/pages/system/users/mod.rs index ce17914..89f86e0 100644 --- a/cosmic-settings/src/pages/system/users/mod.rs +++ b/cosmic-settings/src/pages/system/users/mod.rs @@ -222,10 +222,7 @@ impl page::Page for Page { dialog_element.map(crate::pages::Message::User).into() } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { if let Some(handle) = self.on_enter_handle.take() { handle.abort(); } diff --git a/cosmic-settings/src/pages/time/date.rs b/cosmic-settings/src/pages/time/date.rs index c25801a..9547883 100644 --- a/cosmic-settings/src/pages/time/date.rs +++ b/cosmic-settings/src/pages/time/date.rs @@ -133,10 +133,7 @@ impl page::Page for Page { .description(fl!("time-date", "desc")) } - fn on_enter( - &mut self, - _sender: tokio::sync::mpsc::Sender, - ) -> Task { + fn on_enter(&mut self) -> Task { cosmic::Task::future(async move { let client = match zbus::Connection::system().await { Ok(client) => client, diff --git a/cosmic-settings/src/pages/time/region.rs b/cosmic-settings/src/pages/time/region.rs index 15a1b59..03e3a3c 100644 --- a/cosmic-settings/src/pages/time/region.rs +++ b/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 for Page { .description(fl!("time-region", "desc")) } - fn on_enter( - &mut self, - _sender: mpsc::Sender, - ) -> cosmic::Task { + fn on_enter(&mut self) -> cosmic::Task { cosmic::task::future(async move { Message::Refresh(Arc::new(page_reload().await)) }) } diff --git a/cosmic-settings/src/utils.rs b/cosmic-settings/src/utils.rs index 7fe5c18..9ad6c0c 100644 --- a/cosmic-settings/src/utils.rs +++ b/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 + Send + 'static>( - sender: tokio::sync::mpsc::Sender, - message_map: fn(M) -> crate::pages::Message, - event_loop: impl FnOnce(futures::channel::mpsc::Sender) -> T + Send + 'static, -) -> tokio::sync::oneshot::Sender<()> { + event_loop: impl FnOnce(async_fn_stream::StreamEmitter) -> T + Send + 'static, +) -> (tokio::sync::oneshot::Sender<()>, cosmic::Task) { let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>(); - tokio::task::spawn(async move { - let (tx, mut rx) = futures::channel::mpsc::channel(1); + let task = cosmic::Task::stream(async_fn_stream::fn_stream(|emitter| async move { + select( + std::pin::pin!(cancel_rx), + std::pin::pin!(event_loop(emitter)), + ) + .await; + })); - let cancel = std::pin::pin!(cancel_rx); - - 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 + (cancel_tx, task) } /// On process failure, return stderr as `String`. diff --git a/page/src/binder.rs b/page/src/binder.rs index 4ba6309..85e8de6 100644 --- a/page/src/binder.rs +++ b/page/src/binder.rs @@ -74,7 +74,7 @@ impl Binder { if self.contains_item(id) { self.storage .entry(TypeId::of::()) - .or_insert_with(SecondaryMap::new) + .or_default() .insert(id, Box::new(data)); } } @@ -168,13 +168,9 @@ impl Binder { } /// Calls a page's load function to refresh its data. - pub fn on_enter( - &mut self, - id: crate::Entity, - sender: tokio::sync::mpsc::Sender, - ) -> Task { + pub fn on_enter(&mut self, id: crate::Entity) -> Task { if let Some(page) = self.page.get_mut(id) { - return page.on_enter(sender); + return page.on_enter(); } Task::none() diff --git a/page/src/lib.rs b/page/src/lib.rs index c5cac6b..f774ba8 100644 --- a/page/src/lib.rs +++ b/page/src/lib.rs @@ -65,7 +65,7 @@ pub trait Page: Downcast { /// Reload page metadata via a Task. #[allow(unused)] - fn on_enter(&mut self, sender: tokio::sync::mpsc::Sender) -> Task { + fn on_enter(&mut self) -> Task { Task::none() }