ferroui_freedesktop/
dbus_call_queue.rs

1//! Calls made one at a time, in the order they were queued (the port of
2//! `DBusCallQueue.cs`).
3//!
4//! The calls of an input method context have to arrive in order: a key
5//! must not overtake the focus change that precedes it. D-Bus keeps the
6//! order of messages, but a call that is made from the continuation of
7//! another one is sent whenever that continuation runs; the queue makes
8//! the order the one of `enqueue`.
9
10use ferroui_base::input::LocalBoxFuture;
11use ferroui_base::threading::Dispatcher;
12use std::cell::{Cell, RefCell};
13use std::collections::VecDeque;
14use std::fmt;
15use std::future::Future;
16use std::pin::Pin;
17use std::rc::Rc;
18use std::task::{Context, Poll, Waker};
19
20/// Why a queued call did not give a result.
21#[derive(Debug)]
22pub enum DBusCallError {
23    /// The queue was emptied before the call was made (`FailAll`: the
24    /// `OperationCanceledException` of the reference).
25    Canceled,
26    /// The call failed.
27    DBus(zbus::Error),
28}
29
30impl fmt::Display for DBusCallError {
31    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
32        match self {
33            DBusCallError::Canceled => f.write_str("The operation was canceled."),
34            DBusCallError::DBus(error) => error.fmt(f),
35        }
36    }
37}
38
39impl std::error::Error for DBusCallError {}
40
41impl From<zbus::Error> for DBusCallError {
42    fn from(error: zbus::Error) -> Self {
43        DBusCallError::DBus(error)
44    }
45}
46
47impl From<zbus::fdo::Error> for DBusCallError {
48    fn from(error: zbus::fdo::Error) -> Self {
49        DBusCallError::DBus(error.into())
50    }
51}
52
53/// The result of a call over D-Bus.
54pub type DBusResult<T = ()> = Result<T, DBusCallError>;
55
56type Callback = Box<dyn FnOnce() -> LocalBoxFuture<DBusResult>>;
57type OnFinish = Box<dyn FnOnce(Option<DBusCallError>)>;
58type ErrorHandler = Box<dyn Fn(DBusCallError) -> LocalBoxFuture<()>>;
59
60struct Item {
61    callback: Callback,
62    on_finish: Option<OnFinish>,
63}
64
65struct Inner {
66    error_handler: ErrorHandler,
67    q: RefCell<VecDeque<Item>>,
68    processing: Cell<bool>,
69}
70
71/// A queue of asynchronous calls of which one is in flight at a time.
72#[derive(Clone)]
73pub struct DBusCallQueue {
74    inner: Rc<Inner>,
75}
76
77impl DBusCallQueue {
78    /// `error_handler` is awaited with the failure of a call nobody waits
79    /// for (one queued with [`enqueue`](Self::enqueue)).
80    pub fn new(error_handler: impl Fn(DBusCallError) -> LocalBoxFuture<()> + 'static) -> Self {
81        Self {
82            inner: Rc::new(Inner {
83                error_handler: Box::new(error_handler),
84                q: RefCell::new(VecDeque::new()),
85                processing: Cell::new(false),
86            }),
87        }
88    }
89
90    pub fn enqueue<F>(&self, cb: impl FnOnce() -> F + 'static)
91    where
92        F: Future<Output = DBusResult> + 'static,
93    {
94        self.inner.q.borrow_mut().push_back(Item { callback: Box::new(move || Box::pin(cb())), on_finish: None });
95        self.process();
96    }
97
98    /// Queues a call and gives the future of its completion
99    /// (`EnqueueAsync`).
100    pub fn enqueue_async<F>(&self, cb: impl FnOnce() -> F + 'static) -> impl Future<Output = DBusResult> + 'static
101    where
102        F: Future<Output = DBusResult> + 'static,
103    {
104        let (sender, receiver) = completion::<DBusResult>();
105        self.inner.q.borrow_mut().push_back(Item {
106            callback: Box::new(move || Box::pin(cb())),
107            on_finish: Some(Box::new(move |e| match e {
108                None => sender.try_set(Ok(())),
109                Some(e) => sender.try_set(Err(e)),
110            })),
111        });
112        self.process();
113        receiver
114    }
115
116    /// Queues a call with a result and gives the future of the result
117    /// (`EnqueueAsync<T>`).
118    pub fn enqueue_async_with_result<T, F>(
119        &self,
120        cb: impl FnOnce() -> F + 'static,
121    ) -> impl Future<Output = DBusResult<T>> + 'static
122    where
123        T: 'static,
124        F: Future<Output = DBusResult<T>> + 'static,
125    {
126        let (sender, receiver) = completion::<DBusResult<T>>();
127        let result_sender = sender.clone();
128        self.inner.q.borrow_mut().push_back(Item {
129            callback: Box::new(move || {
130                Box::pin(async move {
131                    let res = cb().await?;
132                    result_sender.try_set(Ok(res));
133                    Ok(())
134                })
135            }),
136            on_finish: Some(Box::new(move |e| {
137                if let Some(e) = e {
138                    sender.try_set(Err(e));
139                }
140            })),
141        });
142        self.process();
143        receiver
144    }
145
146    /// Makes the queued calls, one after the other (`Process`, an
147    /// `async void` of the reference: a task of the UI dispatcher here,
148    /// which starts with the next job of the dispatcher and not inside the
149    /// call that queued).
150    fn process(&self) {
151        if self.inner.processing.get() {
152            return;
153        }
154        self.inner.processing.set(true);
155
156        let inner = self.inner.clone();
157        drop(Dispatcher::ui_thread().invoke_async_task_local(move || async move {
158            // The flag is reset when the loop is left in any way, as the
159            // `finally` of the reference.
160            struct ResetProcessing(Rc<Inner>);
161            impl Drop for ResetProcessing {
162                fn drop(&mut self) {
163                    self.0.processing.set(false);
164                }
165            }
166            let _reset = ResetProcessing(inner.clone());
167
168            loop {
169                let Some(item) = inner.q.borrow_mut().pop_front() else {
170                    break;
171                };
172                match (item.callback)().await {
173                    Ok(()) => {
174                        if let Some(on_finish) = item.on_finish {
175                            on_finish(None);
176                        }
177                    }
178                    Err(e) => {
179                        if let Some(on_finish) = item.on_finish {
180                            on_finish(Some(e));
181                        } else {
182                            (inner.error_handler)(e).await;
183                        }
184                    }
185                }
186            }
187        }));
188    }
189
190    /// Empties the queue; the calls somebody waits for fail as canceled.
191    pub fn fail_all(&self) {
192        loop {
193            let Some(item) = self.inner.q.borrow_mut().pop_front() else {
194                break;
195            };
196            if let Some(on_finish) = item.on_finish {
197                on_finish(Some(DBusCallError::Canceled));
198            }
199        }
200    }
201}
202
203/// The two ends of a result that is set once (a `TaskCompletionSource`).
204/// A receiver whose sender is gone without a result resolves as canceled
205/// where the task of the reference would never complete: nothing waits
206/// for ever on a call that was dropped.
207pub(crate) fn completion<T: CanceledValue>() -> (CompletionSender<T>, CompletionReceiver<T>) {
208    let state = Rc::new(CompletionState { value: RefCell::new(None), waker: RefCell::new(None), senders: Cell::new(1) });
209    (CompletionSender { state: state.clone() }, CompletionReceiver { state })
210}
211
212/// A result type that can say "canceled".
213pub(crate) trait CanceledValue {
214    fn canceled() -> Self;
215}
216
217impl<T> CanceledValue for DBusResult<T> {
218    fn canceled() -> Self {
219        Err(DBusCallError::Canceled)
220    }
221}
222
223impl CanceledValue for () {
224    fn canceled() -> Self {}
225}
226
227struct CompletionState<T> {
228    value: RefCell<Option<T>>,
229    waker: RefCell<Option<Waker>>,
230    senders: Cell<usize>,
231}
232
233impl<T> CompletionState<T> {
234    fn wake(&self) {
235        if let Some(waker) = self.waker.borrow_mut().take() {
236            waker.wake();
237        }
238    }
239}
240
241pub(crate) struct CompletionSender<T> {
242    state: Rc<CompletionState<T>>,
243}
244
245impl<T> CompletionSender<T> {
246    /// Sets the result unless one was set (`TrySetResult`).
247    pub(crate) fn try_set(&self, value: T) {
248        {
249            let mut slot = self.state.value.borrow_mut();
250            if slot.is_some() {
251                return;
252            }
253            *slot = Some(value);
254        }
255        self.state.wake();
256    }
257}
258
259impl<T> Clone for CompletionSender<T> {
260    fn clone(&self) -> Self {
261        self.state.senders.set(self.state.senders.get() + 1);
262        Self { state: self.state.clone() }
263    }
264}
265
266impl<T> Drop for CompletionSender<T> {
267    fn drop(&mut self) {
268        self.state.senders.set(self.state.senders.get() - 1);
269        if self.state.senders.get() == 0 {
270            self.state.wake();
271        }
272    }
273}
274
275pub(crate) struct CompletionReceiver<T> {
276    state: Rc<CompletionState<T>>,
277}
278
279impl<T: CanceledValue> Future for CompletionReceiver<T> {
280    type Output = T;
281
282    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
283        if let Some(value) = self.state.value.borrow_mut().take() {
284            return Poll::Ready(value);
285        }
286        if self.state.senders.get() == 0 {
287            return Poll::Ready(T::canceled());
288        }
289        *self.state.waker.borrow_mut() = Some(cx.waker().clone());
290        Poll::Pending
291    }
292}
293
294#[cfg(test)]
295mod tests {
296    // Not from the reference, which has no tests of this class.
297    use super::*;
298    use crate::test_support::pump_until;
299    use std::cell::RefCell;
300
301    fn queue_with_log(log: &Rc<RefCell<Vec<String>>>) -> DBusCallQueue {
302        let log = log.clone();
303        DBusCallQueue::new(move |e| {
304            let log = log.clone();
305            Box::pin(async move { log.borrow_mut().push(format!("error handler: {e}")) })
306        })
307    }
308
309    fn failure(text: &str) -> DBusCallError {
310        DBusCallError::DBus(zbus::Error::Failure(text.to_string()))
311    }
312
313    #[test]
314    fn calls_run_one_at_a_time_in_the_order_they_were_queued() {
315        let _scope = crate::test_support::scope();
316        let log = Rc::new(RefCell::new(Vec::new()));
317        let queue = queue_with_log(&log);
318        let (release_first, first_released) = completion::<()>();
319
320        {
321            let log = log.clone();
322            queue.enqueue(move || async move {
323                log.borrow_mut().push("first starts".to_string());
324                first_released.await;
325                log.borrow_mut().push("first ends".to_string());
326                Ok(())
327            });
328        }
329        {
330            let log = log.clone();
331            queue.enqueue(move || async move {
332                log.borrow_mut().push("second".to_string());
333                Ok(())
334            });
335        }
336
337        Dispatcher::ui_thread().run_jobs(None);
338        // The second call waits for the first.
339        assert_eq!(*log.borrow(), ["first starts"]);
340
341        release_first.try_set(());
342        Dispatcher::ui_thread().run_jobs(None);
343        assert_eq!(*log.borrow(), ["first starts", "first ends", "second"]);
344    }
345
346    #[test]
347    fn a_failure_nobody_waits_for_goes_to_the_error_handler_and_the_queue_goes_on() {
348        let _scope = crate::test_support::scope();
349        let log = Rc::new(RefCell::new(Vec::new()));
350        let queue = queue_with_log(&log);
351
352        queue.enqueue(|| async { Err(failure("lost")) });
353        {
354            let log = log.clone();
355            queue.enqueue(move || async move {
356                log.borrow_mut().push("next".to_string());
357                Ok(())
358            });
359        }
360
361        Dispatcher::ui_thread().run_jobs(None);
362        assert_eq!(log.borrow().len(), 2);
363        assert!(log.borrow()[0].starts_with("error handler: "), "{:?}", log.borrow());
364        assert!(log.borrow()[0].contains("lost"));
365        assert_eq!(log.borrow()[1], "next");
366    }
367
368    #[test]
369    fn an_awaited_call_gets_its_result_or_its_failure_and_not_the_error_handler() {
370        let _scope = crate::test_support::scope();
371        let log = Rc::new(RefCell::new(Vec::new()));
372        let queue = queue_with_log(&log);
373
374        let done = queue.enqueue_async(|| async { Ok(()) });
375        let value = queue.enqueue_async_with_result(|| async { Ok(42) });
376        let failed = queue.enqueue_async_with_result(|| async { Err::<i32, _>(failure("broken")) });
377
378        let results = Rc::new(RefCell::new(None));
379        {
380            let results = results.clone();
381            drop(Dispatcher::ui_thread().invoke_async_task_local(move || async move {
382                let outcome = (done.await, value.await, failed.await);
383                *results.borrow_mut() = Some(outcome);
384            }));
385        }
386        pump_until(|| results.borrow().is_some());
387
388        let (done, value, failed) = results.borrow_mut().take().unwrap();
389        assert!(done.is_ok());
390        assert_eq!(value.unwrap(), 42);
391        assert!(matches!(failed, Err(DBusCallError::DBus(_))));
392        assert!(log.borrow().is_empty());
393    }
394
395    #[test]
396    fn fail_all_cancels_what_is_queued_behind_the_call_in_flight() {
397        let _scope = crate::test_support::scope();
398        let log = Rc::new(RefCell::new(Vec::new()));
399        let queue = queue_with_log(&log);
400        let (release_first, first_released) = completion::<()>();
401
402        queue.enqueue(move || async move {
403            first_released.await;
404            Ok(())
405        });
406        let waiting = queue.enqueue_async_with_result(|| async { Ok(1) });
407        {
408            let log = log.clone();
409            queue.enqueue(move || async move {
410                log.borrow_mut().push("never".to_string());
411                Ok(())
412            });
413        }
414        Dispatcher::ui_thread().run_jobs(None);
415
416        queue.fail_all();
417        release_first.try_set(());
418
419        let result = Rc::new(RefCell::new(None));
420        {
421            let result = result.clone();
422            drop(Dispatcher::ui_thread().invoke_async_task_local(move || async move {
423                *result.borrow_mut() = Some(waiting.await);
424            }));
425        }
426        pump_until(|| result.borrow().is_some());
427
428        assert!(matches!(result.borrow_mut().take().unwrap(), Err(DBusCallError::Canceled)));
429        assert!(log.borrow().is_empty());
430    }
431}
ferroui_freedesktop/dbus_call_queue.rs

ferroui_freedesktop/
dbus_call_queue.rs
