escouade_plugin_kit/
time.rs

1//! A plugin's timers, whatever the target: tokio's in process; in a module, the host's (`timer/set`, `timer/fired`,
2//! docs/PLUGINS.md §8.4.5), a task waiting for its timer as it waits for any line.
3
4use std::fmt;
5use std::future::Future;
6use std::time::Duration;
7
8/// A `timeout` that ran out.
9#[derive(Debug, Clone, Copy, PartialEq, Eq)]
10pub struct Elapsed;
11
12impl fmt::Display for Elapsed {
13    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
14        f.write_str("timed out")
15    }
16}
17
18impl std::error::Error for Elapsed {}
19
20/// Waits `delay`.
21#[cfg(not(target_family = "wasm"))]
22pub async fn sleep(delay: Duration) {
23    tokio::time::sleep(delay).await;
24}
25
26/// `work`, or `Elapsed` past `limit`; given up then, as when it is dropped.
27#[cfg(not(target_family = "wasm"))]
28pub async fn timeout<F: Future>(limit: Duration, work: F) -> Result<F::Output, Elapsed> {
29    tokio::time::timeout(limit, work).await.map_err(|_| Elapsed)
30}
31
32/// See the native `sleep`.
33#[cfg(target_family = "wasm")]
34pub async fn sleep(delay: Duration) {
35    if let Some(cx) = crate::serve::reactor::cx() {
36        sleep_on(&cx, delay).await;
37    }
38}
39
40/// See the native `timeout`.
41#[cfg(target_family = "wasm")]
42pub async fn timeout<F: Future>(limit: Duration, work: F) -> Result<F::Output, Elapsed> {
43    match crate::serve::reactor::cx() {
44        Some(cx) => timeout_on(&cx, limit, work).await,
45        None => Ok(work.await),
46    }
47}
48
49#[cfg(any(test, target_family = "wasm"))]
50pub(crate) use module::{sleep_on, timeout_on};
51
52#[cfg(any(test, target_family = "wasm"))]
53mod module {
54    use super::*;
55    use crate::serve::Cx;
56    use serde_json::{json, Value};
57    use std::future::poll_fn;
58    use std::pin::pin;
59    use std::task::Poll;
60
61    /// A timer of the host being waited for; let go before it rang, it is cancelled.
62    struct Ringing<'a> {
63        cx: &'a Cx,
64        timer: Value,
65        rung: bool,
66    }
67
68    impl Drop for Ringing<'_> {
69        fn drop(&mut self) {
70            if !self.rung {
71                self.cx
72                    .notify("timer/cancel", json!({ "timer": self.timer }));
73            }
74        }
75    }
76
77    /// Waits `delay` on a timer of the host. A host that refuses one (past its 64 timers) is said in the journal and the
78    /// sleep ends at once: a task that never ends would hold its call for ever.
79    pub(crate) async fn sleep_on(cx: &Cx, delay: Duration) {
80        let ms = delay.as_millis().min(u128::from(u64::MAX)) as u64;
81        let timer = match cx.request("timer/set", json!({ "ms": ms })).await {
82            Ok(v) => v.get("timer").cloned(),
83            Err(e) => {
84                cx.log("warn", &format!("no timer: {}", e.message));
85                return;
86            }
87        };
88        let Some(timer) = timer else { return };
89        let key = timer.to_string();
90        let mut ringing = Ringing {
91            cx,
92            timer,
93            rung: false,
94        };
95        poll_fn(|_| {
96            if cx.rang(&key) {
97                ringing.rung = true;
98                Poll::Ready(())
99            } else {
100                Poll::Pending
101            }
102        })
103        .await;
104    }
105
106    /// `work`, or `Elapsed` past `limit`, on a timer of the host.
107    pub(crate) async fn timeout_on<F: Future>(
108        cx: &Cx,
109        limit: Duration,
110        work: F,
111    ) -> Result<F::Output, Elapsed> {
112        let mut work = pin!(work);
113        let mut limit = pin!(sleep_on(cx, limit));
114        poll_fn(|task| {
115            if let Poll::Ready(done) = work.as_mut().poll(task) {
116                return Poll::Ready(Ok(done));
117            }
118            limit.as_mut().poll(task).map(|()| Err(Elapsed))
119        })
120        .await
121    }
122}
123
124#[cfg(all(test, not(target_family = "wasm")))]
125mod tests {
126    use super::*;
127
128    #[tokio::test(start_paused = true)]
129    async fn a_timeout_lets_through_what_ends_in_time_and_stops_the_rest() {
130        assert_eq!(timeout(Duration::from_secs(1), async { 7 }).await, Ok(7));
131        let slow = sleep(Duration::from_secs(5));
132        assert_eq!(timeout(Duration::from_secs(1), slow).await, Err(Elapsed));
133    }
134}
escouade_plugin_kit/time.rs

escouade_plugin_kit/
time.rs
