Add "update_if_exists", "update_mut_if_exists", "remove_if" methods (#6)

* Add "update_if_exists", "update_mut_if_exists", "remove_if" methods

* No need for Arc and Pin there, only Box

* Create tests badge in readme

Author: ByteAlex

README.md

@@ -1,4 +1,6 @@
 # cache-loader-async
+[![Tests](https://github.com/ZeroTwo-Bot/cache-loader-async-rs/actions/workflows/rust.yml/badge.svg?branch=master&event=push)](https://github.com/ZeroTwo-Bot/cache-loader-async-rs/actions/workflows/rust.yml)
+
 [crates.io](https://crates.io/crates/cache_loader_async)
 
 The goal of this crate is to provide a thread-safe and easy way to access any data structure
@@ -84,5 +86,6 @@ pub trait CacheBacking<K, V>
     fn set(&mut self, key: K, value: V) -> Option<V>;
     fn remove(&mut self, key: &K) -> Option<V>;
     fn contains_key(&self, key: &K) -> bool;
+    fn remove_if(&mut self, predicate: Box<dyn Fn((&K, &V)) -> bool + Send + 'static>);
 }
 ```

src/backing.rs

@@ -17,6 +17,7 @@ pub trait CacheBacking<K, V>
     fn set(&mut self, key: K, value: V) -> Option<V>;
     fn remove(&mut self, key: &K) -> Option<V>;
     fn contains_key(&self, key: &K) -> bool;
+    fn remove_if(&mut self, predicate: Box<dyn Fn((&K, &V)) -> bool + Send + 'static>);
 }
 
 #[cfg(feature = "lru-cache")]
@@ -48,6 +49,22 @@ impl<
     fn contains_key(&self, key: &K) -> bool {
         self.lru.contains(&key.clone())
     }
+
+    fn remove_if(&mut self, predicate: Box<dyn Fn((&K, &V)) -> bool + Send>) {
+        let keys = self.lru.iter()
+            .filter_map(|(key, value)| {
+                if predicate((key, value)) {
+                    Some(key)
+                } else {
+                    None
+                }
+            })
+            .cloned()
+            .collect::<Vec<K>>();
+        for key in keys.into_iter(){
+            self.lru.pop(&key);
+        }
+    }
 }
 
 #[cfg(feature = "lru-cache")]
@@ -118,6 +135,23 @@ impl<
             .filter(|(_, expiry)| SystemTime::now().lt(expiry))
             .is_some()
     }
+
+    fn remove_if(&mut self, predicate: Box<dyn Fn((&K, &V)) -> bool + Send>) {
+        let keys = self.map.iter()
+            .filter_map(|(key, (value, _))| {
+                if predicate((key, value)) {
+                    Some(key)
+                } else {
+                    None
+                }
+            })
+            .cloned()
+            .collect::<Vec<K>>();
+        for key in keys.into_iter() {
+            self.map.remove(&key);
+            self.expiry_queue.retain(|(expiry_key, _)| expiry_key.ne(&key))
+        }
+    }
 }
 
 #[cfg(feature = "ttl-cache")]
@@ -169,6 +203,10 @@ impl<
     fn contains_key(&self, key: &K) -> bool {
         self.map.contains_key(key)
     }
+
+    fn remove_if(&mut self, predicate: Box<dyn Fn((&K, &V)) -> bool + Send>) {
+        self.map.retain(|k, v| !predicate((k, v)));
+    }
 }
 
 impl<K, V> HashMapBacking<K, V> {

src/cache_api.rs

@@ -11,9 +11,10 @@ pub enum CacheLoadingError<E: Debug> {
     #[error(transparent)]
     CommunicationError(CacheCommunicationError),
     #[error("No data found")]
-    NoData(), // todo better handling here? eventually return loadingerror if possible
+    NoData(),
+    // todo better handling here? eventually return loadingerror if possible
     #[error("An error occurred when loading the entity from the loader function")]
-    LoadingError(E)
+    LoadingError(E),
 }
 
 #[derive(Error, Debug)]
@@ -31,7 +32,6 @@ pub enum CacheCommunicationError {
 }
 
 impl<E: Debug> CacheLoadingError<E> {
-
     pub fn as_loading_error(&self) -> Option<&E> {
         match self {
             CacheLoadingError::LoadingError(error) => Some(error),
@@ -204,6 +204,18 @@ impl<
             .map(|meta| meta.result)
     }
 
+    /// Retrieves or loads the value for specified key from either cache or loader function with
+    /// meta information, i.e. if the key was loaded from cache or from the loader function
+    ///
+    /// # Arguments
+    ///
+    /// * `key` - The key which should be loaded
+    ///
+    /// # Return Value
+    ///
+    /// Returns a Result with:
+    /// Ok - Value of type ResultMeta<V>
+    /// Err - Error of type CacheLoadingError
     pub async fn get_with_meta(&self, key: K) -> Result<ResultMeta<V>, CacheLoadingError<E>> {
         self.send_cache_action(CacheAction::Get(key)).await
             .map(|opt_result| opt_result.expect("Get should always return either V or CacheLoadingError"))
@@ -277,7 +289,23 @@ impl<
             .map(|opt_meta| opt_meta.map(|meta| meta.result))
     }
 
-    /// Updates a key on the cache with the given update function and returns the previous value
+    /// Removes all entries which match the specified predicate
+    ///
+    /// # Arguments
+    ///
+    /// * `predicate` - The predicate to test all entries against
+    ///
+    /// # Return Value
+    ///
+    /// Returns a Result with:
+    /// Ok - Nothing, the removed values are discarded
+    /// Err - Error of type CacheLoadingError -> the values were not discarded
+    pub async fn remove_if<P: Fn((&K, Option<&V>)) -> bool + Send + Sync + 'static>(&self, predicate: P) -> Result<(), CacheLoadingError<E>> {
+        self.send_cache_action(CacheAction::RemoveIf(Box::new(predicate))).await
+            .map(|_| ())
+    }
+
+    /// Updates a key on the cache with the given update function and returns the updated value
     ///
     /// If the key is not present yet, it'll be loaded using the loader function and will be
     /// updated once this loader function completes.
@@ -298,18 +326,82 @@ impl<
     /// Err - Error of type CacheLoadingError
     pub async fn update<U>(&self, key: K, update_fn: U) -> Result<V, CacheLoadingError<E>>
         where U: FnOnce(V) -> V + Send + 'static {
-        self.send_cache_action(CacheAction::Update(key, Box::new(update_fn))).await
+        self.send_cache_action(CacheAction::Update(key, Box::new(update_fn), true)).await
             .map(|opt_result| opt_result.expect("Get should always return either V or CacheLoadingError"))
             .map(|meta| meta.result)
     }
 
+    /// Updates a key on the cache with the given update function and returns the updated value if
+    /// it existed
+    ///
+    /// If the key is not present yet, it'll be ignored. This also counts for keys in LOADING state.
+    ///
+    /// # Arguments
+    ///
+    /// * `key` - The key which should be updated
+    /// * `update_fn` - A `FnOnce(V) -> V` which has the current value as parameter and should
+    ///                 return the updated value
+    ///
+    /// # Return Value
+    ///
+    /// Returns a Result with:
+    /// Ok - Optional value of type V which is the previously mapped value
+    /// Err - Error of type CacheLoadingError
+    pub async fn update_if_exists<U>(&self, key: K, update_fn: U) -> Result<Option<V>, CacheLoadingError<E>>
+        where U: FnOnce(V) -> V + Send + 'static {
+        self.send_cache_action(CacheAction::Update(key, Box::new(update_fn), false)).await
+            .map(|opt| opt.map(|meta| meta.result))
+    }
+
+    /// Updates a key on the cache with the given update function and returns the updated value
+    ///
+    /// If the key is not present yet, it'll be loaded using the loader function and will be
+    /// updated once this loader function completes.
+    /// In case the key was manually updated via `set` during the loader function the update will
+    /// take place on the manually updated value, so user-controlled input takes precedence over
+    /// the loader function
+    ///
+    /// # Arguments
+    ///
+    /// * `key` - The key which should be updated
+    /// * `update_fn` - A `FnMut(&mut V) -> ()` which has the current value as parameter and should
+    ///                 update it accordingly
+    ///
+    /// # Return Value
+    ///
+    /// Returns a Result with:
+    /// Ok - Value of type V
+    /// Err - Error of type CacheLoadingError
     pub async fn update_mut<U>(&self, key: K, update_fn: U) -> Result<V, CacheLoadingError<E>>
         where U: FnMut(&mut V) -> () + Send + 'static {
-        self.send_cache_action(CacheAction::UpdateMut(key, Box::new(update_fn))).await
+        self.send_cache_action(CacheAction::UpdateMut(key, Box::new(update_fn), true)).await
             .map(|opt_result| opt_result.expect("Get should always return either V or CacheLoadingError"))
             .map(|meta| meta.result)
     }
 
+    /// Updates a key on the cache with the given update function and returns the updated value if
+    /// it existed
+    ///
+    /// If the key is not present yet, it'll be ignored.
+    /// Keys in LOADING state will still be updated as they get available.
+    ///
+    /// # Arguments
+    ///
+    /// * `key` - The key which should be updated
+    /// * `update_fn` - A `FnMut(&mut V) -> ()` which has the current value as parameter and should
+    ///                 update it accordingly
+    ///
+    /// # Return Value
+    ///
+    /// Returns a Result with:
+    /// Ok - Optional value of type V
+    /// Err - Error of type CacheLoadingError
+    pub async fn update_mut_if_exists<U>(&self, key: K, update_fn: U) -> Result<Option<V>, CacheLoadingError<E>>
+        where U: FnMut(&mut V) -> () + Send + 'static {
+        self.send_cache_action(CacheAction::UpdateMut(key, Box::new(update_fn), false)).await
+            .map(|opt| opt.map(|meta| meta.result))
+    }
+
     async fn send_cache_action(&self, action: CacheAction<K, V>) -> Result<Option<ResultMeta<V>>, CacheLoadingError<E>> {
         let (tx, rx) = tokio::sync::oneshot::channel();
         match self.tx.send(CacheMessage {

src/internal_cache.rs

@@ -9,9 +9,10 @@ pub(crate) enum CacheAction<K, V> {
     GetIfPresent(K),
     Get(K),
     Set(K, V),
-    Update(K, Box<dyn FnOnce(V) -> V + Send + 'static>),
-    UpdateMut(K, Box<dyn FnMut(&mut V) -> () + Send + 'static>),
+    Update(K, Box<dyn FnOnce(V) -> V + Send + 'static>, bool),
+    UpdateMut(K, Box<dyn FnMut(&mut V) -> () + Send + 'static>, bool),
     Remove(K),
+    RemoveIf(Box<dyn Fn((&K, Option<&V>)) -> bool + Send + Sync + 'static>),
     // Internal use
     SetAndUnblock(K, V),
     Unblock(K),
@@ -57,9 +58,10 @@ impl<
                         CacheAction::GetIfPresent(key) => self.get_if_present(key),
                         CacheAction::Get(key) => self.get(key),
                         CacheAction::Set(key, value) => self.set(key, value, false),
-                        CacheAction::Update(key, update_fn) => self.update(key, update_fn),
-                        CacheAction::UpdateMut(key, update_mut_fn) => self.update_mut(key, update_mut_fn),
+                        CacheAction::Update(key, update_fn, load) => self.update(key, update_fn, load),
+                        CacheAction::UpdateMut(key, update_mut_fn, load) => self.update_mut(key, update_mut_fn, load),
                         CacheAction::Remove(key) => self.remove(key),
+                        CacheAction::RemoveIf(predicate) => self.remove_if(predicate),
                         CacheAction::SetAndUnblock(key, value) => self.set(key, value, true),
                         CacheAction::Unblock(key) => {
                             self.unblock(key);
@@ -95,7 +97,26 @@ impl<
         }
     }
 
-    fn update_mut(&mut self, key: K, mut update_mut_fn: Box<dyn FnMut(&mut V) -> () + Send + 'static>) -> CacheResult<V, E> {
+    fn remove_if(&mut self, predicate: Box<dyn Fn((&K, Option<&V>)) -> bool + Send + Sync + 'static>) -> CacheResult<V, E> {
+        self.data.remove_if(self.to_predicate(predicate));
+        CacheResult::None
+    }
+
+    fn to_predicate(&self, predicate: Box<dyn Fn((&K, Option<&V>)) -> bool + Send + Sync + 'static>)
+                    -> Box<dyn Fn((&K, &CacheEntry<V, E>)) -> bool + Send + Sync + 'static> {
+        Box::new(move |(key, value)| {
+            match value {
+                CacheEntry::Loaded(value) => {
+                    predicate((key, Some(value)))
+                }
+                CacheEntry::Loading(_) => {
+                    predicate((key, None))
+                }
+            }
+        })
+    }
+
+    fn update_mut(&mut self, key: K, mut update_mut_fn: Box<dyn FnMut(&mut V) -> () + Send + 'static>, load: bool) -> CacheResult<V, E> {
         match self.data.get_mut(&key) {
             Some(entry) => {
                 match entry {
@@ -110,7 +131,7 @@ impl<
                             rx.recv().await.ok(); // result confirmed
                             let (response_tx, response_rx) = tokio::sync::oneshot::channel();
                             cache_tx.send(CacheMessage {
-                                action: CacheAction::UpdateMut(key, update_mut_fn),
+                                action: CacheAction::UpdateMut(key, update_mut_fn, load),
                                 response: response_tx,
                             }).await.ok();
                             match response_rx.await.unwrap() {
@@ -122,31 +143,41 @@ impl<
                 }
             }
             None => {
-                let result = self.get(key.clone());
-                match result {
-                    CacheResult::Loading(waiter) => {
-                        let cache_tx = self.tx.clone();
-                        CacheResult::Loading(tokio::spawn(async move {
-                            waiter.await.ok(); // result confirmed
-                            let (response_tx, response_rx) = tokio::sync::oneshot::channel();
-                            cache_tx.send(CacheMessage {
-                                action: CacheAction::UpdateMut(key, update_mut_fn),
-                                response: response_tx,
-                            }).await.ok();
-                            match response_rx.await.unwrap() {
-                                CacheResult::Found(data) => Ok(data),
-                                _ => Err(CacheLoadingError::NoData())
-                            }
-                        }))
+                if load {
+                    let result = self.get(key.clone());
+                    match result {
+                        CacheResult::Loading(waiter) => {
+                            let cache_tx = self.tx.clone();
+                            CacheResult::Loading(tokio::spawn(async move {
+                                waiter.await.ok(); // result confirmed
+                                let (response_tx, response_rx) = tokio::sync::oneshot::channel();
+                                cache_tx.send(CacheMessage {
+                                    action: CacheAction::UpdateMut(key, update_mut_fn, load),
+                                    response: response_tx,
+                                }).await.ok();
+                                match response_rx.await.unwrap() {
+                                    CacheResult::Found(data) => Ok(data),
+                                    _ => Err(CacheLoadingError::NoData())
+                                }
+                            }))
+                        }
+                        _ => CacheResult::None,
                     }
-                    _ => CacheResult::None,
+                } else {
+                    CacheResult::None
                 }
             }
         }
     }
 
-    fn update(&mut self, key: K, update_fn: Box<dyn FnOnce(V) -> V + Send + 'static>) -> CacheResult<V, E> {
-        match self.get(key.clone()) {
+    fn update(&mut self, key: K, update_fn: Box<dyn FnOnce(V) -> V + Send + 'static>, load: bool) -> CacheResult<V, E> {
+        let data = if load {
+            self.get(key.clone())
+        } else {
+            self.get_if_present(key.clone())
+        };
+
+        match data {
             CacheResult::Found(data) => {
                 let updated_data = update_fn(data);
                 self.data.set(key, CacheEntry::Loaded(updated_data.clone()));
@@ -161,7 +192,7 @@ impl<
                     // todo: is there a possibility that this loops forever?
                     let (response_tx, rx) = tokio::sync::oneshot::channel();
                     tx.send(CacheMessage {
-                        action: CacheAction::Update(key, update_fn),
+                        action: CacheAction::Update(key, update_fn, load),
                         response: response_tx,
                     }).await.ok();
                     match rx.await {