rskit_stateful/
memory_store.rs

1//! In-memory store implementation.
2
3use std::collections::VecDeque;
4
5use parking_lot::Mutex;
6use tokio::time::Instant;
7
8use rskit_errors::AppResult;
9
10use crate::store::Store;
11
12struct State<V> {
13    values: VecDeque<V>,
14    last_activity: Instant,
15}
16
17/// In-memory [`Store`] backed by a FIFO queue.
18pub struct MemoryStore<V>
19where
20    V: Clone,
21{
22    state: Mutex<State<V>>,
23}
24
25impl<V> MemoryStore<V>
26where
27    V: Clone,
28{
29    /// Create an empty in-memory store.
30    #[must_use]
31    pub fn new() -> Self {
32        Self {
33            state: Mutex::new(State {
34                values: VecDeque::new(),
35                last_activity: Instant::now(),
36            }),
37        }
38    }
39}
40
41impl<V> Default for MemoryStore<V>
42where
43    V: Clone,
44{
45    fn default() -> Self {
46        Self::new()
47    }
48}
49
50impl<V> Store<V> for MemoryStore<V>
51where
52    V: Clone + Send + Sync + 'static,
53{
54    fn append(&self, value: V) -> AppResult<()> {
55        let mut state = self.state.lock();
56        state.values.push_back(value);
57        state.last_activity = Instant::now();
58        Ok(())
59    }
60
61    fn pop_oldest(&self) -> AppResult<Option<V>> {
62        let mut state = self.state.lock();
63        let value = state.values.pop_front();
64        if value.is_some() {
65            state.last_activity = Instant::now();
66        }
67        Ok(value)
68    }
69
70    fn snapshot(&self) -> AppResult<Vec<V>> {
71        let state = self.state.lock();
72        Ok(state.values.iter().cloned().collect())
73    }
74
75    fn flush(&self) -> AppResult<Vec<V>> {
76        let mut state = self.state.lock();
77        let values = state.values.drain(..).collect();
78        state.last_activity = Instant::now();
79        Ok(values)
80    }
81
82    fn size(&self) -> AppResult<usize> {
83        Ok(self.state.lock().values.len())
84    }
85
86    fn touch(&self) -> AppResult<()> {
87        self.state.lock().last_activity = Instant::now();
88        Ok(())
89    }
90
91    fn last_activity(&self) -> AppResult<Instant> {
92        Ok(self.state.lock().last_activity)
93    }
94
95    fn close(&self) -> AppResult<()> {
96        self.state.lock().values.clear();
97        Ok(())
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104
105    #[test]
106    fn memory_store_supports_fifo_operations() {
107        let store = MemoryStore::new();
108        store.append(1).unwrap();
109        store.append(2).unwrap();
110        assert_eq!(store.snapshot().unwrap(), vec![1, 2]);
111        assert_eq!(store.pop_oldest().unwrap(), Some(1));
112        assert_eq!(store.flush().unwrap(), vec![2]);
113    }
114}
rskit_stateful/memory_store.rs

rskit_stateful/
memory_store.rs
