broker_tokio/time/
interval.rs

1use crate::future::poll_fn;
2use crate::time::{delay_until, Delay, Duration, Instant};
3
4use std::future::Future;
5use std::pin::Pin;
6use std::task::{Context, Poll};
7
8/// Creates new `Interval` that yields with interval of `duration`. The first
9/// tick completes immediately.
10///
11/// An interval will tick indefinitely. At any time, the `Interval` value can be
12/// dropped. This cancels the interval.
13///
14/// This function is equivalent to `interval_at(Instant::now(), period)`.
15///
16/// # Panics
17///
18/// This function panics if `period` is zero.
19///
20/// # Examples
21///
22/// ```
23/// use tokio::time::{self, Duration};
24///
25/// #[tokio::main]
26/// async fn main() {
27///     let mut interval = time::interval(Duration::from_millis(10));
28///
29///     interval.tick().await;
30///     interval.tick().await;
31///     interval.tick().await;
32///
33///     // approximately 20ms have elapsed.
34/// }
35/// ```
36pub fn interval(period: Duration) -> Interval {
37    assert!(period > Duration::new(0, 0), "`period` must be non-zero.");
38
39    interval_at(Instant::now(), period)
40}
41
42/// Creates new `Interval` that yields with interval of `period` with the
43/// first tick completing at `at`.
44///
45/// An interval will tick indefinitely. At any time, the `Interval` value can be
46/// dropped. This cancels the interval.
47///
48/// # Panics
49///
50/// This function panics if `period` is zero.
51///
52/// # Examples
53///
54/// ```
55/// use tokio::time::{interval_at, Duration, Instant};
56///
57/// #[tokio::main]
58/// async fn main() {
59///     let start = Instant::now() + Duration::from_millis(50);
60///     let mut interval = interval_at(start, Duration::from_millis(10));
61///
62///     interval.tick().await;
63///     interval.tick().await;
64///     interval.tick().await;
65///
66///     // approximately 70ms have elapsed.
67/// }
68/// ```
69pub fn interval_at(start: Instant, period: Duration) -> Interval {
70    assert!(period > Duration::new(0, 0), "`period` must be non-zero.");
71
72    Interval {
73        delay: delay_until(start),
74        period,
75    }
76}
77
78/// Stream returned by [`interval`](interval) and [`interval_at`](interval_at).
79#[derive(Debug)]
80pub struct Interval {
81    /// Future that completes the next time the `Interval` yields a value.
82    delay: Delay,
83
84    /// The duration between values yielded by `Interval`.
85    period: Duration,
86}
87
88impl Interval {
89    #[doc(hidden)] // TODO: document
90    pub fn poll_tick(&mut self, cx: &mut Context<'_>) -> Poll<Instant> {
91        // Wait for the delay to be done
92        ready!(Pin::new(&mut self.delay).poll(cx));
93
94        // Get the `now` by looking at the `delay` deadline
95        let now = self.delay.deadline();
96
97        // The next interval value is `duration` after the one that just
98        // yielded.
99        let next = now + self.period;
100        self.delay.reset(next);
101
102        // Return the current instant
103        Poll::Ready(now)
104    }
105
106    /// Completes when the next instant in the interval has been reached.
107    ///
108    /// # Examples
109    ///
110    /// ```
111    /// use tokio::time;
112    ///
113    /// use std::time::Duration;
114    ///
115    /// #[tokio::main]
116    /// async fn main() {
117    ///     let mut interval = time::interval(Duration::from_millis(10));
118    ///
119    ///     interval.tick().await;
120    ///     interval.tick().await;
121    ///     interval.tick().await;
122    ///
123    ///     // approximately 20ms have elapsed.
124    /// }
125    /// ```
126    #[allow(clippy::should_implement_trait)] // TODO: rename (tokio-rs/tokio#1261)
127    pub async fn tick(&mut self) -> Instant {
128        poll_fn(|cx| self.poll_tick(cx)).await
129    }
130}
131
132#[cfg(feature = "stream")]
133impl crate::stream::Stream for Interval {
134    type Item = Instant;
135
136    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Instant>> {
137        Poll::Ready(Some(ready!(self.poll_tick(cx))))
138    }
139}
broker_tokio/time/interval.rs

broker_tokio/time/
interval.rs
