1use dbus::arg::PropMap;
24use dbus::blocking::LocalConnection;
25use dbus::message::MatchRule;
26use std::cell::{Cell, RefCell};
27use std::rc::Rc;
28use std::sync::atomic::{AtomicBool, Ordering};
29use std::sync::mpsc::{Receiver, Sender, TryRecvError};
30use std::sync::{Arc, Mutex};
31use std::time::{Duration, Instant};
32
33const SERVER: &str = "org.freedesktop.Notifications";
35const SERVER_PATH: &str = "/org/freedesktop/Notifications";
36const BUS: &str = "org.freedesktop.DBus";
38const BUS_PATH: &str = "/org/freedesktop/DBus";
39const DCONF_WRITER: &str = "ca.desrt.dconf.Writer";
41
42const CALL_WAIT: Duration = Duration::from_secs(3);
44const LISTENED: Duration = Duration::from_millis(250);
46const START_WAIT: Duration = Duration::from_secs(5);
48const HELD: Duration = Duration::from_secs(1);
50const PROPERTIES: &str = "org.freedesktop.DBus.Properties";
52
53#[derive(Clone, Debug, Default, PartialEq, Eq)]
55pub struct Server {
56 pub name: String,
58 pub vendor: String,
60 pub version: String,
61 pub capabilities: Vec<String>,
62 pub inhibits: bool,
65}
66
67pub fn server() -> Option<Server> {
70 let bus = session().ok()?;
71 if !has_owner(&bus, SERVER) {
72 return None;
73 }
74 let proxy = bus.with_proxy(SERVER, SERVER_PATH, CALL_WAIT);
75 let (name, vendor, version, _spec): (String, String, String, String) = proxy.method_call(SERVER, "GetServerInformation", ()).ok()?;
76 let capabilities = proxy.method_call(SERVER, "GetCapabilities", ()).map(|(found,): (Vec<String>,)| found).unwrap_or_default();
77 let mut inhibits = capabilities.iter().any(|c| c == "inhibitions");
78 if !inhibits && vendor == "KDE" {
80 use dbus::blocking::stdintf::org_freedesktop_dbus::Introspectable;
81 inhibits = proxy.introspect().is_ok_and(|xml| xml.contains("name=\"Inhibit\""));
82 }
83 Some(Server { name, vendor, version, capabilities, inhibits })
84}
85
86fn session() -> Result<LocalConnection, dbus::Error> {
88 crate::bus::session().map(LocalConnection::from)
89}
90
91fn has_owner(bus: &LocalConnection, name: &str) -> bool {
93 bus.with_proxy(BUS, BUS_PATH, CALL_WAIT).method_call(BUS, "NameHasOwner", (name,)).is_ok_and(|(owned,): (bool,)| owned)
94}
95
96fn owner_of(bus: &LocalConnection, name: &str) -> Option<String> {
98 bus.with_proxy(BUS, BUS_PATH, CALL_WAIT).method_call(BUS, "GetNameOwner", (name,)).ok().map(|(owner,): (String,)| owner)
99}
100
101pub fn inhibited() -> Option<bool> {
106 let bus = session().ok()?;
107 if !has_owner(&bus, SERVER) {
108 return None;
109 }
110 read_inhibited(&bus)
111}
112
113fn read_inhibited(bus: &LocalConnection) -> Option<bool> {
114 use dbus::blocking::stdintf::org_freedesktop_dbus::Properties;
115 bus.with_proxy(SERVER, SERVER_PATH, CALL_WAIT).get::<bool>(SERVER, "Inhibited").ok()
116}
117
118pub struct Watch {
124 stop: Sender<()>,
125 seen: Arc<Mutex<Option<bool>>>,
126 alive: Arc<AtomicBool>,
127}
128
129impl Watch {
130 pub fn seen(&self) -> Option<bool> {
132 self.seen.lock().ok().and_then(|seen| *seen)
133 }
134
135 pub fn alive(&self) -> bool {
137 self.alive.load(Ordering::Relaxed)
138 }
139}
140
141impl Drop for Watch {
142 fn drop(&mut self) {
143 let _ = self.stop.send(());
144 }
145}
146
147pub fn watch_inhibited(quiet: Arc<Mutex<Instant>>, told: Box<dyn Fn(bool) + Send>) -> Result<Watch, String> {
150 let (stop, stopped) = std::sync::mpsc::channel::<()>();
151 let (ready, started) = std::sync::mpsc::channel();
152 let seen: Arc<Mutex<Option<bool>>> = Arc::default();
153 let alive = Arc::new(AtomicBool::new(true));
154 let (noted, living) = (Arc::clone(&seen), Arc::clone(&alive));
155 std::thread::spawn(move || {
156 follow_inhibited(¬ed, &quiet, &*told, &stopped, &ready);
157 living.store(false, Ordering::Relaxed);
158 });
159 match started.recv_timeout(START_WAIT) {
160 Ok(Ok(())) => Ok(Watch { stop, seen, alive }),
161 Ok(Err(e)) => Err(e),
162 Err(_) => Err("the session bus did not answer".to_string()),
163 }
164}
165
166fn follow_inhibited(seen: &Mutex<Option<bool>>, quiet: &Mutex<Instant>, told: &dyn Fn(bool), stopped: &Receiver<()>, ready: &Sender<Result<(), String>>) {
167 let bus = match session() {
168 Ok(bus) => bus,
169 Err(e) => {
170 let _ = ready.send(Err(said(&e)));
171 return;
172 }
173 };
174 let heard: Rc<Cell<Option<(Instant, bool)>>> = Rc::default();
177 let owner = Rc::new(RefCell::new(owner_of(&bus, SERVER).unwrap_or_default()));
178 let (new_owner, server_changed) = (Rc::clone(&owner), Rc::clone(&heard));
179 let owners = bus.add_match(MatchRule::new_signal(BUS, "NameOwnerChanged").with_sender(BUS), move |(name, _old, new): (String, String, String), _: &LocalConnection, _: &dbus::Message| {
180 if name == SERVER {
181 *new_owner.borrow_mut() = new;
182 server_changed.set(Some((Instant::now(), true)));
183 }
184 true
185 });
186 let (from, property_changed) = (Rc::clone(&owner), Rc::clone(&heard));
187 let changes = bus.add_match(MatchRule::new_signal(PROPERTIES, "PropertiesChanged").with_path(SERVER_PATH), move |(interface, changed, _gone): (String, PropMap, Vec<String>), _: &LocalConnection, message: &dbus::Message| {
188 let ours = message.sender().is_some_and(|sender| *from.borrow() == sender.to_string());
189 if ours && interface == SERVER && changed.contains_key("Inhibited") {
190 let fresh = property_changed.get().is_some_and(|(_, fresh)| fresh);
191 property_changed.set(Some((Instant::now(), fresh)));
192 }
193 true
194 });
195 if let Err(e) = owners.and(changes) {
196 let _ = ready.send(Err(said(&e)));
197 return;
198 }
199 if let Ok(mut noted) = seen.lock() {
201 *noted = read_inhibited(&bus);
202 }
203 let _ = ready.send(Ok(()));
204 loop {
205 match stopped.try_recv() {
206 Ok(()) | Err(TryRecvError::Disconnected) => return,
207 Err(TryRecvError::Empty) => {}
208 }
209 if let Some((at, fresh)) = heard.get() {
210 let settled = quiet.lock().map_or(true, |quiet| Instant::now() >= *quiet);
211 if at.elapsed() >= HELD && settled {
212 heard.set(None);
213 let now = read_inhibited(&bus);
214 let before = seen.lock().ok().and_then(|mut noted| std::mem::replace(&mut *noted, now));
215 if !fresh
216 && let (Some(before), Some(now)) = (before, now)
217 && before != now
218 {
219 told(now);
220 }
221 }
222 }
223 if bus.process(LISTENED).is_err() {
224 return;
225 }
226 }
227}
228
229pub struct Inhibition {
232 orders: Sender<Sender<Result<(), String>>>,
233 alive: Arc<AtomicBool>,
234 cookie: u32,
235}
236
237impl Inhibition {
238 pub fn alive(&self) -> bool {
240 self.alive.load(Ordering::Relaxed)
241 }
242
243 pub fn cookie(&self) -> u32 {
245 self.cookie
246 }
247
248 pub fn release(self, wait: Duration) -> Result<(), String> {
252 let (done, answer) = std::sync::mpsc::channel();
253 if self.orders.send(done).is_err() {
254 return Ok(());
255 }
256 answer.recv_timeout(wait).unwrap_or_else(|_| Err("no answer from the notification server".to_string()))
257 }
258}
259
260pub fn inhibit(desktop_entry: &str, reason: &str) -> Result<Inhibition, String> {
265 let (orders, asked) = std::sync::mpsc::channel();
266 let (ready, started) = std::sync::mpsc::channel();
267 let alive = Arc::new(AtomicBool::new(true));
268 let (entry, reason, living) = (desktop_entry.to_string(), reason.to_string(), Arc::clone(&alive));
269 std::thread::spawn(move || {
271 hold(&entry, &reason, &asked, &ready);
272 living.store(false, Ordering::Relaxed);
273 });
274 match started.recv_timeout(START_WAIT) {
275 Ok(Ok(cookie)) => Ok(Inhibition { orders, alive, cookie }),
276 Ok(Err(e)) => Err(e),
277 Err(_) => Err("no answer from the notification server".to_string()),
278 }
279}
280
281fn hold(entry: &str, reason: &str, asked: &Receiver<Sender<Result<(), String>>>, ready: &Sender<Result<u32, String>>) {
284 let bus = match session() {
285 Ok(bus) => bus,
286 Err(e) => {
287 let _ = ready.send(Err(said(&e)));
288 return;
289 }
290 };
291 let restarted: Rc<Cell<bool>> = Rc::default();
295 let heard = Rc::clone(&restarted);
296 let _ = bus.add_match(MatchRule::new_signal(BUS, "NameOwnerChanged").with_sender(BUS), move |(name, _old, new): (String, String, String), _: &LocalConnection, _: &dbus::Message| {
297 if name == SERVER && !new.is_empty() {
298 heard.set(true);
299 }
300 true
301 });
302 let server = bus.with_proxy(SERVER, SERVER_PATH, CALL_WAIT);
303 let ask = || server.method_call(SERVER, "Inhibit", (entry, reason, PropMap::new())).map(|(cookie,): (u32,)| cookie);
304 let mut cookie = match ask() {
305 Ok(cookie) => cookie,
306 Err(e) => {
307 let _ = ready.send(Err(said(&e)));
308 return;
309 }
310 };
311 let _ = ready.send(Ok(cookie));
312 let lift = |cookie: u32| -> Result<(), String> {
313 if cookie == 0 {
314 return Ok(());
315 }
316 server.method_call(SERVER, "UnInhibit", (cookie,)).map_err(|e| said(&e))
317 };
318 loop {
319 match asked.try_recv() {
320 Ok(done) => {
321 let _ = done.send(lift(cookie));
322 return;
323 }
324 Err(TryRecvError::Disconnected) => {
325 let _ = lift(cookie);
326 return;
327 }
328 Err(TryRecvError::Empty) => {}
329 }
330 if restarted.replace(false) {
331 cookie = ask().unwrap_or(0);
333 }
334 if bus.process(LISTENED).is_err() {
335 return;
336 }
337 }
338}
339
340fn said(e: &dbus::Error) -> String {
342 e.message().or(e.name()).unwrap_or("D-Bus").to_string()
343}
344
345pub struct Changes {
349 stop: Sender<()>,
350 heard: Arc<Mutex<Vec<Instant>>>,
351 alive: Arc<AtomicBool>,
352}
353
354impl Changes {
355 pub fn since(&self, at: Instant) -> usize {
357 self.heard.lock().map_or(0, |heard| heard.iter().filter(|when| **when > at).count())
358 }
359
360 pub fn alive(&self) -> bool {
362 self.alive.load(Ordering::Relaxed)
363 }
364}
365
366impl Drop for Changes {
367 fn drop(&mut self) {
368 let _ = self.stop.send(());
369 }
370}
371
372pub fn watch_dconf(key: &str) -> Result<Changes, String> {
374 watch(key, None)
375}
376
377pub fn watch_dconf_told(key: &str, told: Box<dyn Fn() + Send>) -> Result<Changes, String> {
380 watch(key, Some(told))
381}
382
383fn watch(key: &str, told: Option<Box<dyn Fn() + Send>>) -> Result<Changes, String> {
384 let (stop, stopped) = std::sync::mpsc::channel::<()>();
385 let (ready, started) = std::sync::mpsc::channel();
386 let heard: Arc<Mutex<Vec<Instant>>> = Arc::default();
387 let alive = Arc::new(AtomicBool::new(true));
388 let (key, noted, living) = (key.to_string(), Arc::clone(&heard), Arc::clone(&alive));
389 std::thread::spawn(move || {
390 listen(&key, ¬ed, told, &stopped, &ready);
391 living.store(false, Ordering::Relaxed);
392 });
393 match started.recv_timeout(START_WAIT) {
394 Ok(Ok(())) => Ok(Changes { stop, heard, alive }),
395 Ok(Err(e)) => Err(e),
396 Err(_) => Err("the session bus did not answer".to_string()),
397 }
398}
399
400fn listen(key: &str, heard: &Arc<Mutex<Vec<Instant>>>, told: Option<Box<dyn Fn() + Send>>, stopped: &Receiver<()>, ready: &Sender<Result<(), String>>) {
401 let bus = match session() {
402 Ok(bus) => bus,
403 Err(e) => {
404 let _ = ready.send(Err(said(&e)));
405 return;
406 }
407 };
408 let (key, noted) = (key.to_string(), Arc::clone(heard));
409 let matched = bus.add_match(MatchRule::new_signal(DCONF_WRITER, "Notify"), move |(prefix, changes, _tag): (String, Vec<String>, String), _: &LocalConnection, _: &dbus::Message| {
410 if touches(&key, &prefix, &changes) {
411 if let Ok(mut noted) = noted.lock() {
412 noted.push(Instant::now());
413 }
414 if let Some(told) = &told {
415 told();
416 }
417 }
418 true
419 });
420 if let Err(e) = matched {
421 let _ = ready.send(Err(said(&e)));
422 return;
423 }
424 let _ = ready.send(Ok(()));
425 loop {
426 match stopped.try_recv() {
427 Ok(()) | Err(TryRecvError::Disconnected) => return,
428 Err(TryRecvError::Empty) => {}
429 }
430 if bus.process(LISTENED).is_err() {
431 return;
432 }
433 }
434}
435
436fn touches(key: &str, prefix: &str, changes: &[String]) -> bool {
440 let reaches = |path: &str| path == key || (path.ends_with('/') && key.starts_with(path));
441 if changes.is_empty() {
442 return reaches(prefix);
443 }
444 changes.iter().any(|change| reaches(&format!("{prefix}{change}")))
445}
446
447#[cfg(test)]
448mod tests {
449 use super::*;
450 use dbus::blocking::stdintf::org_freedesktop_dbus::RequestNameReply;
451 use dbus::channel::{MatchingReceiver, Sender as _};
452 use std::sync::mpsc::channel;
453
454 #[test]
455 fn a_dconf_change_touches_its_key_or_a_folder_reset() {
456 let key = "/org/gnome/desktop/notifications/show-banners";
457 assert!(touches(key, key, &[String::new()]));
458 assert!(touches(key, "/org/gnome/desktop/notifications/", &["show-banners".into(), "show-in-lock-screen".into()]));
459 assert!(touches(key, "/org/gnome/desktop/notifications/", &[String::new()]), "the folder reset whole");
460 assert!(touches(key, "/org/gnome/", &["desktop/notifications/".into()]));
461 assert!(!touches(key, "/org/gnome/desktop/notifications/", &["show-in-lock-screen".into()]));
462 assert!(!touches(key, "/org/gnome/desktop/notifications/show-banners-x", &[String::new()]));
463 assert!(!touches(key, "/org/gnome/desktop/interface/", &["color-scheme".into()]));
464 }
465
466 #[derive(Debug, PartialEq)]
468 enum Asked {
469 Inhibit { entry: String, reason: String, caller: String },
470 UnInhibit(u32),
471 }
472
473 use crate::bus::NOTIFICATION_SERVER as ON_THE_BUS;
476
477 fn stand_in(first: u32, told: Sender<Asked>, stop: Arc<AtomicBool>, life: Duration) -> Option<()> {
483 stand_in_saying(first, told, stop, life, None)
484 }
485
486 struct Says {
489 value: Arc<AtomicBool>,
490 set: Receiver<bool>,
491 }
492
493 fn stand_in_saying(first: u32, told: Sender<Asked>, stop: Arc<AtomicBool>, life: Duration, says: Option<Says>) -> Option<()> {
495 let (ready, started) = channel::<bool>();
496 std::thread::spawn(move || {
497 let until = Instant::now() + Duration::from_secs(2);
500 let owner = loop {
501 let bus = session().ok().filter(|bus| matches!(bus.request_name(SERVER, false, false, true), Ok(RequestNameReply::PrimaryOwner)));
502 if bus.is_some() || Instant::now() >= until {
503 break bus;
504 }
505 std::thread::sleep(Duration::from_millis(50));
506 };
507 let Some(bus) = owner else {
508 let _ = ready.send(false);
509 return;
510 };
511 let next = Rc::new(Cell::new(first));
512 let value = says.as_ref().map(|says| Arc::clone(&says.value));
513 bus.start_receive(
514 MatchRule::new_method_call(),
515 Box::new(move |call: dbus::Message, bus: &LocalConnection| {
516 let reply = match call.member().as_deref() {
517 Some("GetServerInformation") => call.method_return().append3("Plasma", "KDE", "6.7.5").append1("1.2"),
518 Some("GetCapabilities") => call.method_return().append1(vec!["body", "actions", "persistence", "inhibitions"]),
519 Some("Inhibit") => {
520 let (entry, reason, _hints): (String, String, PropMap) = call.read3().unwrap();
521 let caller = call.sender().map(|s| s.to_string()).unwrap_or_default();
522 let _ = told.send(Asked::Inhibit { entry, reason, caller });
523 let cookie = next.get();
524 next.set(cookie + 1);
525 call.method_return().append1(cookie)
526 }
527 Some("UnInhibit") => {
528 let (cookie,): (u32,) = call.read_all().unwrap();
529 let _ = told.send(Asked::UnInhibit(cookie));
530 call.method_return()
531 }
532 Some("Get") if call.interface().as_deref() == Some(PROPERTIES) => {
533 let (_interface, name): (String, String) = call.read2().unwrap();
534 match &value {
535 Some(value) if name == "Inhibited" => call.method_return().append1(dbus::arg::Variant(value.load(Ordering::Relaxed))),
536 _ => return true,
537 }
538 }
539 _ => return true,
540 };
541 let _ = bus.send(reply);
542 true
543 }),
544 );
545 let _ = ready.send(true);
546 let until = Instant::now() + life;
547 while !stop.load(Ordering::Relaxed) && Instant::now() < until {
548 while let Some(on) = says.as_ref().and_then(|says| says.set.try_recv().ok()) {
550 if let Some(says) = &says {
551 says.value.store(on, Ordering::Relaxed);
552 }
553 let mut changed = PropMap::new();
554 changed.insert("Inhibited".to_string(), dbus::arg::Variant(Box::new(on) as Box<dyn dbus::arg::RefArg>));
555 let signal = dbus::Message::new_signal(SERVER_PATH, PROPERTIES, "PropertiesChanged").unwrap().append3(SERVER, changed, Vec::<String>::new());
556 let _ = bus.send(signal);
557 }
558 let _ = bus.process(Duration::from_millis(50));
559 }
560 let _ = bus.release_name(SERVER);
561 });
562 started.recv_timeout(Duration::from_secs(5)).unwrap_or(false).then_some(())
563 }
564
565 fn connected(name: &str) -> bool {
567 session().is_ok_and(|bus| has_owner(&bus, name))
568 }
569
570 #[test]
574 fn plasma_inhibition_is_held_while_the_pause_lasts_and_lifted_after() {
575 let _alone = ON_THE_BUS.lock().unwrap_or_else(|e| e.into_inner());
576 let (told, asked) = channel();
577 let stop = Arc::new(AtomicBool::new(false));
578 if stand_in(42, told, Arc::clone(&stop), Duration::from_secs(30)).is_none() {
579 eprintln!("No bus of the test's own (dbus-run-session), or a notification server on it already: skipped.");
580 return;
581 }
582 let wait = Duration::from_secs(5);
583 let found = server().expect("the stand-in answers");
584 assert_eq!((found.name.as_str(), found.vendor.as_str(), found.inhibits), ("Plasma", "KDE", true));
585 let held = inhibit("com.aurelienpierre.Sioul", "Pause").expect("inhibited");
586 assert_eq!(held.cookie(), 42);
587 let Asked::Inhibit { entry, reason, caller } = asked.recv_timeout(wait).unwrap() else { panic!("Inhibit first") };
588 assert_eq!((entry.as_str(), reason.as_str()), ("com.aurelienpierre.Sioul", "Pause"));
589 std::thread::sleep(Duration::from_millis(600));
592 assert!(held.alive() && connected(&caller), "the inhibition's connection is kept");
593 held.release(wait).expect("lifted");
594 assert_eq!(asked.recv_timeout(wait).unwrap(), Asked::UnInhibit(42));
595 let until = Instant::now() + wait;
597 while connected(&caller) && Instant::now() < until {
598 std::thread::sleep(Duration::from_millis(50));
599 }
600 assert!(!connected(&caller), "the connection closed once lifted");
601 let dropped = inhibit("com.aurelienpierre.Sioul", "Free time").expect("inhibited");
603 assert!(matches!(asked.recv_timeout(wait).unwrap(), Asked::Inhibit { reason, .. } if reason == "Free time"));
604 drop(dropped);
605 assert_eq!(asked.recv_timeout(wait).unwrap(), Asked::UnInhibit(43));
606 stop.store(true, Ordering::Relaxed);
607 }
608
609 #[test]
612 fn plasma_restarted_is_asked_again() {
613 let _alone = ON_THE_BUS.lock().unwrap_or_else(|e| e.into_inner());
614 let (told, asked) = channel();
615 let first_stop = Arc::new(AtomicBool::new(false));
616 if stand_in(7, told.clone(), Arc::clone(&first_stop), Duration::from_secs(30)).is_none() {
617 eprintln!("No bus of the test's own (dbus-run-session), or a notification server on it already: skipped.");
618 return;
619 }
620 let wait = Duration::from_secs(5);
621 let held = inhibit("com.aurelienpierre.Sioul", "Pause").expect("inhibited");
622 assert!(matches!(asked.recv_timeout(wait).unwrap(), Asked::Inhibit { .. }));
623 first_stop.store(true, Ordering::Relaxed);
625 let until = Instant::now() + wait;
626 let second_stop = Arc::new(AtomicBool::new(false));
627 while stand_in(100, told.clone(), Arc::clone(&second_stop), Duration::from_secs(30)).is_none() {
628 assert!(Instant::now() < until, "the second stand-in could not take the name");
629 std::thread::sleep(Duration::from_millis(50));
630 }
631 assert!(matches!(asked.recv_timeout(wait).unwrap(), Asked::Inhibit { reason, .. } if reason == "Pause"), "asked again of the new server");
632 held.release(wait).expect("lifted");
633 assert_eq!(asked.recv_timeout(wait).unwrap(), Asked::UnInhibit(100));
634 second_stop.store(true, Ordering::Relaxed);
635 }
636
637 #[test]
642 fn the_session_s_do_not_disturb_is_followed() {
643 let _alone = ON_THE_BUS.lock().unwrap_or_else(|e| e.into_inner());
644 let (told, _asked) = channel();
645 let stop = Arc::new(AtomicBool::new(false));
646 let (set, announced) = channel();
647 let says = Says { value: Arc::new(AtomicBool::new(false)), set: announced };
648 if stand_in_saying(1, told.clone(), Arc::clone(&stop), Duration::from_secs(40), Some(says)).is_none() {
649 eprintln!("No bus of the test's own (dbus-run-session), or a notification server on it already: skipped.");
650 return;
651 }
652 assert_eq!(inhibited(), Some(false));
653 let quiet = Arc::new(Mutex::new(Instant::now()));
654 let (heard, hears) = channel();
655 let watch = watch_inhibited(Arc::clone(&quiet), Box::new(move |on| {
656 let _ = heard.send(on);
657 }))
658 .expect("following");
659 assert_eq!(watch.seen(), Some(false), "read at the start, not told");
660 set.send(true).unwrap();
662 assert_eq!(hears.recv_timeout(Duration::from_secs(5)), Ok(true));
663 assert_eq!(watch.seen(), Some(true));
664 set.send(false).unwrap();
666 set.send(true).unwrap();
667 assert!(hears.recv_timeout(Duration::from_millis(2_500)).is_err(), "a flicker is no change");
668 *quiet.lock().unwrap() = Instant::now() + Duration::from_secs(2);
670 set.send(false).unwrap();
671 assert!(hears.recv_timeout(Duration::from_millis(1_500)).is_err(), "not before Sioul's change settled");
672 assert_eq!(hears.recv_timeout(Duration::from_secs(4)), Ok(false));
673 set.send(true).unwrap();
676 assert_eq!(hears.recv_timeout(Duration::from_secs(5)), Ok(true));
677 *quiet.lock().unwrap() = Instant::now() + Duration::from_secs(2);
678 set.send(false).unwrap();
679 std::thread::sleep(Duration::from_millis(300));
680 set.send(true).unwrap();
681 assert!(hears.recv_timeout(Duration::from_millis(4_000)).is_err(), "on before, on after: nothing");
682 assert_eq!(watch.seen(), Some(true));
683 stop.store(true, Ordering::Relaxed);
685 let wait = Duration::from_secs(5);
686 let second_stop = Arc::new(AtomicBool::new(false));
687 let (_set_again, announced_again) = channel();
688 let says_again = Says { value: Arc::new(AtomicBool::new(true)), set: announced_again };
689 assert!(stand_in_saying(50, told.clone(), Arc::clone(&second_stop), Duration::from_secs(30), Some(says_again)).is_some(), "the second stand-in took the name");
691 let until = Instant::now() + wait;
692 while watch.seen() != Some(true) && Instant::now() < until {
693 std::thread::sleep(Duration::from_millis(50));
694 }
695 assert_eq!(watch.seen(), Some(true), "the new server read");
696 assert!(hears.recv_timeout(Duration::from_millis(500)).is_err(), "and not told");
697 assert!(watch.alive());
698 second_stop.store(true, Ordering::Relaxed);
699 }
700
701 #[test]
703 fn dconf_tells_each_change_of_its_key() {
704 let _alone = ON_THE_BUS.lock().unwrap_or_else(|e| e.into_inner());
705 let free = |bus: &LocalConnection| {
706 let until = Instant::now() + Duration::from_secs(1);
707 while has_owner(bus, SERVER) && Instant::now() < until {
708 std::thread::sleep(Duration::from_millis(50));
709 }
710 !has_owner(bus, SERVER)
711 };
712 let Some(bus) = session().ok().filter(|bus| free(bus)) else {
713 eprintln!("No bus of the test's own (dbus-run-session): skipped.");
714 return;
715 };
716 let key = "/org/gnome/desktop/notifications/show-banners";
717 let (heard, hears) = channel();
718 let _changes = watch_dconf_told(key, Box::new(move || {
719 let _ = heard.send(());
720 }))
721 .expect("listening");
722 let notify = |prefix: &str, changed: Vec<&str>| {
723 let signal = dbus::Message::new_signal("/ca/desrt/dconf/Writer/user", DCONF_WRITER, "Notify").unwrap().append3(prefix, changed, "tag");
724 let _ = bus.send(signal);
725 bus.process(Duration::from_millis(10)).ok();
726 };
727 notify("/org/gnome/desktop/interface/", vec!["color-scheme"]);
728 notify(key, vec![""]);
729 assert_eq!(hears.recv_timeout(Duration::from_secs(5)), Ok(()));
730 assert!(hears.recv_timeout(Duration::from_millis(300)).is_err(), "another key's change is not told");
731 }
732
733 #[test]
737 fn dconf_changes_of_a_key_are_heard() {
738 let _alone = ON_THE_BUS.lock().unwrap_or_else(|e| e.into_inner());
739 let free = |bus: &LocalConnection| {
741 let until = Instant::now() + Duration::from_secs(1);
742 while has_owner(bus, SERVER) && Instant::now() < until {
743 std::thread::sleep(Duration::from_millis(50));
744 }
745 !has_owner(bus, SERVER)
746 };
747 let Some(bus) = session().ok().filter(|bus| free(bus)) else {
748 eprintln!("No bus of the test's own (dbus-run-session): skipped.");
749 return;
750 };
751 let key = "/org/gnome/desktop/notifications/show-banners";
752 let changes = watch_dconf(key).expect("listening");
753 let start = Instant::now();
754 let notify = |prefix: &str, changed: Vec<&str>| {
755 let signal = dbus::Message::new_signal("/ca/desrt/dconf/Writer/user", DCONF_WRITER, "Notify").unwrap().append3(prefix, changed, "tag");
756 let _ = bus.send(signal);
757 bus.process(Duration::from_millis(10)).ok();
758 };
759 notify("/org/gnome/desktop/interface/", vec!["color-scheme"]);
760 notify(key, vec![""]);
761 notify("/org/gnome/desktop/notifications/", vec!["show-in-lock-screen", "show-banners"]);
762 let until = Instant::now() + Duration::from_secs(5);
763 while changes.since(start) < 2 && Instant::now() < until {
764 std::thread::sleep(Duration::from_millis(20));
765 }
766 std::thread::sleep(Duration::from_millis(200));
767 assert_eq!(changes.since(start), 2, "two changes of the key, one of another");
768 assert_eq!(changes.since(Instant::now()), 0);
769 assert!(changes.alive());
770 }
771}