Repository: bikeshedder/deadpool (rust)

Issue:
Max & idle connection lifetime
### Discussed in https://github.com/bikeshedder/deadpool/discussions/177

<div type='discussions-op-text'>

<sup>Originally posted by **Sytten** February 11, 2022</sup>
Hi!
I was wondering if you plan to implement maximum and idle lifetimes in the pool? Right now it seems like the checking for bad connections is done lazily at connection fetch. I know for example that sqlx runs a low priority task to check the health of connections in the background and respect the max/idle lifetimes.
Thanks!</div>

Pull request description:
Add Pool::retain method
This closes #178

Patch:
diff --git a/Cargo.toml b/Cargo.toml
index cbe31ff..8f8153b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -23,6 +23,7 @@ rt_async-std_1 = ["deadpool-runtime/async-std_1"]
 
 [dependencies]
 num_cpus = "1.11.1"
+retain_mut ="0.1.6"
 # `managed` feature
 async-trait = { version = "0.1.17", optional = true }
 # `serde` feature
diff --git a/src/managed/mod.rs b/src/managed/mod.rs
index 210c196..263ed43 100644
--- a/src/managed/mod.rs
+++ b/src/managed/mod.rs
@@ -79,6 +79,7 @@ use std::{
 
 use async_trait::async_trait;
 use deadpool_runtime::Runtime;
+use retain_mut::RetainMut;
 use tokio::sync::{Semaphore, TryAcquireError};
 
 pub use crate::Status;
@@ -492,6 +493,43 @@ impl<M: Manager, W: From<Object<M>>> Pool<M, W> {
         }
     }
 
+    /// Retains only the objects specified by the given function.
+    ///
+    /// This function is typically used to remove objects from
+    /// the pool based on their current state or metrics.
+    ///
+    /// **Caution:** This function blocks the entire pool while
+    /// it is running. Therefore the given function should not
+    /// block.
+    ///
+    /// The following example starts a background task that
+    /// runs every 30 seconds and removes objects from the pool
+    /// that haven't been used for more than one minute.
+    ///
+    /// ```rust,ignore
+    /// let interval = Duration::from_secs(30);
+    /// let max_age = Duration::from_secs(60);
+    /// tokio::spawn(async move {
+    ///     loop {
+    ///         tokio::time::sleep(interval).await;
+    ///         pool.retain(|_, metrics| metrics.last_used() < max_age);
+    ///     }
+    /// });
+    /// ```
+    pub fn retain(&self, f: impl Fn(&M::Type, Metrics) -> bool) {
+        let mut guard = self.inner.slots.lock().unwrap();
+        let len_before = guard.vec.len();
+        guard.vec.retain_mut(|obj| {
+            if f(&obj.obj, obj.metrics) {
+                true
+            } else {
+                self.manager().detach(&mut obj.obj);
+                false
+            }
+        });
+        guard.size -= len_before - guard.vec.len();
+    }
+
     /// Closes this [`Pool`].
     ///
     /// All current and future tasks waiting for [`Object`]s will return


Test patch:
diff --git a/tests/managed.rs b/tests/managed.rs
index a8c2328..af5b101 100644
--- a/tests/managed.rs
+++ b/tests/managed.rs
@@ -253,3 +253,22 @@ async fn resize_pool_grow_concurrent() {
     assert_eq!(pool.status().size, 1);
     assert_eq!(pool.status().available, 1);
 }
+
+#[tokio::test]
+async fn retain() {
+    let mgr = Manager {};
+    let pool = Pool::builder(mgr).max_size(4).build().unwrap();
+    {
+        let _a = pool.get().await;
+        let _b = pool.get().await;
+        tokio::time::sleep(Duration::from_millis(5)).await;
+        let _c = pool.get().await;
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+    assert_eq!(pool.status().size, 3);
+    pool.retain(|_, metrics| metrics.age() <= Duration::from_millis(10));
+    assert_eq!(pool.status().size, 1);
+    tokio::time::sleep(Duration::from_millis(5)).await;
+    pool.retain(|_, metrics| metrics.age() <= Duration::from_millis(10));
+    assert_eq!(pool.status().size, 0);
+}