use crate::AmeBackend;
use crate::primitives::map_core::{ReactiveMapKey, ReactiveMapValue};
use crate::{MapChange, ReactiveMapCore};
use serde::de::DeserializeOwned;
use std::fmt::Display;
use std::str::FromStr;
use std::sync::Arc;
use uuid::Uuid;
pub fn map_get<B, K, V>(backend: &B, path: &str, key: &K) -> Result<Option<V>, B::Error>
where
B: AmeBackend,
K: Display,
V: DeserializeOwned,
{
backend.get(&format!("{}.{}", path, key))
}
pub fn map_contains_key<B, K, V>(backend: &B, path: &str, key: &K) -> Result<bool, B::Error>
where
B: AmeBackend,
K: Display,
V: DeserializeOwned,
{
map_get::<B, K, V>(backend, path, key).map(|v| v.is_some())
}
pub fn map_entries<B, K, V>(backend: &B, path: &str) -> Result<Vec<(K, V)>, B::Error>
where
B: AmeBackend,
K: FromStr,
V: DeserializeOwned + Default,
{
let prefix = format!("{}.", path);
let kvs = backend.scan_prefix(&prefix)?;
let mut results = Vec::new();
for (full_path, raw) in kvs {
if let Some(key_str) = full_path.strip_prefix(&prefix)
&& let Ok(k) = K::from_str(key_str)
&& let Ok(v) = backend.decode::<V>(&raw)
{
results.push((k, v));
}
}
Ok(results)
}
pub fn map_len<B>(backend: &B, path: &str) -> Result<usize, B::Error>
where
B: AmeBackend,
{
backend
.scan_prefix(&format!("{}.", path))
.map(|kvs| kvs.len())
}
pub fn map_set_existing<B, K, V>(
backend: &B,
core: &ReactiveMapCore<K, V>,
path: Arc<str>,
key: K,
value: &V,
notify_after_commit: bool,
source: Option<Uuid>,
) -> Result<(), B::Error>
where
B: AmeBackend,
K: ReactiveMapKey,
V: ReactiveMapValue,
{
let full_path = format!("{}.{}", path, key);
let old_value = match backend.get::<V>(&full_path)? {
Some(old_value) => old_value,
None => return Err(backend.key_not_found(key.to_string())),
};
let change = MapChange::Update {
key,
old_value,
new_value: value.clone(),
source,
};
map_apply_change(backend, core, path, change, notify_after_commit)
}
pub fn map_set_or_create<B, K, V>(
backend: &B,
core: &ReactiveMapCore<K, V>,
path: Arc<str>,
key: K,
value: &V,
notify_after_commit: bool,
source: Option<Uuid>,
) -> Result<(), B::Error>
where
B: AmeBackend,
K: ReactiveMapKey,
V: ReactiveMapValue,
{
let full_path = format!("{}.{}", path, key);
let old_value = backend.get::<V>(&full_path)?;
let change = if let Some(old_value) = old_value {
MapChange::Update {
key,
old_value,
new_value: value.clone(),
source,
}
} else {
MapChange::Insert {
key,
value: value.clone(),
source,
}
};
map_apply_change(backend, core, path, change, notify_after_commit)
}
pub fn map_remove<B, K, V>(
backend: &B,
core: &ReactiveMapCore<K, V>,
path: Arc<str>,
key: K,
notify_after_commit: bool,
source: Option<Uuid>,
) -> Result<Option<V>, B::Error>
where
B: AmeBackend,
K: ReactiveMapKey,
V: ReactiveMapValue,
{
let exists = core.cache.lock().unwrap().contains_key(&key);
if !exists {
return Ok(None);
}
let full_path = format!("{}.{}", path, key);
let old_value = backend.get::<V>(&full_path)?;
if let Some(old_value) = old_value {
let change = MapChange::Remove {
key,
old_value: old_value.clone(),
source,
};
map_apply_change(backend, core, path, change, notify_after_commit)?;
Ok(Some(old_value))
} else {
core.cache.lock().unwrap().remove(&key);
Ok(None)
}
}
pub fn map_clear<B, K, V>(
backend: &B,
core: &ReactiveMapCore<K, V>,
path: Arc<str>,
notify_after_commit: bool,
source: Option<Uuid>,
) -> Result<(), B::Error>
where
B: AmeBackend,
K: ReactiveMapKey,
V: ReactiveMapValue,
{
map_apply_change(
backend,
core,
path,
MapChange::Clear { source },
notify_after_commit,
)
}
pub fn map_apply_change<B, K, V>(
backend: &B,
core: &ReactiveMapCore<K, V>,
path: Arc<str>,
change: MapChange<K, V>,
notify_after_commit: bool,
) -> Result<(), B::Error>
where
B: AmeBackend,
K: ReactiveMapKey,
V: ReactiveMapValue,
{
let context_path: Arc<str> = match change.key() {
Some(key) => format!("{}.{}", path, key).into(),
None => path.clone(),
};
let processed = core
.run_interceptors(context_path, change)
.map_err(|_| backend.intercepted())?;
match &processed {
MapChange::Insert { key, value, .. }
| MapChange::Update {
key,
new_value: value,
..
} => {
backend.set_with_source(&format!("{}.{}", path, key), value, processed.source())?;
}
MapChange::Remove { key, .. } => {
backend.delete_with_source(&format!("{}.{}", path, key), processed.source())?;
}
MapChange::Clear { .. } => {
let prefix = format!("{}.", path);
let kvs = backend.scan_prefix(&prefix)?;
for (full_path, _) in kvs {
backend.delete_with_source(&full_path, processed.source())?;
}
}
}
map_apply_remote_change(core, &processed);
if notify_after_commit {
core.notify(&processed);
}
Ok(())
}
pub fn map_apply_remote_change<K, V>(core: &ReactiveMapCore<K, V>, change: &MapChange<K, V>)
where
K: ReactiveMapKey,
V: ReactiveMapValue,
{
let mut keys = core.cache.lock().unwrap();
match change {
MapChange::Insert { key, value, .. }
| MapChange::Update {
key,
new_value: value,
..
} => {
keys.insert(key.clone(), value.clone());
}
MapChange::Remove { key, .. } => {
keys.remove(key);
}
MapChange::Clear { .. } => {
keys.clear();
}
}
}