opentelemetry_spanprocessor_any/sdk/trace/
evicted_queue.rs

1//! # Evicted Queue
2
3#[cfg(feature = "serialize")]
4use serde::{Deserialize, Serialize};
5use std::collections::VecDeque;
6
7/// This queue maintains an ordered list of elements, and a count of
8/// dropped elements. Elements are removed from the queue in a first
9/// in first out fashion.
10#[cfg_attr(feature = "serialize", derive(Deserialize, Serialize))]
11#[derive(Clone, Debug, PartialEq)]
12pub struct EvictedQueue<T> {
13    queue: Option<VecDeque<T>>,
14    max_len: u32,
15    dropped_count: u32,
16}
17
18impl<T> EvictedQueue<T> {
19    /// Create a new `EvictedQueue` with a given max length.
20    pub fn new(max_len: u32) -> Self {
21        EvictedQueue {
22            queue: None,
23            max_len,
24            dropped_count: 0,
25        }
26    }
27
28    /// Push a new element to the back of the queue, dropping and
29    /// recording dropped count if over capacity.
30    pub(crate) fn push_back(&mut self, value: T) {
31        let queue = self.queue.get_or_insert_with(Default::default);
32        if queue.len() as u32 == self.max_len {
33            queue.pop_front();
34            self.dropped_count += 1;
35        }
36        queue.push_back(value);
37    }
38
39    /// Moves all the elements of other into self, leaving other empty.
40    pub fn append_vec(&mut self, other: &mut Vec<T>) {
41        self.extend(other.drain(..));
42    }
43
44    /// Returns `true` if the `EvictedQueue` is empty.
45    pub fn is_empty(&self) -> bool {
46        self.queue.as_ref().map_or(true, |queue| queue.is_empty())
47    }
48
49    /// Returns a front-to-back iterator.
50    pub fn iter(&self) -> Iter<'_, T> {
51        Iter(self.queue.as_ref().map(|queue| queue.iter()))
52    }
53
54    /// Returns the number of elements in the `EvictedQueue`.
55    pub fn len(&self) -> usize {
56        self.queue.as_ref().map_or(0, |queue| queue.len())
57    }
58
59    /// Count of dropped attributes
60    pub fn dropped_count(&self) -> u32 {
61        self.dropped_count
62    }
63}
64
65/// An owned iterator over the entries of a `EvictedQueue`.
66#[derive(Debug)]
67pub struct IntoIter<T>(Option<std::collections::vec_deque::IntoIter<T>>);
68
69impl<T> Iterator for IntoIter<T> {
70    type Item = T;
71
72    fn next(&mut self) -> Option<Self::Item> {
73        self.0.as_mut().and_then(|iter| iter.next())
74    }
75}
76
77impl<T> IntoIterator for EvictedQueue<T> {
78    type Item = T;
79    type IntoIter = IntoIter<T>;
80
81    fn into_iter(self) -> Self::IntoIter {
82        IntoIter(self.queue.map(|queue| queue.into_iter()))
83    }
84}
85
86/// An iterator over the entries of an `EvictedQueue`.
87#[derive(Debug)]
88pub struct Iter<'a, T>(Option<std::collections::vec_deque::Iter<'a, T>>);
89
90impl<'a, T: 'static> Iterator for Iter<'a, T> {
91    type Item = &'a T;
92
93    fn next(&mut self) -> Option<Self::Item> {
94        self.0.as_mut().and_then(|iter| iter.next())
95    }
96}
97
98impl<T> Extend<T> for EvictedQueue<T> {
99    fn extend<I: IntoIterator<Item = T>>(&mut self, iter: I) {
100        iter.into_iter().for_each(move |elt| self.push_back(elt));
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::EvictedQueue;
107    use std::collections::VecDeque;
108
109    #[test]
110    fn insert_over_capacity_test() {
111        let capacity = 10;
112        let mut queue = EvictedQueue::new(capacity);
113
114        for i in 0..=capacity {
115            queue.push_back(i)
116        }
117
118        assert_eq!(queue.dropped_count, 1);
119        assert_eq!(queue.len(), capacity as usize);
120        assert_eq!(
121            queue.queue.unwrap(),
122            (1..=capacity).collect::<VecDeque<_>>()
123        );
124    }
125}
opentelemetry_spanprocessor_any/sdk/trace/evicted_queue.rs

opentelemetry_spanprocessor_any/sdk/trace/
evicted_queue.rs
