squeue 0.10.3

// Copyright (C) 2024 Christian Mauduit <ufoot@ufoot.org>

use crate::drain::Drain;
use crate::queue::Queue;
use std::fmt;
use std::sync::{Arc, RwLock};

/// Thread-safe FIFO queue with a fixed capacity.
///
/// Similar to the standard non thread-safe version, but can
/// be used for queueing in multiple consumers/producers contexts.
/// There's no subscription though, one needs to do some polling.
///
/// The peek function needs items to support Clone.
///
/// # Examples
///
/// ```
/// use squeue::SyncQueue;
///
/// // Does not need to be `mut`
/// let mut queue: SyncQueue<usize> = SyncQueue::new(3);
/// assert_eq!(None, queue.push(1));
/// assert_eq!(None, queue.push(2));
/// assert_eq!(None, queue.push(3));
/// assert_eq!(Some(1), queue.push(4));
/// assert_eq!(3, queue.len());
/// ```
#[derive(Debug, Clone)]
pub struct SyncQueue<T> {
    inner: Arc<RwLock<Queue<T>>>,
}

/// Pretty-print queue content.
///
/// Prints the next element to be popped, and the len/capacity.
///
/// # Examples
///
/// ```
/// use squeue::SyncQueue;
///
/// let queue = SyncQueue::new(100);
/// queue.push(123);
/// queue.push(4);
/// queue.push(5);
/// assert_eq!("[sync] { next: 123, len: 3, capacity: 100 }", format!("{}", queue));
/// queue.clear();
/// assert_eq!("[sync] { len: 0, capacity: 100 }", format!("{}", queue));
/// ```
impl<T> fmt::Display for SyncQueue<T>
where
    T: fmt::Display,
{
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "[sync] {}", self.inner.read().unwrap())
    }
}

/// Create a thread-safe queue from an ordinary queue.
///
/// # Examples
///
/// ```
/// use squeue::{Queue, SyncQueue};
///
/// let a: Queue<String> = Queue::new(100);
/// let b = SyncQueue::from(a);
/// assert_eq!(100, b.capacity());
/// ```
impl<T> From<Queue<T>> for SyncQueue<T> {
    fn from(queue: Queue<T>) -> SyncQueue<T> {
        SyncQueue {
            inner: Arc::new(RwLock::new(queue)),
        }
    }
}

/// Create an ordinary queue from a thread-safe queue.
///
/// This is possibly slow as it is O(n) since it clones the queue.
///
/// # Examples
///
/// ```
/// use squeue::{Queue, SyncQueue};
///
/// let a: SyncQueue<String> = SyncQueue::new(100);
/// let b = Queue::from(a);
/// assert_eq!(100, b.capacity());
/// ```
impl<T> From<SyncQueue<T>> for Queue<T>
where
    T: Clone,
{
    fn from(queue: SyncQueue<T>) -> Queue<T> {
        queue.inner.read().unwrap().clone()
    }
}

impl<T> SyncQueue<T> {
    /// Create a new queue.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<String> = SyncQueue::new(10);
    /// assert_eq!(10, queue.capacity());
    /// ```
    pub fn new(capacity: usize) -> Self {
        SyncQueue::from(Queue::new(capacity))
    }

    /// Return queue length.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<&str> = SyncQueue::new(10);
    /// assert_eq!(0, queue.len());
    /// queue.push("x");
    /// queue.push("y");
    /// assert_eq!(2, queue.len());
    /// ```
    #[inline]
    pub fn len(&self) -> usize {
        self.inner.read().unwrap().len()
    }

    /// Return queue capacity.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<f64> = SyncQueue::new(100);
    /// assert_eq!(100, queue.capacity());
    /// ```
    #[inline]
    pub fn capacity(&self) -> usize {
        self.inner.read().unwrap().capacity()
    }

