p2panda_sync/
dedup.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! Ring-buffer based deduplication of generic messages.
4use std::collections::{HashSet, VecDeque};
5use std::hash::Hash;
6
7pub static DEFAULT_BUFFER_CAPACITY: usize = 1024;
8
9/// Maintain a ring buffer of generic items and efficiently identify if an item is currently in
10/// the buffer.
11#[derive(Debug)]
12pub struct DeduplicationBuffer<T> {
13    buffer: VecDeque<T>,
14    set: HashSet<T>,
15}
16
17impl<T> Default for DeduplicationBuffer<T> {
18    fn default() -> Self {
19        Self {
20            buffer: VecDeque::with_capacity(DEFAULT_BUFFER_CAPACITY),
21            set: HashSet::with_capacity(DEFAULT_BUFFER_CAPACITY),
22        }
23    }
24}
25
26impl<T> DeduplicationBuffer<T>
27where
28    T: Eq + Hash + Clone,
29{
30    /// Instantiate a new buffer with "capacity" buffer size.
31    pub fn new(capacity: usize) -> Self {
32        Self {
33            buffer: VecDeque::with_capacity(capacity),
34            set: HashSet::with_capacity(capacity),
35        }
36    }
37
38    /// Insert an item into the buffer.
39    ///
40    /// If the buffer capacity has been reached then the oldest item will be evicted from the
41    /// buffer.
42    ///
43    /// Returns `true` if an insertion occurred, or `false` if the item was already in the buffer.
44    pub fn insert(&mut self, item: T) -> bool {
45        if self.set.contains(&item) {
46            return false;
47        }
48
49        // If the buffer is at max capacity then we first need to pop an item from the front and
50        // remove it from the set.
51        if self.buffer.len() + 1 > self.buffer.capacity() {
52            let evicted = self.buffer.pop_front();
53            if let Some(evicted) = evicted {
54                self.set.remove(&evicted);
55            }
56        }
57
58        self.buffer.push_back(item.clone());
59        self.set.insert(item);
60        true
61    }
62
63    /// Returns `true` if the item is currently in the buffer.
64    pub fn contains(&self, item: &T) -> bool {
65        self.set.contains(item)
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72
73    #[test]
74    fn insert_items() {
75        let mut d = DeduplicationBuffer::new(3);
76
77        assert!(d.insert(1));
78        assert!(d.insert(2));
79        assert!(d.insert(3));
80
81        assert!(d.contains(&1));
82        assert!(d.contains(&2));
83        assert!(d.contains(&3));
84    }
85
86    #[test]
87    fn insert_ignores_duplicates() {
88        let mut d = DeduplicationBuffer::new(3);
89
90        assert!(d.insert(42));
91        assert!(!d.insert(42));
92
93        assert!(d.contains(&42));
94        assert_eq!(d.buffer.len(), 1);
95        assert_eq!(d.set.len(), 1);
96    }
97
98    #[test]
99    fn insert_evicts_when_capacity_reached() {
100        let mut d = DeduplicationBuffer::new(3);
101
102        d.insert(1);
103        d.insert(2);
104        d.insert(3);
105        assert!(d.contains(&1));
106
107        d.insert(4);
108
109        assert!(!d.contains(&1));
110        assert!(d.contains(&2));
111        assert!(d.contains(&3));
112        assert!(d.contains(&4));
113
114        assert_eq!(d.buffer.len(), 3);
115    }
116}
p2panda_sync/dedup.rs

p2panda_sync/
dedup.rs
