use std::{
borrow::Borrow,
collections::{HashMap, HashSet},
fmt::{self, Debug},
hash::Hash,
sync::{Arc, Mutex, MutexGuard, PoisonError},
};
use super::{AppliedEvent, Entries, InMemoryView, Indexer, Shared, ViewError};
use crate::Envelope;
pub struct InMemoryIndex<K, T, V> {
view: Arc<Shared<K, T>>,
pub(super) index: Arc<Index<K, T, V>>,
}
pub(super) struct Index<K, T, V> {
value_of: Box<dyn Fn(&T) -> V + Send + Sync>,
keys: Mutex<HashMap<V, HashSet<K>>>,
}
impl<K, T> InMemoryView<K, T>
where
K: Clone + Debug + Eq + Hash + Send + Sync + 'static,
T: Clone + Send + Sync + 'static,
{
pub fn index<V>(
&self,
value_of: impl Fn(&T) -> V + Send + Sync + 'static,
) -> InMemoryIndex<K, T, V>
where
V: Eq + Hash + Send + Sync + 'static,
{
let index = Arc::new(Index {
value_of: Box::new(value_of),
keys: Mutex::new(HashMap::new()),
});
let mut entries = self.shared.write();
for (key, entry) in &entries.items {
index.add(key, &entry.item);
}
let indexer: Arc<dyn Indexer<K, T>> = index.clone();
entries.indexes.push(indexer);
InMemoryIndex {
view: Arc::clone(&self.shared),
index,
}
}
}
impl<K, T, V> InMemoryIndex<K, T, V>
where
K: Clone + Debug + Eq + Hash + Send + Sync + 'static,
T: Clone + Send + Sync + 'static,
V: Eq + Hash + Send + Sync + 'static,
{
pub fn lookup<Q>(&self, value: &Q) -> Vec<T>
where
V: Borrow<Q>,
Q: Eq + Hash + ?Sized,
{
let entries = self.view.read();
self.keys_of(&entries, value)
.iter()
.map(|key| entries.items[key].item.clone())
.collect()
}
pub fn update<E, Q>(
&self,
value: &Q,
event: &Envelope<E>,
change: impl FnMut(&mut T),
) -> Result<usize, ViewError>
where
V: Borrow<Q>,
Q: Eq + Hash + ?Sized,
{
let event = AppliedEvent::of(event)?;
let mut entries = self.view.write();
let keys = self.keys_of(&entries, value);
self.view.change_all(&mut entries, keys, &event, change)
}
pub fn delete<E, Q>(&self, value: &Q, event: &Envelope<E>) -> Result<usize, ViewError>
where
V: Borrow<Q>,
Q: Eq + Hash + ?Sized,
{
let event = AppliedEvent::of(event)?;
let mut entries = self.view.write();
let keys = self.keys_of(&entries, value);
Ok(entries.delete_all(&keys, &event))
}
fn keys_of<Q>(&self, entries: &Entries<K, T>, value: &Q) -> Vec<K>
where
V: Borrow<Q>,
Q: Eq + Hash + ?Sized,
{
let mut keys: Vec<(usize, K)> = self
.index
.keys()
.get(value)
.into_iter()
.flatten()
.filter_map(|key| Some((entries.items.get(key)?.position, key.clone())))
.collect();
keys.sort_unstable_by_key(|(position, _)| *position);
keys.into_iter().map(|(_, key)| key).collect()
}
}
impl<K, T, V> Index<K, T, V> {
pub(super) fn keys(&self) -> MutexGuard<'_, HashMap<V, HashSet<K>>> {
self.keys.lock().unwrap_or_else(PoisonError::into_inner)
}
}
impl<K, T, V> Indexer<K, T> for Index<K, T, V>
where
K: Clone + Eq + Hash + Send + Sync,
V: Eq + Hash + Send + Sync,
{
fn add(&self, key: &K, item: &T) {
let value = (self.value_of)(item);
let _ = self.keys().entry(value).or_default().insert(key.clone());
}
fn remove(&self, key: &K, item: &T) {
let value = (self.value_of)(item);
let mut keys = self.keys();
if let Some(keys_of_value) = keys.get_mut(&value) {
let _ = keys_of_value.remove(key);
if keys_of_value.is_empty() {
let _ = keys.remove(&value);
}
}
}
}
impl<K, T, V> Debug for InMemoryIndex<K, T, V> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("InMemoryIndex")
.field("values", &self.index.keys().len())
.finish_non_exhaustive()
}
}