nvrl/fluxo-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 13 Wiki Activity

poll abstraction

Browse Source
This commit is contained in:
Nils Pukropp
2026-04-04 18:37:27 +02:00
parent 57906de920
commit f1e601f9ed
2 changed files with 135 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/daemon.rs
+87 -62
View File
@@ -31,6 +31,64 @@ use tokio::time::{Duration, sleep};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info};
/// Spawn a health-tracked polling loop. `$poll_expr` is awaited each cycle.
macro_rules! spawn_poll_loop {
($name:expr, $interval:expr, $health:expr, $token:expr, $poll_expr:expr) => {
tokio::spawn(async move {
info!(concat!("Starting ", $name, " polling task"));
loop {
if !crate::health::is_poll_in_backoff($name, &$health).await {
let res: crate::error::Result<()> = $poll_expr.await;
crate::health::handle_poll_result($name, res, &$health).await;
}
tokio::select! {
_ = $token.cancelled() => break,
_ = sleep($interval) => {}
}
}
info!(concat!($name, " task shut down."));
})
};
}
/// Spawn a health-tracked polling loop with an extra trigger channel for early wake-up.
macro_rules! spawn_poll_loop_triggered {
($name:expr, $interval:expr, $health:expr, $token:expr, $trigger:expr, $poll_expr:expr) => {
tokio::spawn(async move {
info!(concat!("Starting ", $name, " polling task"));
loop {
if !crate::health::is_poll_in_backoff($name, &$health).await {
let res: crate::error::Result<()> = $poll_expr.await;
crate::health::handle_poll_result($name, res, &$health).await;
}
tokio::select! {
_ = $token.cancelled() => break,
_ = $trigger.recv() => {},
_ = sleep($interval) => {}
}
}
info!(concat!($name, " task shut down."));
})
};
}
/// Spawn a simple polling loop without health tracking.
macro_rules! spawn_poll_loop_simple {
($name:expr, $interval:expr, $token:expr, $poll_expr:expr) => {
tokio::spawn(async move {
info!(concat!("Starting ", $name, " polling task"));
loop {
$poll_expr.await;
tokio::select! {
_ = $token.cancelled() => break,
_ = sleep($interval) => {}
}
}
info!(concat!($name, " task shut down."));
})
};
}
struct SocketGuard {
path: String,
}
@@ -196,23 +254,16 @@ pub async fn run_daemon(config_path: Option<PathBuf>) -> Result<()> {
// 1. Network Task
#[cfg(feature = "mod-network")]
if config.read().await.network.enabled {
let mut daemon = NetworkDaemon::new();
let token = cancel_token.clone();
let net_health = Arc::clone(&health);
tokio::spawn(async move {
info!("Starting Network polling task");
let mut daemon = NetworkDaemon::new();
loop {
if !crate::health::is_poll_in_backoff("net", &net_health).await {
let res = daemon.poll(&net_tx).await;
crate::health::handle_poll_result("net", res, &net_health).await;
}
tokio::select! {
_ = token.cancelled() => break,
_ = sleep(Duration::from_secs(1)) => {}
}
}
info!("Network task shut down.");
});
let h = Arc::clone(&health);
spawn_poll_loop!(
"net",
Duration::from_secs(1),
h,
token,
daemon.poll(&net_tx)
);
}
// 2. Fast Hardware Task (CPU, Mem, Load)
@@ -222,22 +273,14 @@ pub async fn run_daemon(config_path: Option<PathBuf>) -> Result<()> {
let fast_enabled = cfg.cpu.enabled || cfg.memory.enabled || cfg.sys.enabled;
drop(cfg);
if fast_enabled {
let mut daemon = HardwareDaemon::new();
let token = cancel_token.clone();
let hw_health = Arc::clone(&health);
tokio::spawn(async move {
info!("Starting Fast Hardware polling task");
let mut daemon = HardwareDaemon::new();
loop {
if !crate::health::is_poll_in_backoff("cpu", &hw_health).await {
daemon.poll_fast(&cpu_tx, &mem_tx, &sys_tx).await;
}
tokio::select! {
_ = token.cancelled() => break,
_ = sleep(Duration::from_secs(1)) => {}
}
}
info!("Fast Hardware task shut down.");
});
spawn_poll_loop_simple!(
"fast_hw",
Duration::from_secs(1),
token,
daemon.poll_fast(&cpu_tx, &mem_tx, &sys_tx)
);
}
}
@@ -248,47 +291,29 @@ pub async fn run_daemon(config_path: Option<PathBuf>) -> Result<()> {
let slow_enabled = cfg.gpu.enabled || cfg.disk.enabled;
drop(cfg);
if slow_enabled {
let mut daemon = HardwareDaemon::new();
let token = cancel_token.clone();
let slow_health = Arc::clone(&health);
tokio::spawn(async move {
info!("Starting Slow Hardware polling task");
let mut daemon = HardwareDaemon::new();
loop {
if !crate::health::is_poll_in_backoff("gpu", &slow_health).await {
daemon.poll_slow(&gpu_tx, &disks_tx).await;
}
tokio::select! {
_ = token.cancelled() => break,
_ = sleep(Duration::from_secs(5)) => {}
}
}
info!("Slow Hardware task shut down.");
});
spawn_poll_loop_simple!(
"slow_hw",
Duration::from_secs(5),
token,
daemon.poll_slow(&gpu_tx, &disks_tx)
);
}
}
// 4. Bluetooth Task
#[cfg(feature = "mod-bt")]
if config.read().await.bt.enabled {
let mut daemon = BtDaemon::new();
let token = cancel_token.clone();
let bt_health = Arc::clone(&health);
let h = Arc::clone(&health);
let poll_config = Arc::clone(&config);
let poll_receivers = receivers.clone();
tokio::spawn(async move {
info!("Starting Bluetooth polling task");
let mut daemon = BtDaemon::new();
loop {
if !crate::health::is_poll_in_backoff("bt", &bt_health).await {
let config = poll_config.read().await;
daemon.poll(&bt_tx, &poll_receivers, &config).await;
}
tokio::select! {
_ = token.cancelled() => break,
_ = bt_force_rx.recv() => {},
_ = sleep(Duration::from_secs(2)) => {}
}
}
info!("Bluetooth task shut down.");
spawn_poll_loop_triggered!("bt", Duration::from_secs(2), h, token, bt_force_rx, async {
let config = poll_config.read().await;
daemon.poll(&bt_tx, &poll_receivers, &config).await;
Ok(())
});
}
src/modules/dnd.rs
+48 -61
View File
@@ -7,10 +7,42 @@ use futures::StreamExt;
use tokio::sync::watch;
use tokio::time::Duration;
use tracing::{debug, error, info};
use zbus::{Connection, fdo::PropertiesProxy, names::InterfaceName, proxy};
use zbus::proxy;
use zbus::zvariant::OwnedValue;
use zbus::{Connection, fdo::PropertiesProxy};
pub struct DndModule;
/// Read dunst's `paused` property via raw D-Bus call.
async fn dunst_get_paused(connection: &Connection) -> anyhow::Result<bool> {
let reply = connection
.call_method(
Some("org.freedesktop.Notifications"),
"/org/freedesktop/Notifications",
Some("org.freedesktop.DBus.Properties"),
"Get",
&("org.dunstproject.cmd0", "paused"),
)
.await?;
let value: OwnedValue = reply.body().deserialize()?;
Ok(bool::try_from(&*value)?)
}
/// Set dunst's `paused` property via raw D-Bus call.
async fn dunst_set_paused(connection: &Connection, paused: bool) -> anyhow::Result<()> {
let value = zbus::zvariant::Value::from(paused);
connection
.call_method(
Some("org.freedesktop.Notifications"),
"/org/freedesktop/Notifications",
Some("org.freedesktop.DBus.Properties"),
"Set",
&("org.dunstproject.cmd0", "paused", value),
)
.await?;
Ok(())
}
impl WaybarModule for DndModule {
async fn run(
&self,
@@ -29,7 +61,7 @@ impl WaybarModule for DndModule {
message: format!("DBus connection failed: {}", e),
})?;
// Try toggling SwayNC
// Try SwayNC
if let Ok(proxy) = SwayncControlProxy::new(&connection).await
&& let Ok(is_dnd) = proxy.dnd().await
{
@@ -37,11 +69,9 @@ impl WaybarModule for DndModule {
return Ok(WaybarOutput::default());
}
// Try toggling Dunst
if let Ok(proxy) = DunstControlProxy::new(&connection).await
&& let Ok(is_paused) = proxy.paused().await
{
let _ = proxy.set_paused(!is_paused).await;
// Try Dunst via raw D-Bus
if let Ok(is_paused) = dunst_get_paused(&connection).await {
let _ = dunst_set_paused(&connection, !is_paused).await;
return Ok(WaybarOutput::default());
}
@@ -85,18 +115,6 @@ trait SwayncControl {
fn set_dnd(&self, value: bool) -> zbus::Result<()>;
}
#[proxy(
interface = "org.dunstproject.cmd0",
default_service = "org.freedesktop.Notifications",
default_path = "/org/freedesktop/Notifications"
)]
trait DunstControl {
#[zbus(property)]
fn paused(&self) -> zbus::Result<bool>;
#[zbus(property)]
fn set_paused(&self, value: bool) -> zbus::Result<()>;
}
impl DndDaemon {
pub fn new() -> Self {
Self
@@ -146,53 +164,19 @@ impl DndDaemon {
return Err(anyhow::anyhow!("SwayNC DND stream ended"));
}
// Try Dunst (polling via raw D-Bus Properties.Get for maximum compatibility)
let props_proxy = PropertiesProxy::builder(&connection)
.destination("org.freedesktop.Notifications")?
.path("/org/freedesktop/Notifications")?
.build()
.await;
match &props_proxy {
Ok(_) => debug!("Created Properties proxy for Dunst"),
Err(e) => debug!("Failed to create Properties proxy for Dunst: {}", e),
}
if let Ok(props) = props_proxy {
// Read paused property via raw Properties.Get
let initial = props
.get(
InterfaceName::from_static_str_unchecked("org.dunstproject.cmd0"),
"paused",
)
.await;
match &initial {
Ok(_) => debug!("Successfully read Dunst paused property"),
Err(e) => debug!("Failed to read Dunst paused property: {}", e),
}
if let Ok(value) = initial
&& let Ok(is_paused) = bool::try_from(&*value)
{
// Try Dunst via raw D-Bus calls (bypasses zbus proxy issues)
match dunst_get_paused(&connection).await {
Ok(is_paused) => {
info!("Found Dunst, using polling-based DND monitoring");
let _ = tx.send(DndState { is_dnd: is_paused });
loop {
tokio::time::sleep(Duration::from_secs(2)).await;
match props
.get(
InterfaceName::from_static_str_unchecked("org.dunstproject.cmd0"),
"paused",
)
.await
{
Ok(value) => {
if let Ok(is_paused) = bool::try_from(&*value) {
let current = tx.borrow().is_dnd;
if current != is_paused {
let _ = tx.send(DndState { is_dnd: is_paused });
}
match dunst_get_paused(&connection).await {
Ok(is_paused) => {
let current = tx.borrow().is_dnd;
if current != is_paused {
let _ = tx.send(DndState { is_dnd: is_paused });
}
}
Err(e) => {
@@ -204,6 +188,9 @@ impl DndDaemon {
return Err(anyhow::anyhow!("Dunst connection lost"));
}
Err(e) => {
info!("Dunst not available: {}", e);
}
}
Err(anyhow::anyhow!(