timer_deque_rs/deque_timeout/

timer_tickets.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, sync::{Arc, Weak}};

use crate::
{
    common::{self, TimerDequeueId}, deque_timeout::{OrderdTimerDequeOnce, OrderdTimerDequePeriodic, OrderedTimerDequeMode}, error::{TimerError, TimerErrorType, TimerResult}, timer_err, timer_portable::timer::{AbsoluteTime, RelativeTime, TimerExpMode}, TimerReadRes
};

use super::{OrderedTimerDeque, OrderedTimerDequeIntrf};


impl PartialEq<TimerDequeueTicket> for TimerDequeueId
{
    fn eq(&self, other: &TimerDequeueTicket) -> bool 
    {
        return self == other.status.as_ref();
    }
}

/// A ticket which is issued to the route which added the instance
/// to the queue. If this instance is dropped, the [Weak] reference to [Arc]
/// will no longer be valid and the timer would ignore the instance and remove
/// it from the queue independeltly from the deque operation mode.
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TimerDequeueTicket
{
    /// A single instance of arc which the presense of which indicates that
    /// the instance is valid. It holds the uniq ID which can be used to
    /// remove the instance from the queue manually.
    status: Arc<TimerDequeueId>,
}

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

impl PartialEq<TimerDequeueId> for TimerDequeueTicket
{
    fn eq(&self, other: &TimerDequeueId) -> bool 
    {
        return self.status.as_ref() == other;
    }
}

impl<MODE: OrderedTimerDequeMode> PartialEq<TimerDequeueTicketIssuer<MODE>> for TimerDequeueTicket
{
    fn eq(&self, other: &TimerDequeueTicketIssuer<MODE>) -> bool 
    {
        let Some(up) = other.weak_status.upgrade()
            else { return false };

        return self.status.as_ref() == up.as_ref();
    }
}

impl AsRef<TimerDequeueId> for TimerDequeueTicket
{
    fn as_ref(&self) -> &TimerDequeueId 
    {
        return &self.status;
    }
}

impl TimerDequeueTicket
{
    fn new() -> Self
    {
        return Self{ status: Arc::new(TimerDequeueId::new() ) }
    }

    fn pair(&self) -> Weak<TimerDequeueId>
    {
        return Arc::downgrade(&self.status);
    }

    pub 
    fn get_deque_id(&self) -> &TimerDequeueId
    {
        return self.status.as_ref();
    }

    pub 
    fn is_queued(&self) -> bool
    {
        return Arc::weak_count(&self.status) > 1;
    }
}

/// A type of deque which issues a `tickets` in form of [TimerDequeueTicket] 
/// when the timer is set. This type of timer dequeue allows not to delete the
/// item from the queue and just drop the `ticket` instance. The timer dequeue 
/// would ignore the dropped instances.
/// 
/// # Generics
/// 
/// `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.
/// 
/// # Examples
/// 
/// ```ignore
/// let mut time_list = 
///     OrderedTimerDeque
///         ::<TimerDequeueTicketIssuer<OrderdTimerDequeOnce>>
///         ::new("test_label".into(), 4, false).unwrap();
/// ```
/// or
/// ```ignore
/// let mut time_list = 
///     OrderedTimerDeque
///         ::<TimerDequeueTicketIssuer<OrderdTimerDequePeriodic>>
///         ::new("test_label".into(), 4, false).unwrap();
/// ```
#[derive(Debug)]
pub struct TimerDequeueTicketIssuer<MODE: OrderedTimerDequeMode>
{
    /// A [Weak] reference to [Arc] which holds the queue identification.
    weak_status: Weak<TimerDequeueId>,

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

impl<MODE: OrderedTimerDequeMode> fmt::Display for TimerDequeueTicketIssuer<MODE>
{  
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> 
    { 
        write!(f, "{} until: {}", self.weak_status.upgrade().map_or("dropped".into(), |f| f.to_string()), 
            self.timeout_mode)
    }
}

impl<MODE: OrderedTimerDequeMode> Eq for TimerDequeueTicketIssuer<MODE> {}

impl<MODE: OrderedTimerDequeMode> PartialEq for TimerDequeueTicketIssuer<MODE>
{
    fn eq(&self, other: &Self) -> bool 
    {
        let s = self.weak_status.upgrade();
        let o = other.weak_status.upgrade();

        return s == o;
    }
}

impl<MODE: OrderedTimerDequeMode> PartialEq<TimerDequeueTicket> for TimerDequeueTicketIssuer<MODE>
{
    fn eq(&self, other: &TimerDequeueTicket) -> bool 
    {
        let Some(s) = self.weak_status.upgrade()
            else { return false };

        return s.as_ref() == other.status.as_ref();
    }
}

impl<MODE: OrderedTimerDequeMode> Ord for TimerDequeueTicketIssuer<MODE>
{
    fn cmp(&self, other: &Self) -> std::cmp::Ordering 
    {
        return self.timeout_mode.cmp(&other.timeout_mode);
    }
}

impl<MODE: OrderedTimerDequeMode> PartialOrd for TimerDequeueTicketIssuer<MODE>
{
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> 
    {
        return Some(self.cmp(other));
    }
}


impl TimerDequeueTicketIssuer<OrderdTimerDequePeriodic>
{ 
    /// Initializes new instance.
    fn new(rel_time: RelativeTime) -> TimerResult<(Self, TimerDequeueTicket)>
    {        
        let ext_ticket = 
            TimerDequeueTicket::new();

        let int_ticket = 
            Self
            {
                weak_status: 
                    ext_ticket.pair(),
                timeout_mode: 
                   OrderdTimerDequePeriodic::new(rel_time)
            };


        return Ok((int_ticket, ext_ticket));
    }
}

impl TimerDequeueTicketIssuer<OrderdTimerDequeOnce>
{ 
    /// Initializes new instance.
    fn new(abs_time: AbsoluteTime) -> TimerResult<(Self, TimerDequeueTicket)>
    {

        let cur = AbsoluteTime::now();

        if cur > abs_time
        {
            timer_err!(TimerErrorType::Expired, "time already expired");
        }
        
        let ext_ticket = 
            TimerDequeueTicket::new();

        let int_ticket = 
            Self
            {
                weak_status: 
                    ext_ticket.pair(),
                timeout_mode: 
                    OrderdTimerDequeOnce::new(abs_time)
            };


        return Ok((int_ticket, ext_ticket));
    }
}

impl<MODE: OrderedTimerDequeMode> TimerDequeueTicketIssuer<MODE>
{
    /// Checks if the instance is still actual i.e was not dropeed.
    fn is_valid(&self) -> bool
    {
        return self.weak_status.upgrade().is_some();
    }

