diff --git a/redis/src/cluster/config.rs b/redis/src/cluster/config.rs index 076925f0..f89633fa 100644 --- a/redis/src/cluster/config.rs +++ b/redis/src/cluster/config.rs @@ -47,6 +47,11 @@ pub struct Config { /// [`redis::ConnectionInfo`] structures. pub connections: Option>, + /// [`Manager`] configuration. + /// + /// [`Manager`]: super::Manager + pub manager: Option, + /// Pool configuration. pub pool: Option, @@ -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, } diff --git a/redis/src/config.rs b/redis/src/config.rs index 821fc681..bf4a5526 100644 --- a/redis/src/config.rs +++ b/redis/src/config.rs @@ -45,6 +45,11 @@ pub struct Config { /// [`redis::ConnectionInfo`] structure. pub connection: Option, + /// [`Manager`] configuration. + /// + /// [`Manager`]: super::Manager + pub manager: Option, + /// Pool configuration. pub pool: Option, } @@ -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, +} diff --git a/redis/src/lib.rs b/redis/src/lib.rs index b995be34..96cd4bf1 100644 --- a/redis/src/lib.rs +++ b/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(params: T) -> RedisResult { - 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( params: T, - connection_config: AsyncConnectionConfig, + config: ManagerConfig, ) -> RedisResult { 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 { 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")) + } + } } } } diff --git a/redis/src/sentinel/config.rs b/redis/src/sentinel/config.rs index 13a4cb4e..922d966f 100644 --- a/redis/src/sentinel/config.rs +++ b/redis/src/sentinel/config.rs @@ -57,6 +57,12 @@ pub struct Config { pub connections: Option>, /// [`redis::sentinel::SentinelNodeConnectionInfo`] structures. pub node_connection_info: Option, + + /// [`Manager`] configuration. + /// + /// [`Manager`]: super::Manager + pub manager: Option, + /// Pool configuration. pub pool: Option, } @@ -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, } } }