spire_core/dataset/
memory.rs

1use std::collections::VecDeque;
2use std::fmt;
3use std::future::{Future, ready};
4use std::sync::{Arc, Mutex};
5
6use crate::Result;
7use crate::dataset::Dataset;
8use crate::sync::lock;
9
10/// An in-memory [`Dataset`] read in insertion or reverse order.
11///
12/// Clones share the same storage.
13pub struct InMemDataset<T> {
14    inner: Arc<Inner<T>>,
15}
16
17struct Inner<T> {
18    items: Mutex<VecDeque<T>>,
19    lifo: bool,
20}
21
22impl<T> InMemDataset<T> {
23    /// Creates a dataset read in insertion order.
24    #[must_use]
25    pub fn new() -> Self {
26        Self::queue()
27    }
28
29    /// Creates a dataset read in insertion order.
30    #[must_use]
31    pub fn queue() -> Self {
32        Self::with_order(false)
33    }
34
35    /// Creates a dataset that reads the most recent item first.
36    #[must_use]
37    pub fn stack() -> Self {
38        Self::with_order(true)
39    }
40
41    fn with_order(lifo: bool) -> Self {
42        let inner = Inner {
43            items: Mutex::new(VecDeque::new()),
44            lifo,
45        };
46
47        Self {
48            inner: Arc::new(inner),
49        }
50    }
51}
52
53impl<T> Clone for InMemDataset<T> {
54    fn clone(&self) -> Self {
55        Self {
56            inner: self.inner.clone(),
57        }
58    }
59}
60
61impl<T> Default for InMemDataset<T> {
62    fn default() -> Self {
63        Self::new()
64    }
65}
66
67impl<T> FromIterator<T> for InMemDataset<T> {
68    fn from_iter<I: IntoIterator<Item = T>>(iter: I) -> Self {
69        let dataset = Self::queue();
70        lock(&dataset.inner.items).extend(iter);
71        dataset
72    }
73}
74
75impl<T> fmt::Debug for InMemDataset<T> {
76    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
77        f.debug_struct("InMemDataset")
78            .field("len", &lock(&self.inner.items).len())
79            .field("lifo", &self.inner.lifo)
80            .finish()
81    }
82}
83
84impl<T: Send + 'static> Dataset<T> for InMemDataset<T> {
85    fn write(&self, item: T) -> impl Future<Output = Result<()>> + Send {
86        lock(&self.inner.items).push_back(item);
87        ready(Ok(()))
88    }
89
90    fn read(&self) -> impl Future<Output = Result<Option<T>>> + Send {
91        let mut items = lock(&self.inner.items);
92        let item = if self.inner.lifo {
93            items.pop_back()
94        } else {
95            items.pop_front()
96        };
97        ready(Ok(item))
98    }
99
100    fn size_hint(&self) -> Option<usize> {
101        Some(lock(&self.inner.items).len())
102    }
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108
109    #[tokio::test]
110    async fn queue_and_stack_order() {
111        let queue: InMemDataset<u32> = [1, 2].into_iter().collect();
112        let stack = InMemDataset::stack();
113        stack.write(1).await.unwrap();
114        stack.write(2).await.unwrap();
115
116        assert_eq!(queue.read().await.unwrap(), Some(1));
117        assert_eq!(stack.read().await.unwrap(), Some(2));
118        assert_eq!(queue.size_hint(), Some(1));
119    }
120
121    #[tokio::test]
122    async fn clones_share_storage() {
123        let dataset = InMemDataset::new();
124        dataset.clone().write("item").await.unwrap();
125        assert_eq!(dataset.read().await.unwrap(), Some("item"));
126        assert_eq!(dataset.read().await.unwrap(), None);
127    }
128}
spire_core/dataset/memory.rs

spire_core/dataset/
memory.rs
