timer_deque_rs/

timer_consumer.rs

/*-
 * timer-rs - a Rust crate which provides timer and timer queues based on target OS
 *  functionality.
 * 
 * Copyright (C) 2025 Aleksandr Morozov
 * 
 * The timer-rs crate can be redistributed and/or modified
 * under the terms of either of the following licenses:
 *
 *   1. the Mozilla Public License Version 2.0 (the “MPL”) OR
 *                     
 *   2. EUROPEAN UNION PUBLIC LICENCE v. 1.2 EUPL © the European Union 2007, 2016
 */

use std::fmt;

use crate::error::{TimerError, TimerErrorType, TimerResult};
use crate::timer::{OrderedTimerDeque, OrderedTimerDequeIntrf};
use crate::{common, timer_err, TimerReadRes};
use crate::timer_portable::timer::TimerExpMode;


/// Defines the type of the queue. This type of the queue consumes the 
/// instance for which the timer is set. The item must be [Send] and can
/// be used with, for example, [std::sync::Arc]. This is convinient when it
/// is required to store the instance in the timer and retrive it when the 
/// timeout happens. It would allow to avoid looking for the instance in the 
/// lists.
/// 
/// # Generics
/// 
/// `R` - an instance which should be stored on the timer dequeue. The instance
///     must implement [PartialEq], [Eq], [fmt::Debug], [fmt::Display], [Send].
#[derive(Debug)]
pub struct TimerDequeueConsumer<R: PartialEq + Eq + fmt::Debug + fmt::Display + Send>
{
    /// An internal data
    target: R,

    /// An absolute timestamp in seconds (UTC time) sets the timer.
    timeout_absolute: TimerExpMode,
}

impl<R> fmt::Display 
for TimerDequeueConsumer<R>
where R: PartialEq + Eq + fmt::Display + fmt::Debug + Send
{  
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> 
    { 
        write!(f, "{} until: {}", self.target, self.timeout_absolute)
    }
}


