timer_deque_rs/deque_timeout/

timer_consumer.rs

/*-
 * timer-deque-rs - a Rust crate which provides timer and timer queues based on target OS
 *  functionality.
 * 
 * Copyright (C) 2025 Aleksandr Morozov alex@nixd.org
 *  4neko.org alex@4neko.org
 * 
 * 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::deque_timeout::{OrderdTimerDequeOnce, OrderdTimerDequePeriodic, OrderedTimerDequeMode};
use crate::error::{TimerErrorType, TimerResult};
use crate::{common, timer_err, TimerReadRes};
use crate::timer_portable::timer::{AbsoluteTime, RelativeTime};

use super::{OrderedTimerDeque, OrderedTimerDequeIntrf};

/// 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].
/// 
/// `MODE` - a [OrderedTimerDequeMode] which defines the deque behaviour. There are
///     two types of the operation:
/// 
/// * [OrderdTimerDequeOnce] - after timeout the element is removed from the queue.
/// 
/// * [OrderdTimerDequePeriodic] - after timeout the element timeout is extended
///     until the item is not removed from the queue manually.
/// 
/// # Example
/// 
/// ```ignore
/// let mut time_list = 
///         OrderedTimerDeque
///             ::<TimerDequeueConsumer<TestItem, OrderdTimerDequeOnce>>
///             ::new("test_label".into(), 4, false).unwrap();
/// ```
/// or
/// ```ignore
/// let mut time_list = 
///         OrderedTimerDeque
///             ::<TimerDequeueConsumer<TestItem, OrderdTimerDequePeriodic>>
///             ::new("test_label".into(), 4, false).unwrap();
/// ```
#[derive(Debug)]
pub struct TimerDequeueConsumer<R, MODE>
where 
    R: PartialEq + Eq + fmt::Debug + fmt::Display + Send,
    MODE: OrderedTimerDequeMode
{
    /// An internal data
    target: R,

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

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

impl<R, MODE> TimerDequeueConsumer<R, MODE>
where 
    R: PartialEq + Eq + fmt::Display + fmt::Debug + Send,
    MODE: OrderedTimerDequeMode
{ 
    fn into_inner(self) -> R
    {
        return self.target;
    }
}


impl<R> TimerDequeueConsumer<R, OrderdTimerDequeOnce>
where R: PartialEq + Eq + fmt::Display + fmt::Debug + Send
{ 
    fn new(target: R, abs_time: AbsoluteTime) -> TimerResult<Self>
    {
        let cur = AbsoluteTime::now();

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

        return Ok(
            Self
            {
                target: 
                    target,
                timeout_mode: 
                    OrderdTimerDequeOnce::new(abs_time),
            }
        );
    }
}

impl<R> TimerDequeueConsumer<R, OrderdTimerDequePeriodic>
where R: PartialEq + Eq + fmt::Display + fmt::Debug + Send + Clone
{ 
    fn new(target: R, rel_time: RelativeTime) -> TimerResult<Self>
    {
        return Ok(
            Self
            {
                target: 
                    target,
                timeout_mode: 
                    OrderdTimerDequePeriodic::new(rel_time),
            }
        );
    }

    fn clone_inner(&self) -> R
    {
        return self.target.clone();
    }
}

impl<R, MODE> Eq for TimerDequeueConsumer<R, MODE>
where 
    R: PartialEq + Eq + fmt::Display + fmt::Debug + Send, 
    MODE: OrderedTimerDequeMode
{

}

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

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

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

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

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

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

    #[inline]
    fn get_timeout_absolute(&self) -> AbsoluteTime 
    {
        return self.timeout_mode.get_absolut_timeout();
    }
}

impl<R> OrderedTimerDeque<TimerDequeueConsumer<R, OrderdTimerDequeOnce>>
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` - a [AbsoluteTime] absolute time in future when the timeout
    ///     should happen.
    /// 
    /// # 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: AbsoluteTime) -> TimerResult<common::NoTicket>
    {
        let inst = 
            TimerDequeueConsumer::<R, OrderdTimerDequeOnce>::new(entity, abs_time)?;

        return self.add_to_timer_local(inst).map(|_| common::NoTicket);
    }

    /// 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(());
        }

        let cur_timestamp = AbsoluteTime::now();

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

            let time_until = front_entity.get_timeout_absolute();

            if time_until <= cur_timestamp
            {
                timeout_items.push(self.deque_timeout_list.pop_front().unwrap().into_inner());
            }
            else
            {
                break;
            }
        }

        // call timer reschedule
        self.reschedule_timer()?;

        return Ok(());
    }

}

impl<R> OrderedTimerDeque<TimerDequeueConsumer<R, OrderdTimerDequePeriodic>>
where R: PartialEq + Eq + fmt::Display + fmt::Debug + Send + Clone
{
    /// Adds the new entry to the timout queue which after alarm
    /// extends the timeout and sets the instance in the queue again.
    /// 
    /// The `entity` is an item which should be stored in the dequeue and
    /// impl [Clone].
    /// 
    /// # Arguemnts
    /// 
    /// `entity` - `R` generic which is an item to store in the dequeu.
    /// 
    /// `rel_time` - a [RelativeTime] which will be used to extend the further 
    ///     timeouts. 
    /// 
    /// # 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, rel_time: RelativeTime) -> TimerResult<common::NoTicket>
    {
        let inst = 
            TimerDequeueConsumer::<R, OrderdTimerDequePeriodic>::new(entity, rel_time)?;

        return self.add_to_timer_local(inst).map(|_| common::NoTicket);
    }

    

    /// 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(());
        }

        let cur_timestamp = AbsoluteTime::now();

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

            let time_until = front_entity.get_timeout_absolute();

            if time_until <= cur_timestamp
            {
                let mut deq = self.deque_timeout_list.pop_front().unwrap();
                
                timeout_items.push(deq.clone_inner());

                // advance the timeout
                deq.timeout_mode.advance_timeout();
                
                // return back to deque with new timeout
                self.add_to_timer_local(deq)?;
            }
            else
            {
                break;
            }
        }

        // call timer reschedule
        self.reschedule_timer()?;

        return Ok(());
    }

}

impl<R, MODE> OrderedTimerDeque<TimerDequeueConsumer<R, MODE>>
where 
    R: PartialEq + Eq + fmt::Display + fmt::Debug + Send, 
    MODE: OrderedTimerDequeMode 
{
    /// 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);
            }
        }
    }
}

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

    use crate::{common, deque_timeout::OrderdTimerDequeOnce, timer_portable::timer::{AbsoluteTime, TimerReadRes}, OrderedTimerDeque, TimerDequeueConsumer};

    #[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, OrderdTimerDequeOnce>>
                ::new("test_label".into(), 4, false).unwrap();

        let s = Instant::now();

        let ts = common::get_current_timestamp();

        let tss_set = 
            AbsoluteTime
                ::new_time(ts.timestamp()+2, ts.timestamp_subsec_nanos() as i64)
                    .unwrap();
        
        let ent1 = TestItem(1);
        time_list.add_to_timer(ent1, tss_set).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 = AbsoluteTime::now();

            assert_eq!(ts.seconds_cmp(&tss_set) == Ordering::Equal, true);

            println!("ev: {}, instant: {:?} ts: {}, timerset: {}", event, e, ts, 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!");
    }
}