    /// Returns true if queue is empty.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<String> = SyncQueue::new(100);
    /// assert!(queue.is_empty());
    /// queue.push(String::from("abc"));
    /// assert!(!queue.is_empty());
    /// ```
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.inner.read().unwrap().is_empty()
    }

    /// Returns true if queue is full.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<usize> = SyncQueue::new(10);
    /// assert!(!queue.is_full());
    /// for i in 0..10 {
    ///     queue.push(i);
    /// }
    /// assert!(queue.is_full());
    /// ```
    #[inline]
    pub fn is_full(&self) -> bool {
        self.inner.read().unwrap().is_full()
    }

    /// Clear all data.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<String> = SyncQueue::new(100);
    /// queue.push(String::from("abc"));
    /// queue.clear();
    /// assert!(queue.is_empty());
    /// ```
    #[inline]
    pub fn clear(&self) {
        self.inner.write().unwrap().clear()
    }

    /// Resize queue.
    ///
    /// Returns the number of dropped items, if any.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<usize> = SyncQueue::new(10);
    /// assert_eq!(10, queue.capacity());
    /// assert_eq!(0, queue.resize(100));
    /// assert_eq!(100, queue.capacity());
    /// for i in 0..1000 {
    ///     queue.push(i);
    /// }
    /// assert_eq!(100, queue.len());
    /// assert_eq!(90, queue.resize(10));
    /// assert_eq!(10, queue.capacity());
    /// assert_eq!(10, queue.len());
    /// assert_eq!(Some(990), queue.pop());
    /// ```
    pub fn resize(&self, capacity: usize) -> usize {
        self.inner.write().unwrap().resize(capacity)
    }

    /// Push data into the queue.
    ///
    /// If the queue is full and data needs to be dropped,
    /// that data is returned.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<usize> = SyncQueue::new(2);
    /// assert_eq!(None, queue.push(1));
    /// assert_eq!(None, queue.push(10));
    /// assert_eq!(Some(1), queue.push(100));
    /// assert_eq!(Some(10), queue.push(1000));
    /// assert_eq!(2, queue.len());
    /// ```
    #[inline]
    pub fn push(&self, item: T) -> Option<T> {
        self.inner.write().unwrap().push(item)
    }

    /// Pop data from the queue.
    ///
    /// The oldest item is returned, this works in FIFO mode.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<usize> = SyncQueue::new(5);
    /// assert_eq!(None, queue.pop());
    /// assert_eq!(None, queue.push(1));
    /// assert_eq!(Some(1), queue.pop());
    /// assert_eq!(None, queue.push(2));
    /// assert_eq!(None, queue.push(3));
    /// assert_eq!(Some(2), queue.pop());
    /// assert_eq!(1, queue.len());
    /// ```
    #[inline]
    pub fn pop(&self) -> Option<T> {
        self.inner.write().unwrap().pop()
    }

    /// Creates an iterator which drains the queue.
    ///
    /// Use this when you want to get all the items, at a given point,
    /// and free space in the queue.
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue = SyncQueue::new(10);
    /// queue.push(1);
    /// queue.push(10);
    /// let drain = queue.drain();
    /// assert_eq!(0, queue.len());
    /// assert_eq!(2, drain.count());
    /// ```
    pub fn drain(&self) -> Drain<T> {
        self.inner.write().unwrap().drain()
    }
}

impl<T> SyncQueue<T>
where
    T: Clone,
{
    /// Peek data, get it without removing it.
    ///
    /// This gives an insight on what pop() would return.
    /// Items need to support Clone.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let queue: SyncQueue<usize> = SyncQueue::new(5);
    /// assert_eq!(None, queue.peek());
    /// assert_eq!(None, queue.push(1));
    /// assert_eq!(None, queue.push(2));
    /// assert_eq!(Some(1), queue.peek());
    /// ```
    #[inline]
    pub fn peek(&self) -> Option<T> {
        match self.inner.read().unwrap().peek() {
            Some(item) => Some(item.clone()),
            None => None,
        }
    }
}

impl<T> FromIterator<T> for SyncQueue<T> {
    /// Creates a new queue from an iterator.
    ///
    /// With this, you can use collect() to build a queue.
    ///
    /// # Examples
    ///
    /// ```
    /// use squeue::SyncQueue;
    ///
    /// let src: Vec<usize> = vec![4, 5, 6];
    ///
    /// let queue = src.into_iter().collect::<SyncQueue<usize>>();
    /// assert_eq!(Some(4), queue.pop());
    /// assert_eq!(Some(5), queue.pop());
    /// assert_eq!(Some(6), queue.pop());
    /// assert_eq!(None, queue.pop());
    /// assert_eq!(3, queue.capacity());
    /// ```
    fn from_iter<I: IntoIterator<Item = T>>(iter: I) -> Self {
        let queue = SyncQueue::new(0);

        for i in iter {
            queue.resize(queue.len() + 1);
            queue.push(i);
        }

        queue
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rand::Rng;
    use std::thread;

    #[test]
    fn test_queue_0() {
        let queue: SyncQueue<usize> = SyncQueue::new(0);
        assert_eq!(None, queue.push(42));
        assert_eq!(None, queue.pop());
        assert_eq!(0, queue.capacity());
        assert_eq!(true, queue.is_full());
        assert_eq!(true, queue.is_empty());
    }

    fn monkey_fn(queue: SyncQueue<usize>) {
        let mut rng = rand::thread_rng();
        for i in 0..1_000_000 {
            if i % 1000 == 0 {
                print!(".");
            }
            if i % 100_000 == 0 {
                queue.clear();
                continue;
            }
            let dice = rng.gen_range(0..=100);
            if dice < 50 {
                queue.push(i & 100);
                continue;
            }
            if dice < 80 {
                queue.pop();
                continue;
            }
            if dice < 90 {
                queue.peek();
                continue;
            }
        }
    }

    #[test]
    fn test_monkey() {
        let queue: SyncQueue<usize> = SyncQueue::new(50);

        let queue1 = queue.clone();
        let handle1 = thread::spawn(move || monkey_fn(queue1));
        let queue2 = queue.clone();
        let handle2 = thread::spawn(move || monkey_fn(queue2));

        handle1.join().unwrap();
        handle2.join().unwrap();
    }
}