rust_elm/infra/
shared.rs

1use std::collections::HashMap;
2use std::fmt;
3use std::sync::Arc;
4
5use crossbeam_channel::{Receiver, Sender};
6use parking_lot::Mutex;
7
8/// Errors from [`Storage`] backends.
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub enum StorageError {
11    NotFound,
12    Io(String),
13    #[cfg(feature = "serde")]
14    Serde(String),
15}
16
17impl fmt::Display for StorageError {
18    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
19        match self {
20            Self::NotFound => write!(f, "storage key not found"),
21            Self::Io(msg) => write!(f, "io error: {msg}"),
22            #[cfg(feature = "serde")]
23            Self::Serde(msg) => write!(f, "serde error: {msg}"),
24        }
25    }
26}
27
28impl std::error::Error for StorageError {}
29
30/// Key-value persistence for [`Shared`] values (UDF `PersistenceKey` engine subset).
31pub trait Storage<T> {
32    fn save(&self, key: &str, value: &T) -> Result<(), StorageError>;
33    fn load(&self, key: &str) -> Result<Option<T>, StorageError>;
34    fn remove(&self, key: &str) -> Result<(), StorageError> {
35        let _ = key;
36        Ok(())
37    }
38}
39
40/// In-process storage — useful in tests and previews.
41#[derive(Debug, Clone)]
42pub struct InMemoryStorage<T> {
43    data: Arc<Mutex<HashMap<String, T>>>,
44}
45
46impl<T> Default for InMemoryStorage<T> {
47    fn default() -> Self {
48        Self {
49            data: Arc::new(Mutex::new(HashMap::new())),
50        }
51    }
52}
53
54impl<T: Clone> InMemoryStorage<T> {
55    pub fn new() -> Self {
56        Self::default()
57    }
58}
59
60impl<T: Clone> Storage<T> for InMemoryStorage<T> {
61    fn save(&self, key: &str, value: &T) -> Result<(), StorageError> {
62        self.data.lock().insert(key.to_owned(), value.clone());
63        Ok(())
64    }
65
66    fn load(&self, key: &str) -> Result<Option<T>, StorageError> {
67        Ok(self.data.lock().get(key).cloned())
68    }
69
70    fn remove(&self, key: &str) -> Result<(), StorageError> {
71        self.data.lock().remove(key);
72        Ok(())
73    }
74}
75
76/// JSON file storage under a directory root (requires `serde` feature).
77#[cfg(feature = "serde")]
78use std::path::PathBuf;
79
80#[cfg(feature = "serde")]
81#[derive(Debug, Clone)]
82pub struct FileStorage {
83    root: PathBuf,
84}
85
86#[cfg(feature = "serde")]
87impl FileStorage {
88    pub fn new(root: impl Into<PathBuf>) -> Self {
89        Self { root: root.into() }
90    }
91
92    fn path_for(&self, key: &str) -> PathBuf {
93        self.root.join(format!("{key}.json"))
94    }
95}
96
97#[cfg(feature = "serde")]
98impl FileStorage {
99    pub fn ensure_root(&self) -> Result<(), StorageError> {
100        std::fs::create_dir_all(&self.root)
101            .map_err(|err| StorageError::Io(err.to_string()))
102    }
103}
104
105#[cfg(feature = "serde")]
106impl<T> Storage<T> for FileStorage
107where
108    T: serde::Serialize + for<'de> serde::Deserialize<'de>,
109{
110    fn save(&self, key: &str, value: &T) -> Result<(), StorageError> {
111        self.ensure_root()?;
112        let json = serde_json::to_string_pretty(value)
113            .map_err(|err| StorageError::Serde(err.to_string()))?;
114        std::fs::write(self.path_for(key), json).map_err(|err| StorageError::Io(err.to_string()))
115    }
116
117    fn load(&self, key: &str) -> Result<Option<T>, StorageError> {
118        let path = self.path_for(key);
119        if !path.exists() {
120            return Ok(None);
121        }
122        let bytes = std::fs::read(&path).map_err(|err| StorageError::Io(err.to_string()))?;
123        let value = serde_json::from_slice(&bytes)
124            .map_err(|err| StorageError::Serde(err.to_string()))?;
125        Ok(Some(value))
126    }
127
128    fn remove(&self, key: &str) -> Result<(), StorageError> {
129        let path = self.path_for(key);
130        if path.exists() {
131            std::fs::remove_file(path).map_err(|err| StorageError::Io(err.to_string()))?;
132        }
133        Ok(())
134    }
135}
136
137struct SharedInner<T> {
138    value: Mutex<T>,
139    listeners: Mutex<Vec<Sender<()>>>,
140}
141
142/// Ref-counted shared value with change notification (UDF `Shared` engine subset).
143#[derive(Clone)]
144pub struct Shared<T> {
145    inner: Arc<SharedInner<T>>,
146}
147
148impl<T> Shared<T> {
149    pub fn new(value: T) -> Self {
150        Self {
151            inner: Arc::new(SharedInner {
152                value: Mutex::new(value),
153                listeners: Mutex::new(Vec::new()),
154            }),
155        }
156    }
157
158    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
159        f(&*self.inner.value.lock())
160    }
161
162    pub fn with_mut<R>(&self, f: impl FnOnce(&mut T) -> R) -> R
163    where
164        T: Clone + PartialEq,
165    {
166        let mut guard = self.inner.value.lock();
167        let before = guard.clone();
168        let result = f(&mut guard);
169        let changed = before != *guard;
170        drop(guard);
171        if changed {
172            self.notify();
173        }
174        result
175    }
176
177    pub fn set(&self, value: T)
178    where
179        T: PartialEq,
180    {
181        let mut guard = self.inner.value.lock();
182        if *guard != value {
183            *guard = value;
184            drop(guard);
185            self.notify();
186        }
187    }
188
189    pub fn get(&self) -> T
190    where
191        T: Clone,
192    {
193        self.inner.value.lock().clone()
194    }
195
196    pub fn load<S>(storage: &S, key: &str, default: T) -> Result<Self, StorageError>
197    where
198        S: Storage<T>,
199    {
200        Ok(Self::new(storage.load(key)?.unwrap_or(default)))
201    }
202
203    pub fn persist<S>(&self, storage: &S, key: &str) -> Result<(), StorageError>
204    where
205        S: Storage<T>,
206        T: Clone,
207    {
208        storage.save(key, &self.get())
209    }
210
211    pub fn subscribe(&self) -> SharedSubscriber<T>
212    where
213        T: Clone + PartialEq,
214    {
215        let (tx, rx) = crossbeam_channel::unbounded();
216        self.inner.listeners.lock().push(tx);
217        SharedSubscriber {
218            shared: self.clone(),
219            rx,
220            last: Some(self.get()),
221        }
222    }
223
224    fn notify(&self) {
225        self.inner
226            .listeners
227            .lock()
228            .retain(|tx| tx.send(()).is_ok());
229    }
230}
231
232/// Subscription to a [`Shared`] value — deduplicates consecutive equal snapshots.
233pub struct SharedSubscriber<T> {
234    shared: Shared<T>,
235    rx: Receiver<()>,
236    last: Option<T>,
237}
238
239impl<T: Clone + PartialEq> SharedSubscriber<T> {
240    #[allow(clippy::should_implement_trait)]
241    pub fn next(&mut self) -> Option<T> {
242        loop {
243            match self.rx.try_recv() {
244                Ok(()) => {
245                    while self.rx.try_recv().is_ok() {}
246                    let snapshot = self.shared.get();
247                    if self.last.as_ref() == Some(&snapshot) {
248                        continue;
249                    }
250                    self.last = Some(snapshot.clone());
251                    return Some(snapshot);
252                }
253                Err(crossbeam_channel::TryRecvError::Empty) => return None,
254                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
255            }
256        }
257    }
258
259    pub fn latest(&self) -> Option<T> {
260        self.last.clone()
261    }
262}
263
264#[cfg(test)]
265mod tests {
266    use super::*;
267    use crate::panic_on_state_clone;
268    use crate::test_support::{allow_state_clones, shared_get};
269
270    panic_on_state_clone! {
271        #[derive(Debug, PartialEq, Eq)]
272        #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
273        struct Counter {
274            n: i32,
275        }
276    }
277
278    #[test]
279    fn clones_share_underlying_value() {
280        let shared = Shared::new(Counter { n: 0 });
281        let scoped = shared.clone();
282        shared.set(Counter { n: 3 });
283        assert_eq!(shared_get(&scoped).n, 3);
284    }
285
286    #[test]
287    fn subscribe_notifies_on_change() {
288        let shared = Shared::new(Counter { n: 0 });
289        let mut sub = allow_state_clones(1, || shared.subscribe());
290        shared.set(Counter { n: 1 });
291        let next = allow_state_clones(2, || sub.next());
292        assert_eq!(next.map(|c| c.n), Some(1));
293        assert!(sub.next().is_none());
294    }
295
296    #[test]
297    fn in_memory_storage_round_trip() {
298        let storage = InMemoryStorage::new();
299        let shared = Shared::new(Counter { n: 7 });
300        let persist = allow_state_clones(2, || shared.persist(&storage, "counter"));
301        assert!(persist.is_ok(), "persist failed: {persist:?}");
302        let loaded = allow_state_clones(1, || Shared::load(&storage, "counter", Counter { n: 0 }));
303        let Ok(loaded) = loaded else {
304            panic!("load failed");
305        };
306        assert_eq!(shared_get(&loaded).n, 7);
307    }
308
309    #[cfg(feature = "serde")]
310    #[test]
311    fn file_storage_round_trip() {
312        let dir = std::env::temp_dir().join(format!("rust_elm_shared_{}", std::process::id()));
313        let _ = std::fs::remove_dir_all(&dir);
314        let storage = FileStorage::new(&dir);
315        let shared = Shared::new(Counter { n: 42 });
316        let persist = allow_state_clones(2, || shared.persist(&storage, "counter"));
317        assert!(persist.is_ok(), "persist failed: {persist:?}");
318        let loaded = allow_state_clones(1, || Shared::load(&storage, "counter", Counter { n: 0 }));
319        let Ok(loaded) = loaded else {
320            panic!("load failed");
321        };
322        assert_eq!(shared_get(&loaded).n, 42);
323        let _ = std::fs::remove_dir_all(&dir);
324    }
325}
rust_elm/infra/shared.rs

rust_elm/infra/
shared.rs
