util: Implement RingBuffer as exportable API

src/event_graph/protocol_event.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::VecDeque, fmt::Debug};
+use std::fmt::Debug;
 
 use async_std::sync::{Arc, Mutex};
 use async_trait::async_trait;
@@ -28,36 +28,13 @@ use super::EventMsg;
 use crate::{
     event_graph::model::{Event, EventId, ModelPtr},
     net,
-    util::async_util::sleep,
+    util::{async_util::sleep, ringbuffer::RingBuffer},
     Result,
 };
 
 const SIZE_OF_SEEN_BUFFER: usize = 65536;
 // const MAX_CONFIRM: u8 = 3;
 
-#[derive(Clone)]
-struct RingBuffer<T> {
-    pub items: VecDeque<T>,
-}
-
-impl<T: Eq + PartialEq + Clone> RingBuffer<T> {
-    pub fn new(capacity: usize) -> Self {
-        let items = VecDeque::with_capacity(capacity);
-        Self { items }
-    }
-
-    pub fn push(&mut self, val: T) {
-        if self.items.len() == self.items.capacity() {
-            self.items.pop_front();
-        }
-        self.items.push_back(val);
-    }
-
-    pub fn contains(&self, val: &T) -> bool {
-        self.items.contains(val)
-    }
-}
-
 type InvId = u64;
 
 #[derive(SerialEncodable, SerialDecodable, Clone, Debug, PartialEq, Eq, Hash)]

src/net/channel.rs

@@ -16,8 +16,6 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::collections::VecDeque;
-
 use async_std::sync::{Arc, Mutex};
 use futures::{
     io::{ReadHalf, WriteHalf},
@@ -37,7 +35,7 @@ use super::{
 };
 use crate::{
     system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
-    util::time::NanoTimestamp,
+    util::{ringbuffer::RingBuffer, time::NanoTimestamp},
     Error, Result,
 };
 
@@ -46,25 +44,6 @@ pub type ChannelPtr = Arc<Channel>;
 
 const SIZE_OF_BUFFER: usize = 65536;
 
-#[derive(Clone, serde::Serialize, serde::Deserialize)]
-struct RingBuffer<T> {
-    pub items: VecDeque<T>,
-}
-
-impl<T: Eq + PartialEq + Clone> RingBuffer<T> {
-    pub fn new(capacity: usize) -> Self {
-        let items = VecDeque::with_capacity(capacity);
-        Self { items }
-    }
-
-    pub fn push(&mut self, val: T) {
-        if self.items.len() == self.items.capacity() {
-            self.items.pop_front();
-        }
-        self.items.push_back(val);
-    }
-}
-
 struct ChannelInfo {
     random_id: u32,
     remote_node_id: String,

src/util/mod.rs

@@ -37,3 +37,6 @@ pub mod path;
 
 /// Time utilities
 pub mod time;
+
+/// Ring Buffer implementation
+pub mod ringbuffer;

src/util/ringbuffer.rs

@@ -0,0 +1,95 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::collections::{vec_deque::Iter, VecDeque};
+
+/// A ring buffer of fixed capacity
+#[derive(Eq, PartialEq, Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct RingBuffer<T> {
+    capacity: usize,
+    data: VecDeque<T>,
+}
+
+impl<T: Eq + PartialEq + Clone> RingBuffer<T> {
+    /// Create a new [`RingBuffer`] with given fixed capacity
+    pub fn new(capacity: usize) -> RingBuffer<T> {
+        Self { capacity, data: VecDeque::with_capacity(capacity) }
+    }
+
+    /// Push an element to the back of the `RingBuffer`, removing
+    /// the front element in case the buffer is full.
+    pub fn push(&mut self, value: T) {
+        if self.data.len() == self.capacity {
+            self.data.pop_front();
+        }
+        self.data.push_back(value);
+    }
+
+    /// Returns the current number of items in the buffer
+    pub fn len(&self) -> usize {
+        self.data.len()
+    }
+
+    /// Returns true if buffer is empty, false otherwise
+    pub fn is_empty(&self) -> bool {
+        self.data.is_empty()
+    }
+
+    /// Removes and returns the oldest item in the buffer
+    pub fn pop(&mut self) -> Option<T> {
+        self.data.pop_front()
+    }
+
+    /// Returns a front-to-back iterator
+    pub fn iter(&self) -> Iter<'_, T> {
+        self.data.iter()
+    }
+
+    /// Returns true if the buffer contains an element equal to the given value
+    pub fn contains(&self, x: &T) -> bool {
+        self.data.contains(x)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn behaviour() {
+        const BUF_SIZE: usize = 10;
+        let mut buf = RingBuffer::new(BUF_SIZE);
+
+        for i in 0..BUF_SIZE {
+            buf.push(i);
+        }
+
+        assert!(!buf.is_empty());
+        assert!(buf.len() == BUF_SIZE);
+
+        for i in 0..BUF_SIZE {
+            buf.push(i + 10);
+        }
+
+        assert!(buf.len() == BUF_SIZE);
+
+        for (i, v) in buf.iter().enumerate() {
+            assert_eq!(*v, i + 10);
+        }
+    }
+}