    /// Obtains the [TimerDequeueId] only if the [Weak] pointer `weak_status`
    /// to ID is still actual.
    fn into_inner(&self) -> Option<TimerDequeueId>
    {
        return self.weak_status.upgrade().map(|f| *f.as_ref());
    }
}


impl<MODE: OrderedTimerDequeMode> OrderedTimerDequeIntrf for TimerDequeueTicketIssuer<MODE>
{
    /// No target is requred.
    type Target = common::NoTarget;

    /// Return the ticket.
    type Ticket = TimerDequeueTicket;

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

impl OrderedTimerDeque<TimerDequeueTicketIssuer<OrderdTimerDequeOnce>>
{
    /// Adds the new entry to the timeout qeueu which after alarm
    /// removes the entry from the queue. A `ticket` will be 
    /// issued which can be used to either deactivate the record 
    /// by droppping it or to remove it manually.
    /// 
    /// # Arguemnts
    /// 
    /// `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 [TimerDequeueTicket] ticket.
    /// 
    /// * [Result::Err] with error description.
    #[inline]
    pub   
    fn add_to_timer(&mut self, abs_time: AbsoluteTime) -> TimerResult<TimerDequeueTicket>
    {
        let (inst, ticket) = 
            TimerDequeueTicketIssuer::<OrderdTimerDequeOnce>::new(abs_time)?;

        return self.add_to_timer_local(inst).map(|_| ticket);
    }

    /// 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<TimerDequeueId>) -> 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
            {
                if let Some(item) = self.deque_timeout_list.pop_front().unwrap().into_inner()
                {
                    timeout_items.push(item);
                }

                // else item has gone previously or while we was messing around with queue
            }
            else
            {
                break;
            }
        }

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

        return Ok(());
    }
}

impl OrderedTimerDeque<TimerDequeueTicketIssuer<OrderdTimerDequePeriodic>>
{
    /// Adds the new entry to the timeout queue which after alarm
    /// extends the timeout and sets the instance in the queue again. 
    /// A `ticket` will be issued which can be used to either deactivate 
    /// the record by droppping it or to remove it manually.
    /// 
    /// # Arguemnts
    /// 
    /// `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 [TimerDequeueTicket] ticket.
    /// 
    /// * [Result::Err] with error description.
    #[inline]
    pub   
    fn add_to_timer(&mut self, rel_time: RelativeTime) -> TimerResult<TimerDequeueTicket>
    {
        let (inst, ticket) = 
            TimerDequeueTicketIssuer::<OrderdTimerDequePeriodic>::new(rel_time)?;

        return self.add_to_timer_local(inst).map(|_| ticket);
    }

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

                if let Some(item) = deq.into_inner()
                {
                    // store the ID of item which is dequed
                    timeout_items.push(item);

                    // advance the timeout
                    deq.timeout_mode.advance_timeout();

                    // return back to deque with new timeout
                    self.add_to_timer_local(deq)?;
                }

                // else item has gone previously or while we was messing around with queue
            }
            else
            {
                break;
            }
        }

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

        return Ok(());
    }

}

impl<MODE: OrderedTimerDequeMode> OrderedTimerDeque<TimerDequeueTicketIssuer<MODE>>
{
    /// Removes the instance from the queue by the issued `ticket`.
    /// 
    /// # Arguments
    /// 
    /// `ticket` - [TimerDequeueTicket] an issued ticket.
    /// 
    /// # Returns
    /// 
    /// A [Result] as alias [TimerResult] is returned with:
    /// 
    /// * [Result::Ok] witout any innder data.
    /// 
    /// * [Result::Err] with error description.
    pub 
    fn remove_from_sched_queue(&mut self, ticket: TimerDequeueTicket) -> TimerResult<()>
    {
        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 _ = self.deque_timeout_list.pop_front();

                // stop timer
                self.stop_timer()?;

                return Ok(());
            }
            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 == &ticket
                    {
                        // remove by the index
                        let _ = 
                            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(());
                    }
                }

                return Ok(());
            }
        }
    }
}

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

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

    #[test]
    fn test_ticket_0()
    {
        let mut time_list = 
            OrderedTimerDeque
                ::<TimerDequeueTicketIssuer<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);
        
        let ticket = time_list.add_to_timer(tss_set).unwrap();

        println!("ticket issued: {}", ticket);

        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<TimerDequeueId> = Vec::with_capacity(1);

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

            println!("tickets: {}", tm_items[0]);

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

            return;
        }

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