1#[cfg(feature = "data")]
2use crate::data::apply_menu_diffs;
3use crate::data::TrayItemMap;
4use crate::dbus::dbus_menu_proxy::{DBusMenuProxy, PropertiesUpdate};
5use crate::dbus::notifier_item_proxy::StatusNotifierItemProxy;
6use crate::dbus::notifier_watcher_proxy::StatusNotifierWatcherProxy;
7use crate::dbus::status_notifier_watcher::StatusNotifierWatcher;
8use crate::dbus::{self, OwnedValueExt};
9use crate::error::{Error, Result};
10use crate::item::{self, IconPixmap, Status, StatusNotifierItem, Tooltip};
11use crate::menu::{MenuDiff, TrayMenu};
12use crate::names;
13use dbus::DBusProps;
14use futures_lite::{Stream, StreamExt};
15use std::future::Future;
16use std::result;
17use std::sync::{Arc, Mutex};
18use std::time::{Duration, SystemTime, UNIX_EPOCH};
19use tokio::spawn;
20use tokio::sync::{broadcast, mpsc};
21use tokio::time::{sleep, timeout, Instant};
22use tracing::{debug, error, trace, warn};
23use zbus::fdo::{DBusProxy, PropertiesProxy};
24use zbus::names::InterfaceName;
25use zbus::zvariant::{Array, Structure, Value};
26use zbus::{Connection, Message};
27
28use self::names::ITEM_OBJECT;
29
30#[derive(Debug, Clone)]
34pub enum Event {
35 Add(String, Box<StatusNotifierItem>),
37 Update(String, UpdateEvent),
41 Remove(String),
43}
44
45#[derive(Debug, Clone)]
47pub enum UpdateEvent {
48 AttentionIcon(Option<String>),
49 Icon {
50 icon_name: Option<String>,
51 icon_pixmap: Option<Vec<IconPixmap>>,
52 },
53 OverlayIcon(Option<String>),
54 Status(Status),
55 Title(Option<String>),
56 Tooltip(Option<Tooltip>),
57 Menu(TrayMenu),
60 MenuDiff(Vec<MenuDiff>),
63 MenuConnect(String),
66}
67
68#[derive(Debug, Clone, Eq, PartialEq)]
71pub enum ActivateRequest {
72 MenuItem {
74 address: String,
75 menu_path: String,
76 submenu_id: i32,
77 },
78 Default { address: String, x: i32, y: i32 },
81 Secondary { address: String, x: i32, y: i32 },
84}
85
86const PROPERTIES_INTERFACE: &str = "org.kde.StatusNotifierItem";
87const ITEM_RETRY_DELAYS: [Duration; 3] = [
88 Duration::from_millis(50),
89 Duration::from_millis(200),
90 Duration::from_millis(750),
91];
92
93async fn retry_with_delays<T, E, I, F, Fut, N>(
94 delays: I,
95 mut operation: F,
96 mut on_retry: N,
97) -> std::result::Result<T, E>
98where
99 I: IntoIterator<Item = Duration>,
100 F: FnMut() -> Fut,
101 Fut: Future<Output = std::result::Result<T, E>>,
102 N: FnMut(&E, Duration),
103{
104 let mut delays = delays.into_iter();
105
106 loop {
107 match operation().await {
108 Ok(value) => return Ok(value),
109 Err(error) => match delays.next() {
110 Some(delay) => {
111 on_retry(&error, delay);
112 sleep(delay).await;
113 }
114 None => return Err(error),
115 },
116 }
117 }
118}
119
120async fn process_stream<S, T, E, F, Fut, N>(mut stream: S, mut operation: F, mut on_error: N)
121where
122 S: Stream<Item = T> + Unpin,
123 F: FnMut(T) -> Fut,
124 Fut: Future<Output = std::result::Result<(), E>>,
125 N: FnMut(E),
126{
127 while let Some(item) = stream.next().await {
128 if let Err(error) = operation(item).await {
129 on_error(error);
130 }
131 }
132}
133
134#[derive(Debug)]
136pub struct Client {
137 tx: broadcast::Sender<Event>,
138 _rx: broadcast::Receiver<Event>,
139 connection: Connection,
140
141 #[cfg(feature = "data")]
142 items: TrayItemMap,
143}
144
145impl Client {
146 pub async fn new() -> Result<Self> {
169 let connection = Connection::session().await?;
170 let (tx, rx) = broadcast::channel(32);
171
172 StatusNotifierWatcher::new().attach_to(&connection).await?;
174
175 let watcher_proxy = StatusNotifierWatcherProxy::new(&connection).await?;
177
178 let pid = std::process::id();
181 let mut i = 0;
182 let wellknown = loop {
183 use zbus::fdo::RequestNameReply::{AlreadyOwner, Exists, InQueue, PrimaryOwner};
184
185 i += 1;
186 let wellknown = format!("org.freedesktop.StatusNotifierHost-{pid}-{i}");
187 let wellknown: zbus::names::WellKnownName = wellknown
188 .try_into()
189 .expect("generated well-known name is invalid");
190
191 let flags = [zbus::fdo::RequestNameFlags::DoNotQueue];
192 match connection
193 .request_name_with_flags(&wellknown, flags.into_iter().collect())
194 .await?
195 {
196 PrimaryOwner => break wellknown,
197 Exists | AlreadyOwner => {}
198 InQueue => unreachable!(
199 "request_name_with_flags returned InQueue even though we specified DoNotQueue"
200 ),
201 };
202 };
203
204 debug!("wellknown: {wellknown}");
205 watcher_proxy
206 .register_status_notifier_host(&wellknown)
207 .await?;
208 let items = TrayItemMap::new();
209
210 {
212 let connection = connection.clone();
213 let tx = tx.clone();
214 let items = items.clone();
215
216 let stream = watcher_proxy
217 .receive_status_notifier_item_registered()
218 .await?;
219
220 spawn(async move {
221 process_stream(
222 stream,
223 |item| {
224 let connection = connection.clone();
225 let tx = tx.clone();
226 let items = items.clone();
227
228 async move {
229 let args = item.args().map_err(|error| (None, Error::from(error)))?;
230 let address = args.service;
231
232 debug!("received new item: {address}");
233 Self::handle_item(address, connection, tx, items)
234 .await
235 .map_err(|error| (Some(address.to_string()), error))
236 }
237 },
238 |(address, error)| match address {
239 Some(address) => {
240 error!("failed to initialize tray item {address}: {error}");
241 }
242 None => error!("failed to parse tray item registration: {error}"),
243 },
244 )
245 .await;
246
247 Ok::<(), Error>(())
248 });
249 }
250
251 {
255 let connection = connection.clone();
256 let tx = tx.clone();
257 let items = items.clone();
258
259 spawn(async move {
260 let initial_items = watcher_proxy.registered_status_notifier_items().await?;
261 debug!("initial items: {initial_items:?}");
262
263 for item in initial_items {
264 if let Err(err) =
265 Self::handle_item(&item, connection.clone(), tx.clone(), items.clone())
266 .await
267 {
268 error!("failed to initialize tray item {item}: {err}");
269 }
270 }
271
272 Ok::<(), Error>(())
273 });
274 }
275
276 {
279 let tx = tx.clone();
280 let items = items.clone();
281
282 let dbus_proxy = DBusProxy::new(&connection).await?;
283
284 let mut stream = dbus_proxy.receive_name_acquired().await?;
285
286 spawn(async move {
287 while let Some(thing) = stream.next().await {
288 let body = thing.args()?;
289 if body.name == names::WATCHER_BUS {
290 for dest in items.clear_items() {
291 tx.send(Event::Remove(dest))?;
292 }
293 }
294 }
295
296 Ok::<(), Error>(())
297 });
298 }
299
300 debug!("tray client initialized");
301
302 Ok(Self {
303 connection,
304 tx,
305 _rx: rx,
306 #[cfg(feature = "data")]
307 items,
308 })
309 }
310
311 async fn handle_item(
314 address: &str,
315 connection: Connection,
316 tx: broadcast::Sender<Event>,
317 items: TrayItemMap,
318 ) -> Result<()> {
319 let (destination, path) = parse_address(address);
320
321 let properties_proxy = PropertiesProxy::builder(&connection)
322 .destination(destination.to_string())?
323 .path(path.clone())?
324 .build()
325 .await?;
326
327 let properties = Self::get_item_properties(destination, &path, &properties_proxy).await?;
328
329 items.new_item(destination.into(), &properties);
330
331 tx.send(Event::Add(
332 destination.to_string(),
333 properties.clone().into(),
334 ))?;
335
336 {
337 let connection = connection.clone();
338 let destination = destination.to_string();
339 let items = items.clone();
340 let tx = tx.clone();
341
342 spawn(async move {
343 Self::watch_item_properties(
344 &destination,
345 &path,
346 &connection,
347 properties_proxy,
348 tx,
349 items,
350 )
351 .await?;
352
353 debug!("Stopped watching {destination}{path}");
354 Ok::<(), Error>(())
355 });
356 }
357
358 if let Some(menu) = properties.menu {
359 let destination = destination.to_string();
360
361 tx.send(Event::Update(
362 destination.clone(),
363 UpdateEvent::MenuConnect(menu.clone()),
364 ))?;
365
366 spawn(async move {
367 Self::watch_menu(destination, &menu, &connection, tx, items).await?;
368 Ok::<(), Error>(())
369 });
370 }
371
372 Ok(())
373 }
374
375 async fn get_item_properties(
377 destination: &str,
378 path: &str,
379 properties_proxy: &PropertiesProxy<'_>,
380 ) -> Result<StatusNotifierItem> {
381 let properties = retry_with_delays(
382 ITEM_RETRY_DELAYS,
383 || {
384 properties_proxy.get_all(
385 InterfaceName::from_static_str(PROPERTIES_INTERFACE)
386 .expect("to be valid interface name"),
387 )
388 },
389 |error, delay| {
390 warn!(
391 "failed to read tray item {destination}{path}: {error}; retrying in {} ms",
392 delay.as_millis()
393 );
394 },
395 )
396 .await?;
397
398 StatusNotifierItem::try_from(DBusProps(properties))
399 }
400
401 async fn watch_item_properties(
404 destination: &str,
405 path: &str,
406 connection: &Connection,
407 properties_proxy: PropertiesProxy<'_>,
408 tx: broadcast::Sender<Event>,
409 items: TrayItemMap,
410 ) -> Result<()> {
411 let notifier_item_proxy = StatusNotifierItemProxy::builder(connection)
412 .destination(destination)?
413 .path(path)?
414 .build()
415 .await?;
416
417 let dbus_proxy = DBusProxy::new(connection).await?;
418
419 let mut disconnect_stream = dbus_proxy.receive_name_owner_changed().await?;
420 let mut props_changed = notifier_item_proxy.inner().receive_all_signals().await?;
421
422 loop {
423 tokio::select! {
424 Some(change) = props_changed.next() => {
425 match Self::get_update_event(change, &properties_proxy).await {
426 Ok(Some(event)) => {
427 debug!("[{destination}{path}] received property change: {event:?}");
428
429 cfg_if::cfg_if! {
430 if #[cfg(feature = "data")] {
431 items.apply_update_event(destination, &event);
432 }
433 }
434
435 tx.send(Event::Update(destination.to_string(), event))?;
436 }
437 Err(e) => {
438 error!("Error parsing update properties from {destination}{path}: {e:?}");
439 }
440 _ => {}
441 }
442 }
443 Some(signal) = disconnect_stream.next() => {
444 let args = signal.args()?;
445 let old = args.old_owner();
446 let new = args.new_owner();
447
448 if let (Some(old), None) = (old.as_ref(), new.as_ref()) {
449 if old == destination {
450 debug!("[{destination}{path}] disconnected");
451
452 let watcher_proxy = StatusNotifierWatcherProxy::new(connection)
453 .await
454 .expect("Failed to open StatusNotifierWatcherProxy");
455
456 if let Err(error) = watcher_proxy.unregister_status_notifier_item(old).await {
457 error!("{error:?}");
458 }
459
460
461 items.remove_item(destination);
462
463 tx.send(Event::Remove(destination.to_string()))?;
464 break Ok(());
465 }
466 }
467 }
468 }
469 }
470 }
471
472 async fn get_update_event(
474 change: Message,
475 properties_proxy: &PropertiesProxy<'_>,
476 ) -> Result<Option<UpdateEvent>> {
477 use UpdateEvent::{AttentionIcon, Icon, OverlayIcon, Status, Title, Tooltip};
478
479 let header = change.header();
480 let member = header
481 .member()
482 .ok_or(Error::InvalidData("Update message header missing `member`"))?;
483
484 macro_rules! get_property {
485 ($name:expr) => {
486 match properties_proxy
487 .get(
488 InterfaceName::from_static_str(PROPERTIES_INTERFACE)
489 .expect("to be valid interface name"),
490 $name,
491 )
492 .await
493 {
494 Ok(v) => Ok(Some(v)),
495 Err(e) => match e {
496 zbus::fdo::Error::InvalidArgs(_) => {
498 warn!("{e}");
499 Ok(None)
500 }
501 _ => Err(Into::<Error>::into(e)),
502 },
503 }
504 };
505 }
506
507 let property = match member.as_str() {
508 "NewAttentionIcon" => Some(AttentionIcon(
509 get_property!("AttentionIconName")?
510 .as_ref()
511 .map(OwnedValueExt::to_string)
512 .transpose()?,
513 )),
514 "NewIcon" => {
515 let icon_name = get_property!("IconName")
516 .unwrap_or_else(|e| {
517 warn!("Error getting IconName: {e:?}");
518 None
519 })
520 .as_ref()
521 .map(OwnedValueExt::to_string)
522 .transpose()
523 .ok()
524 .flatten();
525
526 let icon_pixmap = get_property!("IconPixmap")
527 .unwrap_or_else(|e| {
528 warn!("Error getting IconPixmap: {e:?}");
529 None
530 })
531 .as_deref()
532 .map(Value::downcast_ref::<&Array>)
533 .and_then(result::Result::ok)
534 .map(IconPixmap::from_array)
535 .and_then(result::Result::ok);
536
537 Some(Icon {
538 icon_name,
539 icon_pixmap,
540 })
541 }
542 "NewOverlayIcon" => Some(OverlayIcon(
543 get_property!("OverlayIconName")?
544 .as_ref()
545 .map(OwnedValueExt::to_string)
546 .transpose()?,
547 )),
548 "NewStatus" => Some(Status(
549 get_property!("Status")?
550 .as_deref()
551 .map(Value::downcast_ref::<&str>)
552 .transpose()?
553 .map(item::Status::from)
554 .unwrap_or_default(), )),
556 "NewTitle" => Some(Title(
557 get_property!("Title")?
558 .as_ref()
559 .map(OwnedValueExt::to_string)
560 .transpose()?,
561 )),
562 "NewToolTip" => Some(Tooltip(
563 get_property!("ToolTip")?
564 .as_deref()
565 .map(Value::downcast_ref::<&Structure>)
566 .transpose()?
567 .map(crate::item::Tooltip::try_from)
568 .transpose()?,
569 )),
570 _ => {
571 warn!("received unhandled update event: {member}");
572 None
573 }
574 };
575
576 debug!("received tray item update: {member} -> {property:?}");
577
578 Ok(property)
579 }
580
581 async fn watch_menu(
587 destination: String,
588 menu_path: &str,
589 connection: &Connection,
590 tx: broadcast::Sender<Event>,
591 items: TrayItemMap,
592 ) -> Result<()> {
593 const LAYOUT_UPDATE_INTERVAL_MS: Duration = Duration::from_millis(50);
594
595 let dbus_menu_proxy = DBusMenuProxy::builder(connection)
596 .destination(destination.as_str())?
597 .path(menu_path)?
598 .build()
599 .await?;
600
601 debug!("[{destination}{menu_path}] getting initial menu");
602 let menu = dbus_menu_proxy.get_layout(0, -1, &[]).await?;
603 let menu = TrayMenu::try_from(menu)?;
604
605 items.update_menu(&destination, &menu);
606
607 tx.send(Event::Update(destination.clone(), UpdateEvent::Menu(menu)))?;
608
609 let mut layout_updated = dbus_menu_proxy.receive_layout_updated().await?;
610 let mut properties_updated = dbus_menu_proxy.receive_items_properties_updated().await?;
611
612 let last_layout_update = Arc::new(Mutex::new(Instant::now()));
613 let (layout_tx, mut layout_rx) = mpsc::channel(4);
614
615 loop {
616 tokio::select!(
617 Some(ev) = layout_updated.next() => {
618 trace!("received layout update");
619
620 let now = Instant::now();
621 *last_layout_update.lock().expect("should get lock") = now;
622
623 let args = ev.args()?;
624
625 let last_layout_update = last_layout_update.clone();
626 let layout_tx = layout_tx.clone();
627 spawn(async move {
628 sleep(LAYOUT_UPDATE_INTERVAL_MS).await;
629 if *last_layout_update.lock().expect("should get lock") == now {
630 trace!("dispatching layout update");
631 layout_tx.send(args.parent).await.expect("should send");
632 }
633 });
634 }
635 Some(layout_parent) = layout_rx.recv() => {
636 debug!("[{destination}{menu_path}] layout update");
637
638 let get_layout = dbus_menu_proxy.get_layout(layout_parent, -1, &[]);
639
640 let menu = match timeout(Duration::from_secs(1), get_layout).await {
641 Ok(Ok(menu)) => {
642 debug!("got new menu layout");
643 menu
644 }
645 Ok(Err(err)) => {
646 error!("error fetching layout: {err:?}");
647 break;
648 }
649 Err(_) => {
650 error!("Timeout getting layout");
651 break;
652 }
653 };
654
655 let menu = TrayMenu::try_from(menu)?;
656
657 items.update_menu(&destination, &menu);
658
659 debug!("sending new menu for '{destination}'");
660 trace!("new menu for '{destination}': {menu:?}");
661 tx.send(Event::Update(
662 destination.clone(),
663 UpdateEvent::Menu(menu),
664 ))?;
665 }
666 Some(change) = properties_updated.next() => {
667 let body = change.message().body();
668 let update: PropertiesUpdate= body.deserialize::<PropertiesUpdate>()?;
669 let diffs = Vec::try_from(update)?;
670
671 #[cfg(feature = "data")]
672 if let Some((_, Some(menu))) = items
673 .get_map()
674 .lock()
675 .expect("mutex lock should succeed")
676 .get_mut(&destination)
677 {
678 apply_menu_diffs(menu, &diffs);
679 } else {
680 error!("could not find item in state");
681 }
682
683 tx.send(Event::Update(
684 destination.clone(),
685 UpdateEvent::MenuDiff(diffs),
686 ))?;
687
688 }
690 );
691 }
692
693 Ok(())
694 }
695
696 async fn get_notifier_item_proxy(
697 &self,
698 address: String,
699 ) -> Result<StatusNotifierItemProxy<'_>> {
700 let proxy = StatusNotifierItemProxy::builder(&self.connection)
701 .destination(address)?
702 .path(ITEM_OBJECT)?
703 .build()
704 .await?;
705 Ok(proxy)
706 }
707
708 async fn get_menu_proxy(
709 &self,
710 address: String,
711 menu_path: String,
712 ) -> Result<DBusMenuProxy<'_>> {
713 let proxy = DBusMenuProxy::builder(&self.connection)
714 .destination(address)?
715 .path(menu_path)?
716 .build()
717 .await?;
718
719 Ok(proxy)
720 }
721
722 #[must_use]
727 pub fn subscribe(&self) -> broadcast::Receiver<Event> {
728 self.tx.subscribe()
729 }
730
731 #[cfg(feature = "data")]
733 #[must_use]
734 pub fn items(&self) -> std::sync::Arc<std::sync::Mutex<crate::data::BaseMap>> {
735 self.items.get_map()
736 }
737
738 pub async fn about_to_show_menuitem(
747 &self,
748 address: String,
749 menu_path: String,
750 id: i32,
751 ) -> Result<bool> {
752 let proxy = self.get_menu_proxy(address, menu_path).await?;
753 Ok(proxy.about_to_show(id).await?)
754 }
755
756 pub async fn activate(&self, req: ActivateRequest) -> Result<()> {
767 macro_rules! timeout_event {
768 ($event:expr) => {
769 if timeout(Duration::from_secs(1), $event).await.is_err() {
770 error!("Timed out sending activate event");
771 }
772 };
773 }
774 match req {
775 ActivateRequest::MenuItem {
776 address,
777 menu_path,
778 submenu_id,
779 } => {
780 let proxy = self.get_menu_proxy(address, menu_path).await?;
781 let timestamp = SystemTime::now()
782 .duration_since(UNIX_EPOCH)
783 .expect("time should flow forwards");
784
785 let event = proxy.event(
786 submenu_id,
787 "clicked",
788 &Value::I32(0),
789 timestamp.as_secs() as u32,
790 );
791
792 timeout_event!(event);
793 }
794 ActivateRequest::Default { address, x, y } => {
795 let proxy = self.get_notifier_item_proxy(address).await?;
796 let event = proxy.activate(x, y);
797
798 timeout_event!(event);
799 }
800 ActivateRequest::Secondary { address, x, y } => {
801 let proxy = self.get_notifier_item_proxy(address).await?;
802 let event = proxy.secondary_activate(x, y);
803
804 timeout_event!(event);
805 }
806 }
807
808 Ok(())
809 }
810}
811
812fn parse_address(address: &str) -> (&str, String) {
813 address
814 .split_once('/')
815 .map_or((address, String::from("/StatusNotifierItem")), |(d, p)| {
816 (d, format!("/{p}"))
817 })
818}
819
820#[cfg(test)]
821mod tests {
822 use super::*;
823
824 #[tokio::test]
825 async fn item_initialization_retries_transient_failures() {
826 use std::{cell::Cell, future::ready};
827
828 let attempts = Cell::new(0);
829 let result = retry_with_delays(
830 [Duration::ZERO; 2],
831 || {
832 let attempt = attempts.get() + 1;
833 attempts.set(attempt);
834 ready((attempt == 3).then_some(attempt).ok_or(attempt))
835 },
836 |_, _| {},
837 )
838 .await;
839
840 assert_eq!(result, Ok(3));
841 assert_eq!(attempts.get(), 3);
842 }
843
844 #[tokio::test]
845 async fn item_initialization_returns_last_failure() {
846 use std::{cell::Cell, future::ready};
847
848 let attempts = Cell::new(0);
849 let result = retry_with_delays(
850 [Duration::ZERO; 2],
851 || {
852 let attempt = attempts.get() + 1;
853 attempts.set(attempt);
854 ready(Err::<(), _>(attempt))
855 },
856 |_, _| {},
857 )
858 .await;
859
860 assert_eq!(result, Err(3));
861 assert_eq!(attempts.get(), 3);
862 }
863
864 #[tokio::test]
865 async fn registration_stream_continues_after_item_failure() {
866 use std::{cell::RefCell, future::ready};
867
868 let handled = RefCell::new(Vec::new());
869 let stream = futures_lite::stream::iter([1, 2]);
870
871 process_stream(
872 stream,
873 |item| {
874 handled.borrow_mut().push(item);
875 ready(if item == 1 { Err("broken") } else { Ok(()) })
876 },
877 |_| {},
878 )
879 .await;
880
881 assert_eq!(*handled.borrow(), [1, 2]);
882 }
883
884 #[test]
885 fn parse_unnamed() {
886 let address = ":1.58/StatusNotifierItem";
887 let (destination, path) = parse_address(address);
888
889 assert_eq!(":1.58", destination);
890 assert_eq!("/StatusNotifierItem", path);
891 }
892
893 #[test]
894 fn parse_named() {
895 let address = ":1.72/org/ayatana/NotificationItem/dropbox_client_1398";
896 let (destination, path) = parse_address(address);
897
898 assert_eq!(":1.72", destination);
899 assert_eq!("/org/ayatana/NotificationItem/dropbox_client_1398", path);
900 }
901}