use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;
use crossbeam_channel::{Receiver, Sender};
use parking_lot::Mutex;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StorageError {
NotFound,
Io(String),
#[cfg(feature = "serde")]
Serde(String),
}
impl fmt::Display for StorageError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::NotFound => write!(f, "storage key not found"),
Self::Io(msg) => write!(f, "io error: {msg}"),
#[cfg(feature = "serde")]
Self::Serde(msg) => write!(f, "serde error: {msg}"),
}
}
}
impl std::error::Error for StorageError {}
pub trait Storage<T> {
fn save(&self, key: &str, value: &T) -> Result<(), StorageError>;
fn load(&self, key: &str) -> Result<Option<T>, StorageError>;
fn remove(&self, key: &str) -> Result<(), StorageError> {
let _ = key;
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct InMemoryStorage<T> {
data: Arc<Mutex<HashMap<String, T>>>,
}
impl<T> Default for InMemoryStorage<T> {
fn default() -> Self {
Self {
data: Arc::new(Mutex::new(HashMap::new())),
}
}
}
impl<T: Clone> InMemoryStorage<T> {
pub fn new() -> Self {
Self::default()
}
}
impl<T: Clone> Storage<T> for InMemoryStorage<T> {
fn save(&self, key: &str, value: &T) -> Result<(), StorageError> {
self.data.lock().insert(key.to_owned(), value.clone());
Ok(())
}
fn load(&self, key: &str) -> Result<Option<T>, StorageError> {
Ok(self.data.lock().get(key).cloned())
}
fn remove(&self, key: &str) -> Result<(), StorageError> {
self.data.lock().remove(key);
Ok(())
}
}
#[cfg(feature = "serde")]
use std::path::PathBuf;
#[cfg(feature = "serde")]
#[derive(Debug, Clone)]
pub struct FileStorage {
root: PathBuf,
}
#[cfg(feature = "serde")]
impl FileStorage {
pub fn new(root: impl Into<PathBuf>) -> Self {
Self { root: root.into() }
}
fn path_for(&self, key: &str) -> PathBuf {
self.root.join(format!("{key}.json"))
}
}
#[cfg(feature = "serde")]
impl FileStorage {
pub fn ensure_root(&self) -> Result<(), StorageError> {
std::fs::create_dir_all(&self.root)
.map_err(|err| StorageError::Io(err.to_string()))
}
}
#[cfg(feature = "serde")]
impl<T> Storage<T> for FileStorage
where
T: serde::Serialize + for<'de> serde::Deserialize<'de>,
{
fn save(&self, key: &str, value: &T) -> Result<(), StorageError> {
self.ensure_root()?;
let json = serde_json::to_string_pretty(value)
.map_err(|err| StorageError::Serde(err.to_string()))?;
std::fs::write(self.path_for(key), json).map_err(|err| StorageError::Io(err.to_string()))
}
fn load(&self, key: &str) -> Result<Option<T>, StorageError> {
let path = self.path_for(key);
if !path.exists() {
return Ok(None);
}
let bytes = std::fs::read(&path).map_err(|err| StorageError::Io(err.to_string()))?;
let value = serde_json::from_slice(&bytes)
.map_err(|err| StorageError::Serde(err.to_string()))?;
Ok(Some(value))
}
fn remove(&self, key: &str) -> Result<(), StorageError> {
let path = self.path_for(key);
if path.exists() {
std::fs::remove_file(path).map_err(|err| StorageError::Io(err.to_string()))?;
}
Ok(())
}
}
struct SharedInner<T> {
value: Mutex<T>,
listeners: Mutex<Vec<Sender<()>>>,
}
#[derive(Clone)]
pub struct Shared<T> {
inner: Arc<SharedInner<T>>,
}
impl<T> Shared<T> {
pub fn new(value: T) -> Self {
Self {
inner: Arc::new(SharedInner {
value: Mutex::new(value),
listeners: Mutex::new(Vec::new()),
}),
}
}
pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
f(&*self.inner.value.lock())
}
pub fn with_mut<R>(&self, f: impl FnOnce(&mut T) -> R) -> R
where
T: Clone + PartialEq,
{
let mut guard = self.inner.value.lock();
let before = guard.clone();
let result = f(&mut guard);
let changed = before != *guard;
drop(guard);
if changed {
self.notify();
}
result
}
pub fn set(&self, value: T)
where
T: PartialEq,
{
let mut guard = self.inner.value.lock();
if *guard != value {
*guard = value;
drop(guard);
self.notify();
}
}
pub fn get(&self) -> T
where
T: Clone,
{
self.inner.value.lock().clone()
}
pub fn load<S>(storage: &S, key: &str, default: T) -> Result<Self, StorageError>
where
S: Storage<T>,
{
Ok(Self::new(storage.load(key)?.unwrap_or(default)))
}
pub fn persist<S>(&self, storage: &S, key: &str) -> Result<(), StorageError>
where
S: Storage<T>,
T: Clone,
{
storage.save(key, &self.get())
}
pub fn subscribe(&self) -> SharedSubscriber<T>
where
T: Clone + PartialEq,
{
let (tx, rx) = crossbeam_channel::unbounded();
self.inner.listeners.lock().push(tx);
SharedSubscriber {
shared: self.clone(),
rx,
last: Some(self.get()),
}
}
fn notify(&self) {
self.inner
.listeners
.lock()
.retain(|tx| tx.send(()).is_ok());
}
}
pub struct SharedSubscriber<T> {
shared: Shared<T>,
rx: Receiver<()>,
last: Option<T>,
}
impl<T: Clone + PartialEq> SharedSubscriber<T> {
#[allow(clippy::should_implement_trait)]
pub fn next(&mut self) -> Option<T> {
loop {
match self.rx.try_recv() {
Ok(()) => {
while self.rx.try_recv().is_ok() {}
let snapshot = self.shared.get();
if self.last.as_ref() == Some(&snapshot) {
continue;
}
self.last = Some(snapshot.clone());
return Some(snapshot);
}
Err(crossbeam_channel::TryRecvError::Empty) => return None,
Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
}
}
}
pub fn latest(&self) -> Option<T> {
self.last.clone()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::panic_on_state_clone;
use crate::test_support::{allow_state_clones, shared_get};
panic_on_state_clone! {
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
struct Counter {
n: i32,
}
}
#[test]
fn clones_share_underlying_value() {
let shared = Shared::new(Counter { n: 0 });
let scoped = shared.clone();
shared.set(Counter { n: 3 });
assert_eq!(shared_get(&scoped).n, 3);
}
#[test]
fn subscribe_notifies_on_change() {
let shared = Shared::new(Counter { n: 0 });
let mut sub = allow_state_clones(1, || shared.subscribe());
shared.set(Counter { n: 1 });
let next = allow_state_clones(2, || sub.next());
assert_eq!(next.map(|c| c.n), Some(1));
assert!(sub.next().is_none());
}
#[test]
fn in_memory_storage_round_trip() {
let storage = InMemoryStorage::new();
let shared = Shared::new(Counter { n: 7 });
allow_state_clones(2, || shared.persist(&storage, "counter")).unwrap();
let loaded =
allow_state_clones(1, || Shared::load(&storage, "counter", Counter { n: 0 })).unwrap();
assert_eq!(shared_get(&loaded).n, 7);
}
#[cfg(feature = "serde")]
#[test]
fn file_storage_round_trip() {
let dir = std::env::temp_dir().join(format!("rust_elm_shared_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let storage = FileStorage::new(&dir);
let shared = Shared::new(Counter { n: 42 });
allow_state_clones(2, || shared.persist(&storage, "counter")).unwrap();
let loaded =
allow_state_clones(1, || Shared::load(&storage, "counter", Counter { n: 0 })).unwrap();
assert_eq!(shared_get(&loaded).n, 42);
let _ = std::fs::remove_dir_all(&dir);
}
}