zbus 6.0.0

API for D-Bus communication and its wire format
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

use std::{
    future::{Future, poll_fn},
    io::ErrorKind,
    pin::pin,
    task::Poll,
    time::Duration,
};

use super::{Runtime, traits};
use crate::{Error, Result};

impl Runtime {
    /// Sleeps for `duration` on this runtime's timer.
    pub(crate) async fn sleep(&self, duration: Duration) {
        match self {
            #[cfg(feature = "default-rt")]
            Self::ZRuntime(runtime) => traits::Runtime::sleep(runtime, duration).await,
            #[cfg(feature = "tokio")]
            Self::Tokio(runtime) => traits::Runtime::sleep(runtime, duration).await,
            Self::External(runtime) => traits::Runtime::sleep(runtime, duration).await,
        }
    }

    /// Awaits `fut`, failing with a timed-out error once `duration` has passed.
    pub(crate) async fn timeout<F, T>(&self, fut: F, duration: Duration) -> Result<T>
    where
        F: Future<Output = Result<T>>,
    {
        let mut fut = pin!(fut);
        let mut sleep = pin!(self.sleep(duration));

        // `fut` is polled first, so that an outcome it has by the time the sleep ends still wins.
        poll_fn(|cx| {
            if let Poll::Ready(output) = fut.as_mut().poll(cx) {
                return Poll::Ready(output);
            }

            sleep.as_mut().poll(cx).map(|()| {
                Err(Error::from(std::io::Error::new(
                    ErrorKind::TimedOut,
                    "timed out",
                )))
            })
        })
        .await
    }
}

#[cfg(all(test, feature = "tokio"))]
mod tests {
    use std::pin::pin;

    use futures_util::poll;
    use ntest::timeout;

    use super::*;
    use crate::runtime::Tokio;

    /// A timeout is measured on the runtime's clock rather than the standard one.
    ///
    /// Tokio's clock is paused here and pushed a minute ahead of where it started, which is far
    /// enough that a deadline worked out from `Instant::now` on the standard clock would already
    /// have passed before the call was even made.
    #[tokio::test(start_paused = true)]
    #[timeout(15000)]
    async fn a_tokio_timeout_expires_on_the_tokio_clock() {
        tokio::time::advance(Duration::from_secs(60)).await;
        let runtime = Runtime::Tokio(Tokio::current().expect("a Tokio runtime is current"));
        let mut timing_out =
            pin!(runtime.timeout(std::future::pending::<Result<()>>(), Duration::from_secs(1)));

        // The first poll is what arms the timer, so the clock only moves once it is running.
        assert!(poll!(timing_out.as_mut()).is_pending());
        tokio::time::advance(Duration::from_millis(500)).await;
        assert!(poll!(timing_out.as_mut()).is_pending());

        tokio::time::advance(Duration::from_millis(600)).await;
        let error = timing_out.await.unwrap_err();
        assert!(
            matches!(&error, Error::InputOutput(e) if e.kind() == ErrorKind::TimedOut),
            "got {error:?}",
        );
    }
}