hara_native/lang/data/

atom.rs

use std::fmt;
use std::sync::{Arc, Mutex};

use crate::lang::protocol::{IDeref, IDisplay, IReset, IWatch};

type Validator<V> = Arc<dyn Fn(&V) -> bool + Send + Sync>;
type Watch<V> = Arc<dyn Fn(&WatchEntry<V>) + Send + Sync>;

#[derive(Clone)]
pub struct WatchEntry<V> {
    pub key: String,
    pub old_value: V,
    pub new_value: V,
}

#[derive(Clone)]
pub struct Atom<V> {
    state: Arc<Mutex<V>>,
    validator: Option<Validator<V>>,
    watches: Arc<Mutex<Vec<(String, Watch<V>)>>>,
}
impl<V> Atom<V> {
    pub fn new(value: V) -> Self {
        Self {
            state: Arc::new(Mutex::new(value)),
            validator: None,
            watches: Arc::new(Mutex::new(Vec::new())),
        }
    }
    pub fn same_identity(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.state, &other.state)
    }
    pub fn identity_address(&self) -> usize {
        Arc::as_ptr(&self.state) as usize
    }
    pub fn with_validator(
        value: V,
        validator: impl Fn(&V) -> bool + Send + Sync + 'static,
    ) -> Self {
        Self {
            validator: Some(Arc::new(validator)),
            ..Self::new(value)
        }
    }
    pub fn add_watch(
        &self,
        key: impl Into<String>,
        watch: impl Fn(&WatchEntry<V>) + Send + Sync + 'static,
    ) {
        let key = key.into();
        let mut watches = self.watches.lock().expect("atom watches");
        watches.retain(|(candidate, _)| candidate != &key);
        watches.push((key, Arc::new(watch)));
    }
    pub fn remove_watch(&self, key: &str) {
        self.watches
            .lock()
            .expect("atom watches")
            .retain(|(candidate, _)| candidate != key);
    }
    fn accepts(&self, value: &V) -> bool {
        self.validator
            .as_ref()
            .map_or(true, |validator| validator(value))
    }
}
impl<V: Clone> Atom<V> {
    pub fn deref_value(&self) -> V {
        self.state.lock().expect("atom state").clone()
    }
    pub fn reset(&self, new_value: V) -> Result<V, String> {
        if !self.accepts(&new_value) {
            return Err("atom validator rejected value".into());
        }
        let old_value = {
            let mut state = self.state.lock().expect("atom state");
            std::mem::replace(&mut *state, new_value.clone())
        };
        self.notify(old_value, new_value.clone());
        Ok(new_value)
    }
    pub fn swap(&self, f: impl FnOnce(&V) -> V) -> Result<V, String> {
        let (old_value, new_value) = {
            let mut state = self.state.lock().expect("atom state");
            let old = state.clone();
            let new = f(&old);
            if !self.accepts(&new) {
                return Err("atom validator rejected value".into());
            }
            *state = new.clone();
            (old, new)
        };
        self.notify(old_value, new_value.clone());
        Ok(new_value)
    }
    fn notify(&self, old_value: V, new_value: V) {
        for (key, watch) in self.watches.lock().expect("atom watches").iter() {
            watch(&WatchEntry {
                key: key.clone(),
                old_value: old_value.clone(),
                new_value: new_value.clone(),
            });
        }
    }
}
impl<V: Clone + PartialEq> Atom<V> {
    pub fn compare_and_set(&self, old: &V, new: V) -> Result<bool, String> {
        if !self.accepts(&new) {
            return Ok(false);
        }
        let prior = {
            let mut state = self.state.lock().expect("atom state");
            if &*state != old {
                return Ok(false);
            }
            let prior = state.clone();
            *state = new.clone();
            prior
        };
        self.notify(prior, new);
        Ok(true)
    }
}
impl<V: Clone> IDeref for Atom<V> {
    type Output = V;
    fn deref(&self) -> V {
        self.deref_value()
    }
}
impl<V: Clone> IReset<V> for Atom<V> {
    type Error = String;

    fn reset(&self, value: V) -> Result<V, Self::Error> {
        Atom::reset(self, value)
    }
}
impl<V: Clone> IWatch<V> for Atom<V> {
    type Key = String;
    type WatchEntry = WatchEntry<V>;

    fn add_watch(&self, key: Self::Key, watch: impl Fn(&Self::WatchEntry) + Send + Sync + 'static) {
        Atom::add_watch(self, key, watch);
    }

    fn remove_watch(&self, key: &Self::Key) {
        Atom::remove_watch(self, key);
    }

    fn notify_watches(&self, old_value: V, new_value: V) {
        self.notify(old_value, new_value);
    }
}
impl<V: fmt::Display + Clone> IDisplay for Atom<V> {
    fn display(&self) -> String {
        format!("#atom <{}>", self.deref_value())
    }
}
impl<V: fmt::Display + Clone> fmt::Display for Atom<V> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.display())
    }
}
impl<V> fmt::Debug for Atom<V> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Atom").finish_non_exhaustive()
    }
}

#[cfg(test)]
mod tests {
    use super::Atom;
    use crate::lang::protocol::{IReset, IWatch};
    use std::sync::{Arc, Mutex};
    #[test]
    fn validates_swaps_and_notifies() {
        let atom = Atom::with_validator(1, |v| *v >= 0);
        let seen = Arc::new(Mutex::new(Vec::new()));
        let output = seen.clone();
        atom.add_watch("test", move |event| {
            output
                .lock()
                .unwrap()
                .push((event.old_value, event.new_value))
        });
        assert_eq!(atom.swap(|v| v + 2).unwrap(), 3);
        assert!(atom.reset(-1).is_err());
        assert_eq!(&*seen.lock().unwrap(), &[(1, 3)]);
        assert_eq!(IReset::reset(&atom, 4).unwrap(), 4);
        IWatch::remove_watch(&atom, &"test".to_string());
        IWatch::notify_watches(&atom, 4, 5);
        assert_eq!(seen.lock().unwrap().len(), 2);
    }
}