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

use core::{future::Future, pin::Pin, time::Duration};
use std::time::Instant;

use futures_util::{stream, Stream};

#[cfg(feature = "impl_async_io")]
pub mod impl_async_io;
#[cfg(feature = "impl_async_timer")]
pub mod impl_async_timer;
#[cfg(feature = "impl_tokio")]
pub mod impl_tokio;

//
pub trait Intervalable {
    fn interval(dur: Duration) -> Self;

    fn wait<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Option<Instant>> + Send + 'a>>;
}

/// [Ref](https://docs.rs/futures-util/0.3.21/futures_util/stream/fn.iter.html)
pub fn intervalable_iter_stream<I, INTVL>(iter: I, interval: INTVL) -> impl Stream<Item = I::Item>
where
    I: IntoIterator,
    INTVL: Intervalable,
{
    stream::unfold(
        (iter.into_iter(), interval),
        |(mut iter, mut interval)| async move {
            if let Some(item) = iter.next() {
                interval.wait().await;
                Some((item, (iter, interval)))
            } else {
                None
            }
        },
    )
}

/// [Ref](https://docs.rs/futures-util/0.3.21/futures_util/stream/fn.repeat.html)
pub fn intervalable_repeat_stream<T, INTVL>(item: T, interval: INTVL) -> impl Stream<Item = T>
where
    T: Clone,
    INTVL: Intervalable,
{
    stream::unfold((item, interval), |(item, mut interval)| async move {
        interval.wait().await;
        Some((item.clone(), (item, interval)))
    })
}

/// [Ref](https://docs.rs/futures-util/0.3.21/futures_util/stream/fn.repeat_with.html)
pub fn intervalable_repeat_with_stream<A, F, INTVL>(
    repeater: F,
    interval: INTVL,
) -> impl Stream<Item = A>
where
    F: FnMut() -> A,
    INTVL: Intervalable,
{
    stream::unfold(
        (repeater, interval),
        |(mut repeater, mut interval)| async move {
            interval.wait().await;
            Some((repeater(), (repeater, interval)))
        },
    )
}