timer_deque_rs/

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 rand::RngCore;

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

/// A uniq ID which is generated for every ticket. It is a 128bit
/// value which consists from timestamp sec and ns and random number.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TimerDequeueId
{
    /// seconds
    timestamp: i64,

    /// nanoseconds
    timestamp_ns: u32,

    /// random number
    ran_id: u32,
}

impl fmt::Display for TimerDequeueId
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result 
    {
        write!(f, "{:X}/{:X}/{:X}", self.timestamp, self.timestamp_ns, self.ran_id)
    }
}

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

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

impl TimerDequeueId
{
    fn new() -> Self
    {
        let mut rng = rand::rng();

        let ts = common::get_current_timestamp();

        return 
            Self
            { 
                timestamp: ts.timestamp(), 
                timestamp_ns: ts.timestamp_subsec_nanos(), 
                ran_id: rng.next_u32(), 
            }; 
    }
}

#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TimerDequeueTicket
{
   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 PartialEq<TimerDequeueTicketIssuer> for TimerDequeueTicket
{
    fn eq(&self, other: &TimerDequeueTicketIssuer) -> 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.
#[derive(Debug)]
pub struct TimerDequeueTicketIssuer
{
    /// 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_absolute: TimerExpMode,
}

impl fmt::Display for TimerDequeueTicketIssuer
{  
    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_absolute)
    }
}

impl Eq for TimerDequeueTicketIssuer {}

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

        return s == o;
    }
}

impl PartialEq<TimerDequeueTicket> for TimerDequeueTicketIssuer
{
    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 Ord for TimerDequeueTicketIssuer
{
    fn cmp(&self, other: &Self) -> std::cmp::Ordering 
    {
        return self.timeout_absolute.cmp(&other.timeout_absolute);
    }
}

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


impl TimerDequeueTicketIssuer
{ 
    /// Initializes new instance.
    fn new(abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<(Self, TimerDequeueTicket)>
    {
        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");
        }

        
        let ext_ticket = 
            TimerDequeueTicket::new();

        let int_ticket = 
            Self
            {
                weak_status: ext_ticket.pair(),
                timeout_absolute: req
            };


        return Ok((int_ticket, ext_ticket));
    }

    /// Returns the absolute timeout of the current instance.
    fn get_time_until(&self) -> TimerExpMode
    {
        return self.timeout_absolute;
    }

    /// 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 OrderedTimerDequeIntrf for TimerDequeueTicketIssuer
{
    /// No target is requred.
    type Target = common::NoTarget;

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

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

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

impl OrderedTimerDeque<TimerDequeueTicketIssuer>
{
    /// Adds the new absolute timeout to the timer dequeue.
    /// 
    /// # Arguemnts
    /// 
    /// `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 [TimerDequeueTicket] ticket.
    /// 
    /// * [Result::Err] with error description.
    #[inline]
    pub   
    fn add_to_timer(&mut self, abs_time_sec: i64, abs_time_nsec: i64) -> TimerResult<TimerDequeueTicket>
    {
        return self.add_to_timer_local(common::NoTarget, abs_time_sec, abs_time_nsec);
    }

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

    /// 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 tickets which were not dropped and timeout.
    /// 
    /// 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(());
        }
        
        // check if the time reset is required
        let force_timer_resched = 
            TimerReadRes::Cancelled == res || TimerReadRes::Ok(0) == res;

            
        let mut to_remove: Vec<TimerDequeueId> = Vec::new();
        let mut error: Option<TimerError> = None;
        let mut last_time_until: TimerExpMode = TimerExpMode::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 
                { 
                    // queue is empty, stop timer if needed.
                    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)
            {
                if let Some(item) = self.deque_timeout_list.pop_front().unwrap().into_inner()
                {
                    to_remove.push(item);
                }

                // else item has gone previously or while we was messing around with queue
            }
            else
            {
                // we need to rearm the timer if so
                if last_time_until != TimerExpMode::None || force_timer_resched == true
                {
                    // call timer reshedule
                    self.reschedule_timer()?;
                }

                break;
            }

            last_time_until = time_until;
        }

        *timeout_items = to_remove;

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

}

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

    use crate::{common, timer::OrderedTimerDeque, timer_portable::timer::TimerReadRes, timer_tickets::{TimerDequeueId, TimerDequeueTicketIssuer}};

    #[test]
    fn test_ticket_0()
    {
        let mut time_list = 
            OrderedTimerDeque
                ::<TimerDequeueTicketIssuer>
                ::new("test_label".into(), 4, false).unwrap();


        let s = Instant::now();

        let tss_set = common::get_current_timestamp().timestamp()+2;
        
        let ticket = time_list.add_to_timer(tss_set, 0).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 = common::get_current_timestamp();

            assert_eq!(ts.timestamp(), tss_set);
            println!("instant: {:?} ts: {}, timerset: {}", e, ts.timestamp(), 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!");
    }
}