#[cfg(feature = "data")]
use crate::data::apply_menu_diffs;
use crate::data::TrayItemMap;
use crate::dbus::dbus_menu_proxy::{DBusMenuProxy, PropertiesUpdate};
use crate::dbus::notifier_item_proxy::StatusNotifierItemProxy;
use crate::dbus::notifier_watcher_proxy::StatusNotifierWatcherProxy;
use crate::dbus::status_notifier_watcher::StatusNotifierWatcher;
use crate::dbus::{self, OwnedValueExt};
use crate::error::{Error, Result};
use crate::item::{self, IconPixmap, Status, StatusNotifierItem, Tooltip};
use crate::menu::{MenuDiff, TrayMenu};
use crate::names;
use dbus::DBusProps;
use futures_lite::{Stream, StreamExt};
use std::future::Future;
use std::result;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::spawn;
use tokio::sync::{broadcast, mpsc};
use tokio::time::{sleep, timeout, Instant};
use tracing::{debug, error, trace, warn};
use zbus::fdo::{DBusProxy, PropertiesProxy};
use zbus::names::InterfaceName;
use zbus::zvariant::{Array, Structure, Value};
use zbus::{Connection, Message};
use self::names::ITEM_OBJECT;
#[derive(Debug, Clone)]
pub enum Event {
Add(String, Box<StatusNotifierItem>),
Update(String, UpdateEvent),
Remove(String),
}
#[derive(Debug, Clone)]
pub enum UpdateEvent {
AttentionIcon(Option<String>),
Icon {
icon_name: Option<String>,
icon_pixmap: Option<Vec<IconPixmap>>,
},
OverlayIcon(Option<String>),
Status(Status),
Title(Option<String>),
Tooltip(Option<Tooltip>),
Menu(TrayMenu),
MenuDiff(Vec<MenuDiff>),
MenuConnect(String),
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum ActivateRequest {
MenuItem {
address: String,
menu_path: String,
submenu_id: i32,
},
Default { address: String, x: i32, y: i32 },
Secondary { address: String, x: i32, y: i32 },
}
const PROPERTIES_INTERFACE: &str = "org.kde.StatusNotifierItem";
const ITEM_RETRY_DELAYS: [Duration; 3] = [
Duration::from_millis(50),
Duration::from_millis(200),
Duration::from_millis(750),
];
async fn retry_with_delays<T, E, I, F, Fut, N>(
delays: I,
mut operation: F,
mut on_retry: N,
) -> std::result::Result<T, E>
where
I: IntoIterator<Item = Duration>,
F: FnMut() -> Fut,
Fut: Future<Output = std::result::Result<T, E>>,
N: FnMut(&E, Duration),
{
let mut delays = delays.into_iter();
loop {
match operation().await {
Ok(value) => return Ok(value),
Err(error) => match delays.next() {
Some(delay) => {
on_retry(&error, delay);
sleep(delay).await;
}
None => return Err(error),
},
}
}
}
async fn process_stream<S, T, E, F, Fut, N>(mut stream: S, mut operation: F, mut on_error: N)
where
S: Stream<Item = T> + Unpin,
F: FnMut(T) -> Fut,
Fut: Future<Output = std::result::Result<(), E>>,
N: FnMut(E),
{
while let Some(item) = stream.next().await {
if let Err(error) = operation(item).await {
on_error(error);
}
}
}
#[derive(Debug)]
pub struct Client {
tx: broadcast::Sender<Event>,
_rx: broadcast::Receiver<Event>,
connection: Connection,
#[cfg(feature = "data")]
items: TrayItemMap,
}
impl Client {
pub async fn new() -> Result<Self> {
let connection = Connection::session().await?;
let (tx, rx) = broadcast::channel(32);
StatusNotifierWatcher::new().attach_to(&connection).await?;
let watcher_proxy = StatusNotifierWatcherProxy::new(&connection).await?;
let pid = std::process::id();
let mut i = 0;
let wellknown = loop {
use zbus::fdo::RequestNameReply::{AlreadyOwner, Exists, InQueue, PrimaryOwner};
i += 1;
let wellknown = format!("org.freedesktop.StatusNotifierHost-{pid}-{i}");
let wellknown: zbus::names::WellKnownName = wellknown
.try_into()
.expect("generated well-known name is invalid");
let flags = [zbus::fdo::RequestNameFlags::DoNotQueue];
match connection
.request_name_with_flags(&wellknown, flags.into_iter().collect())
.await?
{
PrimaryOwner => break wellknown,
Exists | AlreadyOwner => {}
InQueue => unreachable!(
"request_name_with_flags returned InQueue even though we specified DoNotQueue"
),
};
};
debug!("wellknown: {wellknown}");
watcher_proxy
.register_status_notifier_host(&wellknown)
.await?;
let items = TrayItemMap::new();
{
let connection = connection.clone();
let tx = tx.clone();
let items = items.clone();
let stream = watcher_proxy
.receive_status_notifier_item_registered()
.await?;
spawn(async move {
process_stream(
stream,
|item| {
let connection = connection.clone();
let tx = tx.clone();
let items = items.clone();
async move {
let args = item.args().map_err(|error| (None, Error::from(error)))?;
let address = args.service;
debug!("received new item: {address}");
Self::handle_item(address, connection, tx, items)
.await
.map_err(|error| (Some(address.to_string()), error))
}
},
|(address, error)| match address {
Some(address) => {
error!("failed to initialize tray item {address}: {error}");
}
None => error!("failed to parse tray item registration: {error}"),
},
)
.await;
Ok::<(), Error>(())
});
}
{
let connection = connection.clone();
let tx = tx.clone();
let items = items.clone();
spawn(async move {
let initial_items = watcher_proxy.registered_status_notifier_items().await?;
debug!("initial items: {initial_items:?}");
for item in initial_items {
if let Err(err) =
Self::handle_item(&item, connection.clone(), tx.clone(), items.clone())
.await
{
error!("failed to initialize tray item {item}: {err}");
}
}
Ok::<(), Error>(())
});
}
{
let tx = tx.clone();
let items = items.clone();
let dbus_proxy = DBusProxy::new(&connection).await?;
let mut stream = dbus_proxy.receive_name_acquired().await?;
spawn(async move {
while let Some(thing) = stream.next().await {
let body = thing.args()?;
if body.name == names::WATCHER_BUS {
for dest in items.clear_items() {
tx.send(Event::Remove(dest))?;
}
}
}
Ok::<(), Error>(())
});
}
debug!("tray client initialized");
Ok(Self {
connection,
tx,
_rx: rx,
#[cfg(feature = "data")]
items,
})
}
async fn handle_item(
address: &str,
connection: Connection,
tx: broadcast::Sender<Event>,
items: TrayItemMap,
) -> Result<()> {
let (destination, path) = parse_address(address);
let properties_proxy = PropertiesProxy::builder(&connection)
.destination(destination.to_string())?
.path(path.clone())?
.build()
.await?;
let properties = Self::get_item_properties(destination, &path, &properties_proxy).await?;
items.new_item(destination.into(), &properties);
tx.send(Event::Add(
destination.to_string(),
properties.clone().into(),
))?;
{
let connection = connection.clone();
let destination = destination.to_string();
let items = items.clone();
let tx = tx.clone();
spawn(async move {
Self::watch_item_properties(
&destination,
&path,
&connection,
properties_proxy,
tx,
items,
)
.await?;
debug!("Stopped watching {destination}{path}");
Ok::<(), Error>(())
});
}
if let Some(menu) = properties.menu {
let destination = destination.to_string();
tx.send(Event::Update(
destination.clone(),
UpdateEvent::MenuConnect(menu.clone()),
))?;
spawn(async move {
Self::watch_menu(destination, &menu, &connection, tx, items).await?;
Ok::<(), Error>(())
});
}
Ok(())
}
async fn get_item_properties(
destination: &str,
path: &str,
properties_proxy: &PropertiesProxy<'_>,
) -> Result<StatusNotifierItem> {
let properties = retry_with_delays(
ITEM_RETRY_DELAYS,
|| {
properties_proxy.get_all(
InterfaceName::from_static_str(PROPERTIES_INTERFACE)
.expect("to be valid interface name"),
)
},
|error, delay| {
warn!(
"failed to read tray item {destination}{path}: {error}; retrying in {} ms",
delay.as_millis()
);
},
)
.await?;
StatusNotifierItem::try_from(DBusProps(properties))
}
async fn watch_item_properties(
destination: &str,
path: &str,
connection: &Connection,
properties_proxy: PropertiesProxy<'_>,
tx: broadcast::Sender<Event>,
items: TrayItemMap,
) -> Result<()> {
let notifier_item_proxy = StatusNotifierItemProxy::builder(connection)
.destination(destination)?
.path(path)?
.build()
.await?;
let dbus_proxy = DBusProxy::new(connection).await?;
let mut disconnect_stream = dbus_proxy.receive_name_owner_changed().await?;
let mut props_changed = notifier_item_proxy.inner().receive_all_signals().await?;
loop {
tokio::select! {
Some(change) = props_changed.next() => {
match Self::get_update_event(change, &properties_proxy).await {
Ok(Some(event)) => {
debug!("[{destination}{path}] received property change: {event:?}");
cfg_if::cfg_if! {
if #[cfg(feature = "data")] {
items.apply_update_event(destination, &event);
}
}
tx.send(Event::Update(destination.to_string(), event))?;
}
Err(e) => {
error!("Error parsing update properties from {destination}{path}: {e:?}");
}
_ => {}
}
}
Some(signal) = disconnect_stream.next() => {
let args = signal.args()?;
let old = args.old_owner();
let new = args.new_owner();
if let (Some(old), None) = (old.as_ref(), new.as_ref()) {
if old == destination {
debug!("[{destination}{path}] disconnected");
let watcher_proxy = StatusNotifierWatcherProxy::new(connection)
.await
.expect("Failed to open StatusNotifierWatcherProxy");
if let Err(error) = watcher_proxy.unregister_status_notifier_item(old).await {
error!("{error:?}");
}
items.remove_item(destination);
tx.send(Event::Remove(destination.to_string()))?;
break Ok(());
}
}
}
}
}
}
async fn get_update_event(
change: Message,
properties_proxy: &PropertiesProxy<'_>,
) -> Result<Option<UpdateEvent>> {
use UpdateEvent::{AttentionIcon, Icon, OverlayIcon, Status, Title, Tooltip};
let header = change.header();
let member = header
.member()
.ok_or(Error::InvalidData("Update message header missing `member`"))?;
macro_rules! get_property {
($name:expr) => {
match properties_proxy
.get(
InterfaceName::from_static_str(PROPERTIES_INTERFACE)
.expect("to be valid interface name"),
$name,
)
.await
{
Ok(v) => Ok(Some(v)),
Err(e) => match e {
zbus::fdo::Error::InvalidArgs(_) => {
warn!("{e}");
Ok(None)
}
_ => Err(Into::<Error>::into(e)),
},
}
};
}
let property = match member.as_str() {
"NewAttentionIcon" => Some(AttentionIcon(
get_property!("AttentionIconName")?
.as_ref()
.map(OwnedValueExt::to_string)
.transpose()?,
)),
"NewIcon" => {
let icon_name = get_property!("IconName")
.unwrap_or_else(|e| {
warn!("Error getting IconName: {e:?}");
None
})
.as_ref()
.map(OwnedValueExt::to_string)
.transpose()
.ok()
.flatten();
let icon_pixmap = get_property!("IconPixmap")
.unwrap_or_else(|e| {
warn!("Error getting IconPixmap: {e:?}");
None
})
.as_deref()
.map(Value::downcast_ref::<&Array>)
.and_then(result::Result::ok)
.map(IconPixmap::from_array)
.and_then(result::Result::ok);
Some(Icon {
icon_name,
icon_pixmap,
})
}
"NewOverlayIcon" => Some(OverlayIcon(
get_property!("OverlayIconName")?
.as_ref()
.map(OwnedValueExt::to_string)
.transpose()?,
)),
"NewStatus" => Some(Status(
get_property!("Status")?
.as_deref()
.map(Value::downcast_ref::<&str>)
.transpose()?
.map(item::Status::from)
.unwrap_or_default(), )),
"NewTitle" => Some(Title(
get_property!("Title")?
.as_ref()
.map(OwnedValueExt::to_string)
.transpose()?,
)),
"NewToolTip" => Some(Tooltip(
get_property!("ToolTip")?
.as_deref()
.map(Value::downcast_ref::<&Structure>)
.transpose()?
.map(crate::item::Tooltip::try_from)
.transpose()?,
)),
_ => {
warn!("received unhandled update event: {member}");
None
}
};
debug!("received tray item update: {member} -> {property:?}");
Ok(property)
}
async fn watch_menu(
destination: String,
menu_path: &str,
connection: &Connection,
tx: broadcast::Sender<Event>,
items: TrayItemMap,
) -> Result<()> {
const LAYOUT_UPDATE_INTERVAL_MS: Duration = Duration::from_millis(50);
let dbus_menu_proxy = DBusMenuProxy::builder(connection)
.destination(destination.as_str())?
.path(menu_path)?
.build()
.await?;
debug!("[{destination}{menu_path}] getting initial menu");
let menu = dbus_menu_proxy.get_layout(0, -1, &[]).await?;
let menu = TrayMenu::try_from(menu)?;
items.update_menu(&destination, &menu);
tx.send(Event::Update(destination.clone(), UpdateEvent::Menu(menu)))?;
let mut layout_updated = dbus_menu_proxy.receive_layout_updated().await?;
let mut properties_updated = dbus_menu_proxy.receive_items_properties_updated().await?;
let last_layout_update = Arc::new(Mutex::new(Instant::now()));
let (layout_tx, mut layout_rx) = mpsc::channel(4);
loop {
tokio::select!(
Some(ev) = layout_updated.next() => {
trace!("received layout update");
let now = Instant::now();
*last_layout_update.lock().expect("should get lock") = now;
let args = ev.args()?;
let last_layout_update = last_layout_update.clone();
let layout_tx = layout_tx.clone();
spawn(async move {
sleep(LAYOUT_UPDATE_INTERVAL_MS).await;
if *last_layout_update.lock().expect("should get lock") == now {
trace!("dispatching layout update");
layout_tx.send(args.parent).await.expect("should send");
}
});
}
Some(layout_parent) = layout_rx.recv() => {
debug!("[{destination}{menu_path}] layout update");
let get_layout = dbus_menu_proxy.get_layout(layout_parent, -1, &[]);
let menu = match timeout(Duration::from_secs(1), get_layout).await {
Ok(Ok(menu)) => {
debug!("got new menu layout");
menu
}
Ok(Err(err)) => {
error!("error fetching layout: {err:?}");
break;
}
Err(_) => {
error!("Timeout getting layout");
break;
}
};
let menu = TrayMenu::try_from(menu)?;
items.update_menu(&destination, &menu);
debug!("sending new menu for '{destination}'");
trace!("new menu for '{destination}': {menu:?}");
tx.send(Event::Update(
destination.clone(),
UpdateEvent::Menu(menu),
))?;
}
Some(change) = properties_updated.next() => {
let body = change.message().body();
let update: PropertiesUpdate= body.deserialize::<PropertiesUpdate>()?;
let diffs = Vec::try_from(update)?;
#[cfg(feature = "data")]
if let Some((_, Some(menu))) = items
.get_map()
.lock()
.expect("mutex lock should succeed")
.get_mut(&destination)
{
apply_menu_diffs(menu, &diffs);
} else {
error!("could not find item in state");
}
tx.send(Event::Update(
destination.clone(),
UpdateEvent::MenuDiff(diffs),
))?;
}
);
}
Ok(())
}
async fn get_notifier_item_proxy(
&self,
address: String,
) -> Result<StatusNotifierItemProxy<'_>> {
let proxy = StatusNotifierItemProxy::builder(&self.connection)
.destination(address)?
.path(ITEM_OBJECT)?
.build()
.await?;
Ok(proxy)
}
async fn get_menu_proxy(
&self,
address: String,
menu_path: String,
) -> Result<DBusMenuProxy<'_>> {
let proxy = DBusMenuProxy::builder(&self.connection)
.destination(address)?
.path(menu_path)?
.build()
.await?;
Ok(proxy)
}
#[must_use]
pub fn subscribe(&self) -> broadcast::Receiver<Event> {
self.tx.subscribe()
}
#[cfg(feature = "data")]
#[must_use]
pub fn items(&self) -> std::sync::Arc<std::sync::Mutex<crate::data::BaseMap>> {
self.items.get_map()
}
pub async fn about_to_show_menuitem(
&self,
address: String,
menu_path: String,
id: i32,
) -> Result<bool> {
let proxy = self.get_menu_proxy(address, menu_path).await?;
Ok(proxy.about_to_show(id).await?)
}
pub async fn activate(&self, req: ActivateRequest) -> Result<()> {
macro_rules! timeout_event {
($event:expr) => {
if timeout(Duration::from_secs(1), $event).await.is_err() {
error!("Timed out sending activate event");
}
};
}
match req {
ActivateRequest::MenuItem {
address,
menu_path,
submenu_id,
} => {
let proxy = self.get_menu_proxy(address, menu_path).await?;
let timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("time should flow forwards");
let event = proxy.event(
submenu_id,
"clicked",
&Value::I32(0),
timestamp.as_secs() as u32,
);
timeout_event!(event);
}
ActivateRequest::Default { address, x, y } => {
let proxy = self.get_notifier_item_proxy(address).await?;
let event = proxy.activate(x, y);
timeout_event!(event);
}
ActivateRequest::Secondary { address, x, y } => {
let proxy = self.get_notifier_item_proxy(address).await?;
let event = proxy.secondary_activate(x, y);
timeout_event!(event);
}
}
Ok(())
}
}
fn parse_address(address: &str) -> (&str, String) {
address
.split_once('/')
.map_or((address, String::from("/StatusNotifierItem")), |(d, p)| {
(d, format!("/{p}"))
})
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn item_initialization_retries_transient_failures() {
use std::{cell::Cell, future::ready};
let attempts = Cell::new(0);
let result = retry_with_delays(
[Duration::ZERO; 2],
|| {
let attempt = attempts.get() + 1;
attempts.set(attempt);
ready((attempt == 3).then_some(attempt).ok_or(attempt))
},
|_, _| {},
)
.await;
assert_eq!(result, Ok(3));
assert_eq!(attempts.get(), 3);
}
#[tokio::test]
async fn item_initialization_returns_last_failure() {
use std::{cell::Cell, future::ready};
let attempts = Cell::new(0);
let result = retry_with_delays(
[Duration::ZERO; 2],
|| {
let attempt = attempts.get() + 1;
attempts.set(attempt);
ready(Err::<(), _>(attempt))
},
|_, _| {},
)
.await;
assert_eq!(result, Err(3));
assert_eq!(attempts.get(), 3);
}
#[tokio::test]
async fn registration_stream_continues_after_item_failure() {
use std::{cell::RefCell, future::ready};
let handled = RefCell::new(Vec::new());
let stream = futures_lite::stream::iter([1, 2]);
process_stream(
stream,
|item| {
handled.borrow_mut().push(item);
ready(if item == 1 { Err("broken") } else { Ok(()) })
},
|_| {},
)
.await;
assert_eq!(*handled.borrow(), [1, 2]);
}
#[test]
fn parse_unnamed() {
let address = ":1.58/StatusNotifierItem";
let (destination, path) = parse_address(address);
assert_eq!(":1.58", destination);
assert_eq!("/StatusNotifierItem", path);
}
#[test]
fn parse_named() {
let address = ":1.72/org/ayatana/NotificationItem/dropbox_client_1398";
let (destination, path) = parse_address(address);
assert_eq!(":1.72", destination);
assert_eq!("/org/ayatana/NotificationItem/dropbox_client_1398", path);
}
}