Add RecyclingMethod to deadpool-redis

redis/src/cluster/config.rs

@@ -47,6 +47,11 @@ pub struct Config {
     /// [`redis::ConnectionInfo`] structures.
     pub connections: Option<Vec<ConnectionInfo>>,
 
+    /// [`Manager`] configuration.
+    ///
+    /// [`Manager`]: super::Manager
+    pub manager: Option<ManagerConfig>,
+
     /// Pool configuration.
     pub pool: Option<PoolConfig>,
 
@@ -114,6 +119,7 @@ impl Config {
         Config {
             urls: Some(urls.into()),
             connections: None,
+            manager: None,
             pool: None,
             read_from_replicas: false,
         }
@@ -125,6 +131,7 @@ impl Default for Config {
         Self {
             urls: None,
             connections: Some(vec![ConnectionInfo::default()]),
+            manager: None,
             pool: None,
             read_from_replicas: false,
         }

redis/src/config.rs

@@ -45,6 +45,11 @@ pub struct Config {
     /// [`redis::ConnectionInfo`] structure.
     pub connection: Option<ConnectionInfo>,
 
+    /// [`Manager`] configuration.
+    ///
+    /// [`Manager`]: super::Manager
+    pub manager: Option<ManagerConfig>,
+
     /// Pool configuration.
     pub pool: Option<PoolConfig>,
 }
@@ -93,6 +98,7 @@ impl Config {
         Config {
             url: Some(url.into()),
             connection: None,
+            manager: None,
             pool: None,
         }
     }
@@ -104,6 +110,7 @@ impl Config {
         Config {
             url: None,
             connection: Some(connection_info.into()),
+            manager: None,
             pool: None,
         }
     }
@@ -114,6 +121,7 @@ impl Default for Config {
         Self {
             url: None,
             connection: Some(ConnectionInfo::default()),
+            manager: None,
             pool: None,
         }
     }
@@ -328,3 +336,32 @@ impl fmt::Display for ConfigError {
 }
 
 impl std::error::Error for ConfigError {}
+
+/// Configuration object for a [`Manager`].
+///
+/// This currently only makes it possible to specify which [`RecyclingMethod`]
+/// should be used when retrieving existing objects from the [`Pool`].
+///
+/// [`Manager`]: super::Manager
+#[derive(Clone, Copy, Debug, Default)]
+#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
+pub struct ManagerConfig {
+    /// Method of how a connection is recycled. See [`RecyclingMethod`].
+    pub recycling_method: RecyclingMethod,
+}
+
+/// Possible methods of how a connection is recycled.
+///
+/// The default is [`Fast`] which does not check the connection health or
+/// perform any clean-up queries.
+///
+/// [`Fast`]: RecyclingMethod::Fast
+/// [`Verified`]: RecyclingMethod::Verified
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Default)]
+#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
+pub enum RecyclingMethod {
+    /// Run `UNWATCH` to discard transaction state, then send a ping to verify
+    /// the connection is still alive.
+    #[default]
+    Clean,
+}

redis/src/lib.rs

@@ -42,7 +42,8 @@ use redis::{
 pub use redis;
 
 pub use self::config::{
-    Config, ConfigError, ConnectionAddr, ConnectionInfo, ProtocolVersion, RedisConnectionInfo,
+    Config, ConfigError, ConnectionAddr, ConnectionInfo, ManagerConfig, ProtocolVersion,
+    RecyclingMethod, RedisConnectionInfo,
 };
 
 pub use deadpool::managed::reexports::*;
@@ -127,20 +128,11 @@ impl ConnectionLike for Connection {
 /// [`Manager`] for creating and recycling [`redis`] connections.
 ///
 /// [`Manager`]: managed::Manager
+#[derive(Debug)]
 pub struct Manager {
+    config: ManagerConfig,
     client: Client,
     ping_number: AtomicUsize,
-    connection_config: AsyncConnectionConfig,
-}
-
-// `redis::AsyncConnectionConfig: !Debug`
-impl std::fmt::Debug for Manager {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Manager")
-            .field("client", &self.client)
-            .field("ping_number", &self.ping_number)
-            .finish()
-    }
 }
 
 impl Manager {
@@ -150,22 +142,22 @@ impl Manager {
     ///
     /// If establishing a new [`Client`] fails.
     pub fn new<T: IntoConnectionInfo>(params: T) -> RedisResult<Self> {
-        Self::from_config(params, AsyncConnectionConfig::default())
+        Self::from_config(params, ManagerConfig::default())
     }
 
-    /// Creates a new [`Manager`] from the given `params` and [`AsyncConnectionConfig`].
+    /// Creates a new [`Manager`] from the given `params` and [`ManagerConfig`].
     ///
     /// # Errors
     ///
     /// If establishing a new [`Client`] fails.
     pub fn from_config<T: IntoConnectionInfo>(
         params: T,
-        connection_config: AsyncConnectionConfig,
+        config: ManagerConfig,
     ) -> RedisResult<Self> {
         Ok(Self {
+            config,
             client: Client::open(params)?,
             ping_number: AtomicUsize::new(0),
-            connection_config,
         })
     }
 }
@@ -177,25 +169,29 @@ impl managed::Manager for Manager {
     async fn create(&self) -> Result<MultiplexedConnection, RedisError> {
         let conn = self
             .client
-            .get_multiplexed_async_connection_with_config(&self.connection_config)
+            .get_multiplexed_async_connection_with_config(&AsyncConnectionConfig::default())
             .await?;
         Ok(conn)
     }
 
     async fn recycle(&self, conn: &mut MultiplexedConnection, _: &Metrics) -> RecycleResult {
-        let ping_number = self.ping_number.fetch_add(1, Ordering::Relaxed).to_string();
-        // Using pipeline to avoid roundtrip for UNWATCH
-        let (n,) = redis::Pipeline::with_capacity(2)
-            .cmd("UNWATCH")
-            .ignore()
-            .cmd("PING")
-            .arg(&ping_number)
-            .query_async::<(String,)>(conn)
-            .await?;
-        if n == ping_number {
-            Ok(())
-        } else {
-            Err(managed::RecycleError::message("Invalid PING response"))
+        match self.config.recycling_method {
+            RecyclingMethod::Clean => {
+                let ping_number = self.ping_number.fetch_add(1, Ordering::Relaxed).to_string();
+                // Using pipeline to avoid roundtrip for UNWATCH
+                let (n,) = redis::Pipeline::with_capacity(2)
+                    .cmd("UNWATCH")
+                    .ignore()
+                    .cmd("PING")
+                    .arg(&ping_number)
+                    .query_async::<(String,)>(conn)
+                    .await?;
+                if n == ping_number {
+                    Ok(())
+                } else {
+                    Err(managed::RecycleError::message("Invalid PING response"))
+                }
+            }
         }
     }
 }

redis/src/sentinel/config.rs

@@ -57,6 +57,12 @@ pub struct Config {
     pub connections: Option<Vec<ConnectionInfo>>,
     /// [`redis::sentinel::SentinelNodeConnectionInfo`] structures.
     pub node_connection_info: Option<SentinelNodeConnectionInfo>,
+
+    /// [`Manager`] configuration.
+    ///
+    /// [`Manager`]: super::Manager
+    pub manager: Option<ManagerConfig>,
+
     /// Pool configuration.
     pub pool: Option<PoolConfig>,
 }
@@ -123,11 +129,12 @@ impl Config {
     ) -> Config {
         Config {
             urls: Some(urls.into()),
-            connections: None,
-            master_name,
             server_type,
-            pool: None,
+            master_name,
+            connections: None,
             node_connection_info: None,
+            manager: None,
+            pool: None,
         }
     }
 
@@ -150,11 +157,12 @@ impl Default for Config {
 
         Self {
             urls: None,
-            connections: Some(vec![default_connection_info.clone()]),
             server_type: SentinelServerType::Master,
             master_name: default_master_name(),
-            pool: None,
+            connections: Some(vec![default_connection_info.clone()]),
             node_connection_info: None,
+            manager: None,
+            pool: None,
         }
     }
 }