prosa/event/
pending.rs

1use std::{cmp::Ordering, marker::PhantomData, ops::Add, time::Duration};
2
3use prosa_utils::{
4    hash::{BuildIntHasher, IntHashMap},
5    msg::tvf::Tvf,
6};
7use tokio::time::{Instant, Sleep, sleep_until};
8
9use crate::core::msg::Msg;
10
11/// Pending timer use to track timeout with timer ID, and their associate timeout
12#[derive(Debug)]
13struct PendingTimer<T>
14where
15    T: Copy,
16{
17    timer_id: T,
18    timeout: Instant,
19}
20
21impl<T> PendingTimer<T>
22where
23    T: Copy,
24{
25    /// Method to create a new pending timer from an id and a duration
26    pub(crate) fn new(timer_id: T, timeout_duration: Duration) -> PendingTimer<T> {
27        PendingTimer {
28            timer_id,
29            timeout: Instant::now().add(timeout_duration),
30        }
31    }
32
33    /// Method to create a new pending timer from an id and an instant
34    pub(crate) fn new_at(timer_id: T, timeout: Instant) -> PendingTimer<T> {
35        PendingTimer { timer_id, timeout }
36    }
37
38    /// Getter of the timer id (object link to the timer)
39    pub(crate) fn get_timer_id(&self) -> T {
40        self.timer_id
41    }
42
43    /// Method to know if the timer is already expire
44    pub(crate) fn is_expired(&self) -> bool {
45        self.timeout <= Instant::now()
46    }
47
48    /// Method to get a Tokio Sleep object to wait on
49    pub(crate) fn sleep(&self) -> Sleep {
50        sleep_until(self.timeout)
51    }
52}
53
54impl<T> Ord for PendingTimer<T>
55where
56    T: Copy,
57{
58    fn cmp(&self, other: &Self) -> Ordering {
59        self.timeout.cmp(&other.timeout)
60    }
61}
62
63impl<T> PartialOrd for PendingTimer<T>
64where
65    T: Copy,
66{
67    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
68        Some(self.cmp(other))
69    }
70}
71
72impl<T> PartialEq for PendingTimer<T>
73where
74    T: Copy,
75{
76    fn eq(&self, other: &Self) -> bool {
77        self.timeout == other.timeout
78    }
79}
80
81impl<T> Eq for PendingTimer<T> where T: Copy {}
82
83/// ProSA pending timer to have a timer list
84/// This object is not thread safe, you must use it within the same Tokio thread
85///
86/// ```
87/// use std::time::Duration;
88/// use prosa::event::pending::Timers;
89///
90/// async fn processing() {
91///     let mut pending_timer: Timers<u64> = Default::default();
92///     tokio::select! {
93///         Some(timer_id) = pending_timer.pull(), if !pending_timer.is_empty() => {
94///             println!("Timer {:?}", timer_id);
95///             // Do your processing
96///         },
97///     }
98/// }
99/// ```
100#[derive(Debug, Default)]
101pub struct Timers<T>
102where
103    T: Copy,
104{
105    timers: Vec<PendingTimer<T>>,
106}
107
108impl<T> Timers<T>
109where
110    T: Copy,
111{
112    /// Returns the number of pending timers, also referred to as its ‘length’.
113    pub fn len(&self) -> usize {
114        self.timers.len()
115    }
116
117    /// Returns the capacity of the internal timer list.
118    pub fn capacity(&self) -> usize {
119        self.timers.capacity()
120    }
121
122    /// Returns true if there is no pending timer
123    pub fn is_empty(&self) -> bool {
124        self.timers.is_empty()
125    }
126
127    /// Method to create a new pending timer with a specific capacity
128    pub fn with_capacity(capacity: usize) -> Self {
129        Timers {
130            timers: Vec::with_capacity(capacity),
131        }
132    }
133
134    /// Method to push a pending timer
135    fn push_timer(&mut self, timer: PendingTimer<T>) {
136        let mut timer_iter = self.timers.iter();
137        let index = loop {
138            if let Some(val) = timer_iter.next() {
139                if timer > *val {
140                    break self.timers.len() - (timer_iter.count() + 1);
141                }
142            } else {
143                break self.timers.len();
144            }
145        };
146
147        self.timers.insert(index, timer);
148    }
149
150    /// Method to push a pending timer with a specifc timeout duration
151    pub fn push(&mut self, timer_id: T, timeout_duration: Duration) {
152        self.push_timer(PendingTimer::new(timer_id, timeout_duration));
153    }
154
155    /// Method to push a pending timer with a specifc timeout
156    pub fn push_at(&mut self, timer_id: T, timeout: Instant) {
157        self.push_timer(PendingTimer::new_at(timer_id, timeout));
158    }
159
160    /// Method to wait for the first timer
161    /// If there is no pending timer (`is_empty` == `true`) the method return immediatelly. It doesn't block until a timer is pending
162    ///
163    /// ```
164    /// use std::time::Duration;
165    /// use prosa::event::pending::Timers;
166    ///
167    /// async fn processing() {
168    ///     let mut pending_timer: Timers<u64> = Default::default();
169    ///     let mut timer_id: Option<u64> = pending_timer.pull().await;
170    ///     assert!(timer_id.is_none());
171    ///     pending_timer.push(1, Duration::from_millis(200));
172    ///     timer_id = pending_timer.pull().await;
173    ///     assert!(timer_id.is_some());
174    /// }
175    /// ```
176    pub async fn pull(&mut self) -> Option<T> {
177        if let Some(timer) = self.timers.last() {
178            if !timer.is_expired() {
179                timer.sleep().await;
180            }
181
182            self.timers.pop().map(|t| t.get_timer_id())
183        } else {
184            None
185        }
186    }
187
188    /// Retains only the elements specified by the predicate.
189    ///
190    /// ```
191    /// use std::time::Duration;
192    /// use prosa::event::pending::Timers;
193    ///
194    /// async fn processing() {
195    ///     let mut pending_timer: Timers<u64> = Default::default();
196    ///     pending_timer.push(1, Duration::from_secs(1));
197    ///     pending_timer.push(2, Duration::from_secs(2));
198    ///     pending_timer.push(3, Duration::from_secs(3));
199    ///     pending_timer.push(4, Duration::from_secs(4));
200    ///     assert_eq!(4, pending_timer.len());
201    ///     pending_timer.retain(|x| x % 2 == 0);
202    ///     assert_eq!(2, pending_timer.len());
203    /// }
204    /// ```
205    pub fn retain<F>(&mut self, mut f: F)
206    where
207        F: FnMut(T) -> bool,
208    {
209        self.timers.retain(|t| f(t.timer_id));
210    }
211
212    /// Method to pop the inner Pending timer of the timer list
213    fn pop(&mut self) -> Option<PendingTimer<T>> {
214        self.timers.pop()
215    }
216
217    /// Method to get a reference on the last pending timer or None if the list is empty
218    fn last(&self) -> Option<&PendingTimer<T>> {
219        self.timers.last()
220    }
221}
222
223/// ProSA pending message to keep track of the message and trigger a timeout if a message is expire
224/// This object is not thread safe, you must use it within the same Tokio thread
225///
226/// ```
227/// use std::time::Duration;
228/// use prosa::event::pending::PendingMsgs;
229/// use tokio::sync::mpsc::Receiver;
230/// use prosa::core::msg::{Msg, RequestMsg, InternalMsg};
231/// use prosa_utils::msg::simple_string_tvf::SimpleStringTvf;
232///
233/// async fn processing(mut queue: Receiver<InternalMsg<SimpleStringTvf>>) {
234///     let mut pending_msg: PendingMsgs<RequestMsg<SimpleStringTvf>, SimpleStringTvf> = Default::default();
235///     tokio::select! {
236///         Some(msg) = queue.recv() => {
237///             match msg {
238///                 InternalMsg::Request(msg) => {
239///                     // Push in the pending message, the message will wait a timeout of 200ms
240///                     pending_msg.push(msg, Duration::from_millis(200));
241///                 },
242///                 InternalMsg::Response(msg) => {
243///                     let original_request: Option<RequestMsg<SimpleStringTvf>> = pending_msg.pull_msg(msg.get_id());
244///                     println!("Receive a response: {:?}, from original request {:?}", msg, original_request);
245///                 },
246///                 _ => {},
247///             }
248///         },
249///         Some(msg) = pending_msg.pull(), if !pending_msg.is_empty() => {
250///             println!("Timeout message {:?}", msg);
251///             // Do your processing
252///         },
253///     }
254/// }
255/// ```
256#[derive(Debug)]
257pub struct PendingMsgs<T, M>
258where
259    T: Msg<M>,
260    M: Sized + Clone + Tvf,
261{
262    pending_messages: IntHashMap<u64, T>,
263    timers: Timers<u64>,
264    phantom: PhantomData<M>,
265}
266
267impl<T, M> PendingMsgs<T, M>
268where
269    T: Msg<M>,
270    M: Sized + Clone + Tvf,
271{
272    /// Returns the number of pending messages, also referred to as its ‘length’.
273    pub fn len(&self) -> usize {
274        self.pending_messages.len()
275    }
276
277    /// Returns the capacity of the internal message map.
278    pub fn capacity(&self) -> usize {
279        self.pending_messages.capacity()
280    }
281
282    /// Returns true if there is no pending message
283    pub fn is_empty(&self) -> bool {
284        self.pending_messages.is_empty()
285    }
286
287    /// Method to create a new pending message list with a specific capacity
288    pub fn with_capacity(capacity: usize) -> Self
289    where
290        T: Msg<M>,
291        M: Sized + Clone + Tvf,
292    {
293        PendingMsgs {
294            pending_messages: IntHashMap::with_capacity_and_hasher(
295                capacity,
296                BuildIntHasher::default(),
297            ),
298            timers: Timers::with_capacity(capacity),
299            phantom: PhantomData,
300        }
301    }
302
303    /// Method to push a pending message
304    pub fn push(&mut self, msg: T, timeout: Duration) {
305        self.push_with_id(msg.get_id(), msg, timeout);
306    }
307
308    /// Method to push a pending message with a custom id
309    pub fn push_with_id(&mut self, id: u64, msg: T, timeout: Duration) {
310        self.timers.push(id, timeout);
311        self.pending_messages.insert(id, msg);
312    }
313
314    /// Method to pull a pending message to process it
315    pub fn pull_msg(&mut self, msg_id: u64) -> Option<T> {
316        if let Some(msg) = self.pending_messages.remove(&msg_id) {
317            return Some(msg);
318        }
319
320        None
321    }
322
323    /// Method to wait for expired message (timeout)
324    /// If there is no pending message (`is_empty` == `true`) the method return immediatelly. It doesn't block until a message is pending
325    ///
326    /// ```
327    /// use std::time::Duration;
328    /// use tokio::sync::mpsc::Sender;
329    /// use prosa::event::pending::PendingMsgs;
330    /// use prosa::core::msg::{Msg, RequestMsg, InternalMsg};
331    /// use prosa_utils::msg::simple_string_tvf::SimpleStringTvf;
332    ///
333    /// async fn processing(tvf: SimpleStringTvf, queue: Sender<InternalMsg<SimpleStringTvf>>) {
334    ///     let mut pending_msg: PendingMsgs<RequestMsg<SimpleStringTvf>, SimpleStringTvf> = Default::default();
335    ///     let mut msg: Option<RequestMsg<SimpleStringTvf>> = pending_msg.pull().await;
336    ///     assert!(msg.is_none());
337    ///     pending_msg.push(RequestMsg::new(String::from("service"), tvf, queue), Duration::from_millis(200));
338    ///     tokio::select! {
339    ///         Some(msg) = pending_msg.pull(), if !pending_msg.is_empty() => {
340    ///             println!("Timeout message {:?}", msg);
341    ///         }
342    ///     }
343    /// }
344    /// ```
345    pub async fn pull(&mut self) -> Option<T> {
346        while let Some(timer) = self.timers.last() {
347            if self.pending_messages.contains_key(&timer.get_timer_id()) {
348                if !timer.is_expired() {
349                    timer.sleep().await;
350                }
351
352                let time = self.timers.pop()?;
353                return self.pull_msg(time.get_timer_id());
354            } else {
355                self.timers.pop();
356            }
357        }
358
359        None
360    }
361}
362
363impl<T, M> Default for PendingMsgs<T, M>
364where
365    T: Msg<M>,
366    M: Sized + Clone + Tvf,
367{
368    fn default() -> Self {
369        PendingMsgs::<T, M> {
370            pending_messages: Default::default(),
371            timers: Default::default(),
372            phantom: PhantomData,
373        }
374    }
375}
376
377#[cfg(test)]
378mod tests {
379    extern crate self as prosa;
380
381    use std::time::Duration;
382
383    use prosa_macros::{proc, settings};
384    use prosa_utils::msg::{simple_string_tvf::SimpleStringTvf, tvf::Tvf};
385    use serde::Serialize;
386    use tokio::time::timeout;
387
388    use crate::core::{
389        error::BusError,
390        main::{MainProc, MainRunnable},
391        msg::{InternalMsg, Msg, RequestMsg},
392        proc::{ProcBusParam, ProcConfig},
393    };
394
395    use super::{PendingMsgs, Timers};
396
397    #[proc]
398    pub(crate) struct TestProc {}
399
400    #[proc]
401    impl TestProc<SimpleStringTvf> {
402        async fn timers_run(&mut self) -> Result<(), BusError> {
403            // Add proc and its service
404            self.proc.add_proc().await?;
405            self.proc
406                .add_service_proc(vec![String::from("TEST")])
407                .await?;
408
409            let mut pending_timer: Timers<u64> = Default::default();
410            loop {
411                tokio::select! {
412                    Some(msg) = self.internal_rx_queue.recv() => {
413                        match msg {
414                            InternalMsg::Request(_) => {
415                                assert_eq!(0, pending_timer.len());
416                                pending_timer.push(1, Duration::from_millis(100));
417                                assert_eq!(1, pending_timer.len());
418                            },
419                            InternalMsg::Service(table) => {
420                                if let Some(service) = table.get_proc_service("TEST") {
421                                    service.proc_queue.send(InternalMsg::Request(RequestMsg::new(String::from("TEST"), Default::default(), self.proc.get_service_queue().clone()))).await.expect("Internal msg should be send");
422                                }
423                            },
424                            _ => return Err(BusError::ProcComm(self.get_proc_id(), 0, String::from("Wrong message"))),
425                        }
426                    },
427                    Some(timer_id) = pending_timer.pull(), if !pending_timer.is_empty() => {
428                        assert_eq!(0, pending_timer.len());
429                        assert_eq!(1, timer_id);
430                        self.proc.remove_proc(None).await?;
431                        return Ok(())
432                    },
433                }
434            }
435        }
436
437        async fn pending_msgs_run(&mut self) -> Result<(), BusError> {
438            // Add proc and its service
439            self.proc.add_proc().await?;
440            self.proc
441                .add_service_proc(vec![String::from("TEST")])
442                .await?;
443
444            let mut pending_msg: PendingMsgs<RequestMsg<SimpleStringTvf>, SimpleStringTvf> =
445                Default::default();
446            loop {
447                tokio::select! {
448                    Some(msg) = self.internal_rx_queue.recv() => {
449                        match msg {
450                            InternalMsg::Request(msg) => {
451                                assert_eq!(0, pending_msg.len());
452                                pending_msg.push(msg, Duration::from_millis(100));
453                                assert_eq!(1, pending_msg.len());
454                            },
455                            InternalMsg::Service(table) => {
456                                if let Some(service) = table.get_proc_service("TEST") {
457                                    let mut msg: SimpleStringTvf = Default::default();
458                                    msg.put_string(1, "good");
459                                    service.proc_queue.send(InternalMsg::Request(RequestMsg::new(String::from("TEST"), msg, self.proc.get_service_queue().clone()))).await.expect("Internal msg should be send");
460                                }
461                            },
462                            _ => return Err(BusError::ProcComm(self.get_proc_id(), 0, String::from("Wrong message"))),
463                        }
464                    },
465                    Some(msg) = pending_msg.pull(), if !pending_msg.is_empty() => {
466                        assert_eq!(0, pending_msg.len());
467                        assert_eq!(String::from("good"), msg.get_data()?.get_string(1)?.into_owned());
468                        self.proc.remove_proc(None).await?;
469                        return Ok(())
470                    },
471                }
472            }
473        }
474
475        pub(crate) async fn timers_timeout_run(&mut self) -> Result<(), BusError> {
476            if timeout(Duration::from_millis(200), self.timers_run())
477                .await
478                .is_err()
479            {
480                Err(BusError::InternalQueue(String::from(
481                    "Timer is not working",
482                )))
483            } else {
484                Ok(())
485            }
486        }
487
488        pub(crate) async fn pending_msgs_timeout_run(&mut self) -> Result<(), BusError> {
489            if timeout(Duration::from_millis(200), self.pending_msgs_run())
490                .await
491                .is_err()
492            {
493                Err(BusError::InternalQueue(String::from(
494                    "pending msgs is not working",
495                )))
496            } else {
497                Ok(())
498            }
499        }
500    }
501
502    #[test]
503    fn test_with_capacity() {
504        let capacity = 10;
505        let pending_msg: PendingMsgs<RequestMsg<SimpleStringTvf>, SimpleStringTvf> =
506            PendingMsgs::with_capacity(capacity);
507        assert_eq!(pending_msg.len(), 0);
508        assert!(pending_msg.is_empty());
509        assert!(pending_msg.capacity() >= capacity);
510
511        let pending_timer: Timers<u64> = Timers::with_capacity(capacity);
512        assert_eq!(pending_timer.len(), 0);
513        assert!(pending_timer.is_empty());
514        assert!(pending_timer.capacity() >= capacity);
515    }
516
517    #[tokio::test]
518    async fn test_pending() {
519        /// Dummy settings
520        #[settings]
521        #[derive(Default, Debug, Serialize)]
522        struct DummySettings {}
523
524        // Create bus and main processor
525        let (bus, main) = MainProc::<SimpleStringTvf>::create(&DummySettings::default(), Some(2));
526
527        // Launch the main task
528        let main_task = tokio::spawn(main.run());
529
530        // Launch the test processor
531        assert_eq!(
532            Ok(()),
533            TestProc::<SimpleStringTvf>::create_raw(1, "test1".to_string(), bus.clone())
534                .timers_timeout_run()
535                .await
536        );
537
538        assert_eq!(
539            Ok(()),
540            TestProc::<SimpleStringTvf>::create_raw(2, "test2".to_string(), bus.clone())
541                .pending_msgs_timeout_run()
542                .await
543        );
544
545        bus.stop("ProSA unit test end".into())
546            .await
547            .expect("ProSA should stop");
548        main_task.await.expect("Main task should end correctly");
549    }
550}
prosa/event/pending.rs

prosa/event/
pending.rs
