diff --git a/shell/ferese-shell/src/control.rs b/shell/ferese-shell/src/control.rs index 2be1155..1a1db5a 100644 --- a/shell/ferese-shell/src/control.rs +++ b/shell/ferese-shell/src/control.rs @@ -1,8 +1,8 @@ use std::error::Error; use std::os::fd::{FromRawFd, RawFd}; use std::os::unix::net::UnixStream; -use std::sync::mpsc; use std::sync::mpsc::{Receiver, Sender, TryRecvError}; +use std::sync::{Arc, mpsc}; use std::thread; use ferese_protocols::shell::v1::client::ferese_shell_manager_v1::FereseShellManagerV1; @@ -60,6 +60,7 @@ pub(crate) struct ShellControl { _manager: FereseShellManagerV1, shell: FereseShellV1, updates: Receiver, + wake: Wake, } pub(crate) struct ControlPoll { @@ -67,9 +68,44 @@ pub(crate) struct ControlPoll { pub(crate) snapshot: Option, pub(crate) overview_active: Option, pub(crate) disconnected: bool, - pub(crate) logout: Option<(u32, String)>, - pub(crate) logout_cancelled: Vec, - pub(crate) guide_toggles: Vec, + pub(crate) commands: Vec, +} + +pub(crate) enum ControlCommand { + Logout(u32, String), + CancelLogout(u32), + ToggleGuide(String), +} + +#[derive(Clone)] +struct Wake(Arc); + +impl std::hash::Hash for Wake { + fn hash(&self, state: &mut H) { + std::ptr::hash(Arc::as_ptr(&self.0), state); + } +} + +impl Wake { + fn stream(&self) -> impl cosmic::iced::futures::Stream + use<> { + cosmic::iced::futures::stream::unfold(self.clone(), |wake| async move { + wake.0.notified().await; + Some(((), wake)) + }) + } +} + +struct UpdateSender { + sender: Sender, + wake: Wake, +} + +impl UpdateSender { + fn send(&self, update: ControlUpdate) -> Result<(), mpsc::SendError> { + self.sender.send(update)?; + self.wake.0.notify_one(); + Ok(()) + } } enum ControlUpdate { @@ -90,7 +126,11 @@ impl ShellControl { let manager = globals.bind::(&qh, 4..=5, ())?; let shell = manager.get_shell(&qh, ()); let (sender, updates) = mpsc::channel(); - let mut state = ControlState::new(sender); + let wake = Wake(Arc::new(tokio::sync::Notify::new())); + let mut state = ControlState::new(UpdateSender { + sender, + wake: wake.clone(), + }); connection.flush()?; thread::Builder::new() @@ -105,30 +145,28 @@ impl ShellControl { _manager: manager, shell, updates, + wake, }) } + pub(crate) fn subscription(&self) -> cosmic::iced::Subscription<()> { + cosmic::iced::Subscription::run_with(self.wake.clone(), Wake::stream) + } + pub(crate) fn poll(&self) -> ControlPoll { let mut poll = ControlPoll { config: None, snapshot: None, overview_active: None, disconnected: false, - logout: None, - logout_cancelled: Vec::new(), - guide_toggles: Vec::new(), + commands: Vec::new(), }; loop { match self.updates.try_recv() { - Ok(ControlUpdate::Logout(serial, output)) => poll.logout = Some((serial, output)), - Ok(ControlUpdate::ToggleGuide(output)) => poll.guide_toggles.push(output), - Ok(ControlUpdate::LogoutCancelled(serial)) => { - if poll.logout.as_ref().is_some_and(|(pending, _)| *pending == serial) { - poll.logout = None; - } - poll.logout_cancelled.push(serial); - } + Ok(ControlUpdate::Logout(serial, output)) => poll.commands.push(ControlCommand::Logout(serial, output)), + Ok(ControlUpdate::ToggleGuide(output)) => poll.commands.push(ControlCommand::ToggleGuide(output)), + Ok(ControlUpdate::LogoutCancelled(serial)) => poll.commands.push(ControlCommand::CancelLogout(serial)), Ok(ControlUpdate::Config(source)) => poll.config = Some(source), Ok(ControlUpdate::Snapshot(snapshot)) => poll.snapshot = Some(snapshot), Ok(ControlUpdate::OverviewState(active)) => poll.overview_active = Some(active), @@ -190,13 +228,13 @@ fn control_connection() -> Result> { struct ControlState { config: Option, - sender: Sender, + sender: UpdateSender, pending: ShellSnapshot, serial: Option, } impl ControlState { - fn new(sender: Sender) -> Self { + fn new(sender: UpdateSender) -> Self { Self { sender, config: None, @@ -400,3 +438,35 @@ mod workspace_tests { assert_eq!(snapshot.workspaces_for_output(Some(99)).count(), 0); } } + +#[cfg(test)] +mod wake_tests { + use cosmic::iced::futures::{FutureExt, StreamExt}; + + use super::*; + + #[test] + fn control_updates_wake_without_a_clock_tick_even_before_subscription() { + let wake = Wake(Arc::new(tokio::sync::Notify::new())); + let (sender, receiver) = mpsc::channel(); + let sender = UpdateSender { + sender, + wake: wake.clone(), + }; + sender.send(ControlUpdate::OverviewState(true)).ok().unwrap(); + let stream = wake.stream(); + cosmic::iced::futures::pin_mut!(stream); + + assert_eq!(stream.next().now_or_never(), Some(Some(()))); + assert!(matches!(receiver.try_recv(), Ok(ControlUpdate::OverviewState(true)))); + assert_eq!(stream.next().now_or_never(), None); + + sender.send(ControlUpdate::Logout(7, "display".into())).ok().unwrap(); + sender.send(ControlUpdate::LogoutCancelled(7)).ok().unwrap(); + sender.send(ControlUpdate::Disconnected).ok().unwrap(); + assert_eq!(stream.next().now_or_never(), Some(Some(()))); + assert!(matches!(receiver.try_recv(), Ok(ControlUpdate::Logout(7, _)))); + assert!(matches!(receiver.try_recv(), Ok(ControlUpdate::LogoutCancelled(7)))); + assert!(matches!(receiver.try_recv(), Ok(ControlUpdate::Disconnected))); + } +} diff --git a/shell/ferese-shell/src/main.rs b/shell/ferese-shell/src/main.rs index ece77d8..4efd91a 100644 --- a/shell/ferese-shell/src/main.rs +++ b/shell/ferese-shell/src/main.rs @@ -235,6 +235,7 @@ enum Message { Event(Event, window::Id), NativeSurface(window::Id, Result<(Connection, wl_surface::WlSurface), String>), Tick, + ControlReady, ActivateWorkspace(u64), ToggleOverview, StatusUpdated(status::Update), @@ -340,6 +341,9 @@ impl cosmic::Application for FereseShell { fn subscription(&self) -> Subscription { Subscription::batch([ + self.control.as_ref().map_or_else(Subscription::none, |control| { + control.subscription().map(|_| Message::ControlReady) + }), ferese_theme::service::subscription().map(|snapshot| Message::ThemeChanged(Box::new(snapshot))), if self .system_modal @@ -764,6 +768,9 @@ impl cosmic::Application for FereseShell { .unwrap_or_default(); } + Task::none() + } + Message::ControlReady => { let mut reload_task = Task::none(); if let Some(control) = &self.control { let poll = control.poll(); @@ -810,17 +817,15 @@ impl cosmic::Application for FereseShell { } reload_task = Task::batch(tasks); } - for output in poll.guide_toggles { - reload_task = Task::batch([reload_task, self.toggle_guide(output)]); - } - for serial in poll.logout_cancelled { - reload_task = Task::batch([reload_task, self.cancel_logout_modal(serial)]); - } - if let Some((serial, output)) = poll.logout { - reload_task = Task::batch([ - reload_task, - self.open_system_modal(system_modal::PowerAction::Logout(serial), Some(&output)), - ]); + for command in poll.commands { + let task = match command { + control::ControlCommand::ToggleGuide(output) => self.toggle_guide(output), + control::ControlCommand::CancelLogout(serial) => self.cancel_logout_modal(serial), + control::ControlCommand::Logout(serial, output) => { + self.open_system_modal(system_modal::PowerAction::Logout(serial), Some(&output)) + } + }; + reload_task = Task::batch([reload_task, task]); } } reload_task