rx_rust/operators/creating/
timer.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::{Observable, Subscription},
4    observer::{Observer, Termination},
5    scheduler::Scheduler,
6};
7use educe::Educe;
8use std::{convert::Infallible, time::Duration};
9
10/// Creates an Observable that emits a single item after a given delay.
11/// See <https://reactivex.io/documentation/operators/timer.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::timer::Timer,
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///
35///     let subscription = Timer::new("tick", Duration::from_millis(5), handle)
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(), &["tick"]);
48///     assert_eq!(
49///         &*terminations.lock().unwrap(),
50///         &[Termination::Completed]
51///     );
52/// }
53/// ```
54#[derive(Educe)]
55#[educe(Debug, Clone)]
56pub struct Timer<T, S> {
57    value: T,
58    delay: Duration,
59    scheduler: S,
60}
61
62impl<T, S> Timer<T, S> {
63    pub fn new(value: T, delay: Duration, scheduler: S) -> Self {
64        Self {
65            value,
66            delay,
67            scheduler,
68        }
69    }
70}
71
72impl<T, S> Observable<'static, T, Infallible> for Timer<T, S>
73where
74    T: MaybeSend + 'static,
75    S: Scheduler,
76{
77    type D = S::D;
78
79    fn subscribe(
80        self,
81        mut observer: impl Observer<T, Infallible> + MaybeSend + 'static,
82    ) -> Subscription<Self::D> {
83        self.scheduler.schedule(
84            || {
85                if observer.on_next(self.value).is_continue() {
86                    observer.on_termination(Termination::Completed);
87                }
88            },
89            Some(self.delay),
90        )
91    }
92}
rx_rust/operators/creating/timer.rs

rx_rust/operators/creating/
timer.rs