impl<R: PartialEq + Eq + fmt::Display + fmt::Debug + Send> TimerDequeueConsumer<R>
{ 
    fn new(target: R, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<Self>
    {
        let abs_time_nsec = abs_time_nsec & 999_999_999;

        let cur = common::get_current_timestamp();
        let req = TimerExpMode::OneShot { sec: abs_time_sec, nsec: abs_time_nsec };

        if TimerExpMode::from(cur) > req
        {
            timer_err!(TimerErrorType::Expired, "time already expired");
        }

        return Ok(
            Self
            {
                target: 
                    target,
                timeout_absolute: 
                    req
            }
        );
    }

    fn get_time_until(&self) -> TimerExpMode
    {
        return self.timeout_absolute;
    }

    fn into_inner(self) -> R
    {
        return self.target;
    }
}

impl<R: PartialEq + Eq + fmt::Debug + fmt::Display + Send> Eq for TimerDequeueConsumer<R> {}

impl<R: PartialEq + Eq + fmt::Display + fmt::Debug + Send> PartialEq for TimerDequeueConsumer<R> 
{
    fn eq(&self, other: &Self) -> bool 
    {
        return self.target == other.target;
    }
}

impl<R: PartialEq + Eq + fmt::Display + fmt::Debug + Send> PartialEq<R> for TimerDequeueConsumer<R> 
{
    fn eq(&self, other: &R) -> bool 
    {
        return &self.target == other;
    }
}

impl<R: PartialEq + Eq + fmt::Display + fmt::Debug + Send> Ord for TimerDequeueConsumer<R>
{
    fn cmp(&self, other: &Self) -> std::cmp::Ordering 
    {
        return self.timeout_absolute.cmp(&other.timeout_absolute);
    }
}

impl<R: PartialEq + Eq + fmt::Display + fmt::Debug + Send> PartialOrd for TimerDequeueConsumer<R>
{
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> 
    {
        return Some(self.cmp(other));
    }
}

impl<R> OrderedTimerDequeIntrf for TimerDequeueConsumer<R>
where R: PartialEq + Eq + fmt::Display + fmt::Debug + Send
{
    /// Input item.
    type Target = R;

    /// Nothing is returned.
    type Ticket = common::NoTicket;

    #[inline]
    fn wrap(target: Self::Target, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<(Self, Self::Ticket)> 
    {
        return Ok((Self::new(target, abs_time_sec, abs_time_nsec)?, common::NoTicket));
    }

    #[inline]
    fn get_timeout_absolute(&self) -> TimerExpMode 
    {
        return self.timeout_absolute;
    }
}

impl<R> OrderedTimerDeque<TimerDequeueConsumer<R>>
where R: PartialEq + Eq + fmt::Display + fmt::Debug + Send
{
    /// Adds the new absolute timeout to the timer dequeue instance.
    /// 
    /// The `entity` is an item which should be stored in the dequeue.
    /// 
    /// # Arguemnts
    /// 
    /// `entity` - `R` generic which is an item to store in the dequeu.
    /// 
    /// `abs_time_sec` - a [i64] absolute time in seconds. Can not be 0.
    /// 
    /// `abs_time_nsec` - a [i64] asolute time in nanosecodns. Can be 0.
    /// 
    /// # Returns
    /// 
    /// A [Result] as alias [TimerResult] is returned with:
    /// 
    /// * [Result::Ok] with the [common::NoTicket] ticket which is dummy value.
    /// 
    /// * [Result::Err] with error description.
    pub   
    fn add_to_timer(&mut self, entity: R, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<common::NoTicket>
    {
        return self.add_to_timer_local(entity, abs_time_sec, abs_time_nsec);
    }

    /// Removes the instance from the queue by the reference to entity.
    /// 
    /// # Arguments
    /// 
    /// `entity` - `R` reference to added item which will be used to identify the
    ///     item ont he record.
    /// 
    /// # Returns
    /// 
    /// A [Result] as alias [TimerResult] is returned with:
    /// 
    /// * [Result::Ok] with the inner type [Option] where
    ///     * [Option::Some] is returned with the consumed `entity`.
    ///     * [Option::None] is returned when item was not found.
    /// 
    /// * [Result::Err] with error description.
    pub 
    fn remove_from_sched_queue(&mut self, entity: &R) -> TimerResult<Option<R>>
    {
        if self.deque_timeout_list.len() == 0
        {
            timer_err!(TimerErrorType::QueueEmpty, "queue list is empty!");
        }
        else
        {
            // search for the item in list
            if self.deque_timeout_list.len() == 1
            {
                // just pop the from the front
                let ret_ent = self.deque_timeout_list.pop_front().unwrap();

                // stop timer
                self.stop_timer()?;

                return Ok(Some(ret_ent.into_inner()));
            }
            else
            {
                // in theory the `ticket` is a reference to ARC, so the weak should be upgraded 
                // succesfully for temoved instance.

                for (pos, q_item) 
                in self.deque_timeout_list.iter().enumerate()
                {
                    if q_item == entity
                    {
                        // remove by the index
                        let ret_ent = 
                            self.deque_timeout_list.remove(pos).unwrap().into_inner();

                        // call timer reset if index is 0 (front)
                        if pos == 0 
                        {
                            self.reschedule_timer()?;
                        }

                        return Ok(Some(ret_ent));
                    }
                }

                return Ok(None);
            }
        }
    }

    /// Handles the event which was `read` from the timer. The functions [Self::wait_for_event]
    /// or [Self::poll] can be used to obtain the event.
    /// 
    /// # Arguments
    /// 
    /// `res` - [TimerReadRes] an event from the timer to handle.
    /// 
    /// `timeout_items` - a vector of an enitities for which timeout have happened.
    /// 
    /// A [Result] as alias [TimerResult] is returned with:
    /// 
    /// * [Result::Ok] witout any innder data.
    /// 
    /// * [Result::Err] with error description.
    pub 
    fn timeout_event_handler(&mut self, res: TimerReadRes<u64>, timeout_items: &mut Vec<R>) -> TimerResult<()>
    {
        // ignore wouldblock
        if let TimerReadRes::WouldBlock = res
        {
            return Ok(());
        }
        
        // check if the time reset is required
        let force_timer_resched = 
            TimerReadRes::Cancelled == res || TimerReadRes::Ok(0) == res;

        let mut timeouts: Vec<R> = Vec::new();
        let mut error: Option<TimerError> = None;

        let cur_timestamp = common::get_current_timestamp();

        loop 
        {
            // get from front of the queue
            let Some(front_entity) = 
                self.deque_timeout_list.front() 
                else 
                { 
                    let Err(e) = self.stop_timer()
                    else
                    {
                        break;
                    };

                    error = Some(e);

                    break;
                };

            let time_until = front_entity.get_time_until();

            if time_until <= TimerExpMode::from(cur_timestamp)
            {
                timeouts.push(self.deque_timeout_list.pop_front().unwrap().into_inner());
            }
            else
            {
                if timeouts.is_empty() == false || force_timer_resched == true
                {
                    // call timer reshedule
                    self.reschedule_timer()?;
                }

                break;
            }
        }

        *timeout_items = timeouts;

        return error.map_or(Ok(()), |f| Err(f));
    }

}

#[cfg(test)]
mod tests
{
    use std::{fmt, time::{Duration, Instant}};

    use crate::{common, timer::OrderedTimerDeque, timer_consumer::TimerDequeueConsumer, timer_portable::timer::TimerReadRes};

    #[test]
    fn test_0()
    {
        #[derive(Debug, PartialEq, Eq, Clone)]
        struct TestItem(u64);

        impl fmt::Display for TestItem
        {
            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
            {
                write!(f, "0 = {}", self.0)
            }
        }

        let mut time_list = 
            OrderedTimerDeque
                ::<TimerDequeueConsumer<TestItem>>
                ::new("test_label".into(), 4, false).unwrap();

        let s = Instant::now();

        let tss_set = common::get_current_timestamp().timestamp()+2;
        let ent1 = TestItem(1);
        time_list.add_to_timer(ent1, tss_set, 0).unwrap();


        for _ in 0..3
        {
            let event = time_list.wait_for_event().unwrap();
            if let TimerReadRes::WouldBlock = event
            {
                std::thread::sleep(Duration::from_millis(1001));

                continue;
            }

            assert_eq!(event, TimerReadRes::Ok(1));

            let e = s.elapsed();
            let ts = common::get_current_timestamp();

            assert_eq!(ts.timestamp(), tss_set);
            println!("instant: {:?} ts: {}, timerset: {}", e, ts.timestamp(), tss_set);

            let mut tm_items: Vec<TestItem> = Vec::with_capacity(1);

            time_list.timeout_event_handler(event, &mut tm_items,).unwrap();

            assert_eq!(tm_items.len(), 1);
            assert_eq!(tm_items.first().is_some(), true);
            assert_eq!(tm_items.first().unwrap(), &TestItem(1));

            return;
        }

        panic!("timeout befor timer!");
    }
}