rx_rust/operators/creating/
interval.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::{Observable, Subscription},
4    observer::Observer,
5    scheduler::Scheduler,
6};
7use educe::Educe;
8use std::{convert::Infallible, time::Duration};
9
10/// Creates an Observable that emits a sequence of integers spaced by a given time interval.
11/// See <https://reactivex.io/documentation/operators/interval.html>
12///
13/// # Examples
14/// ```rust
15/// # #[cfg(not(feature = "tokio-scheduler"))]
16/// # fn main() {}
17/// # #[cfg(feature = "tokio-scheduler")]
18/// #[tokio::main]
19/// async fn main() {
20///     use rx_rust::{
21///         observable::ObservableExt,
22///         observer::Termination,
23///         operators::creating::interval::Interval,
24///     };
25///     use std::sync::{Arc, Mutex};
26///     use std::time::Duration;
27///     use tokio::time::sleep;
28///
29///     let handle = tokio::runtime::Handle::current();
30///     let values = Arc::new(Mutex::new(Vec::new()));
31///     let terminations = Arc::new(Mutex::new(Vec::new()));
32///     let values_observer = Arc::clone(&values);
33///     let terminations_observer = Arc::clone(&terminations);
34///     let subscription = Interval::new(Duration::from_millis(1), handle, None)
35///         .take(3)
36///         .subscribe_with_callback(
37///             move |value| values_observer.lock().unwrap().push(value),
38///             move |termination| terminations_observer
39///                 .lock()
40///                 .unwrap()
41///                 .push(termination),
42///         );
43///
44///     sleep(Duration::from_millis(10)).await;
45///     drop(subscription);
46///
47///     assert_eq!(&*values.lock().unwrap(), &[0, 1, 2]);
48///     assert_eq!(
49///         &*terminations.lock().unwrap(),
50///         &[Termination::Completed]
51///     );
52/// }
53/// ```
54#[derive(Educe)]
55#[educe(Debug, Clone)]
56pub struct Interval<S> {
57    period: Duration,
58    scheduler: S,
59    delay: Option<Duration>,
60}
61
62impl<S> Interval<S> {
63    pub fn new(period: Duration, scheduler: S, delay: Option<Duration>) -> Self {
64        Self {
65            period,
66            scheduler,
67            delay,
68        }
69    }
70}
71
72impl<S> Observable<'static, usize, Infallible> for Interval<S>
73where
74    S: Scheduler + Clone + MaybeSend + 'static,
75{
76    type D = S::D;
77
78    fn subscribe(
79        self,
80        mut observer: impl Observer<usize, Infallible> + MaybeSend + 'static,
81    ) -> Subscription<Self::D> {
82        self.scheduler.schedule_periodically(
83            // The callback's answer is what keeps the schedule running, so an observer that
84            // stopped ends it: nothing is completed, since an interval never completes anyway.
85            move |count| observer.on_next(count).is_continue(),
86            self.period,
87            self.delay,
88        )
89    }
90}
rx_rust/operators/creating/interval.rs

rx_rust/operators/creating/
interval.rs
