Merge pull request jaemk#155 from a-nickol/redis-connection-manager

Additional `redis_connection_manager` for usage of `ConnectionManager` from `redis`

Cargo.toml

@@ -21,6 +21,7 @@ proc_macro = ["tokio", "cached_proc_macro", "cached_proc_macro_types"]
 async = ["futures", "tokio", "async-trait"]
 async_tokio_rt_multi_thread = ["async", "tokio/rt-multi-thread"]
 redis_store = ["redis", "r2d2", "serde", "serde_json"]
+redis_connection_manager = ["redis_store", "redis/connection-manager"]
 redis_async_std = ["redis_store", "async", "redis/aio", "redis/async-std-comp", "redis/tls", "redis/async-std-tls-comp"]
 redis_tokio = ["redis_store", "async", "redis/aio", "redis/tokio-comp", "redis/tls", "redis/tokio-native-tls-comp"]
 redis_ahash = ["redis_store", "redis/ahash"]

README.md

@@ -29,6 +29,8 @@ of un-cached arguments, specify `#[cached(sync_writes = true)]` / `#[once(sync_w
 - `redis_store`: Include Redis cache store
 - `redis_async_std`: Include async Redis support using `async-std` and `async-std` tls support, implies `redis_store` and `async`
 - `redis_tokio`: Include async Redis support using `tokio` and `tokio` tls support, implies `redis_store` and `async`
+- `redis_connection_manager`: Enable the optional `connection-manager` feature of `redis`. Any async redis caches created
+                              will use a connection manager instead of a `MultiplexedConnection`
 - `redis_ahash`: Enable the optional `ahash` feature of `redis`
 - `wasm`: Enable WASM support. Note that this feature is incompatible with `tokio`'s multi-thread
    runtime (`async_tokio_rt_multi_thread`) and all Redis features (`redis_store`, `redis_async_std`, `redis_tokio`, `redis_ahash`)

src/lib.rs

@@ -28,6 +28,8 @@ of un-cached arguments, specify `#[cached(sync_writes = true)]` / `#[once(sync_w
 - `redis_store`: Include Redis cache store
 - `redis_async_std`: Include async Redis support using `async-std` and `async-std` tls support, implies `redis_store` and `async`
 - `redis_tokio`: Include async Redis support using `tokio` and `tokio` tls support, implies `redis_store` and `async`
+- `redis_connection_manager`: Enable the optional `connection-manager` feature of `redis`. Any async redis caches created
+                              will use a connection manager instead of a `MultiplexedConnection`
 - `redis_ahash`: Enable the optional `ahash` feature of `redis`
 - `wasm`: Enable WASM support. Note that this feature is incompatible with `tokio`'s multi-thread
    runtime (`async_tokio_rt_multi_thread`) and all Redis features (`redis_store`, `redis_async_std`, `redis_tokio`, `redis_ahash`)

src/stores/redis.rs

@@ -453,6 +453,9 @@ mod async_redis {
             }
         }
 
+        /// Create a multiplexed redis connection. This is a single connection that can
+        /// be used asynchronously by multiple futures.
+        #[cfg(not(feature = "redis_connection_manager"))]
         async fn create_multiplexed_connection(
             &self,
         ) -> Result<redis::aio::MultiplexedConnection, RedisCacheBuildError> {
@@ -462,6 +465,19 @@ mod async_redis {
             Ok(conn)
         }
 
+        /// Create a multiplexed connection wrapped in a manager. The manager provides access
+        /// to a multiplexed connection and will automatically reconnect to the server when
+        /// necessary.
+        #[cfg(feature = "redis_connection_manager")]
+        async fn create_connection_manager(
+            &self,
+        ) -> Result<redis::aio::ConnectionManager, RedisCacheBuildError> {
+            let s = self.connection_string()?;
+            let client = redis::Client::open(s)?;
+            let conn = redis::aio::ConnectionManager::new(client).await?;
+            Ok(conn)
+        }
+
         /// The last step in building a `RedisCache` is to call `build()`
         ///
         /// # Errors
@@ -472,7 +488,10 @@ mod async_redis {
                 seconds: self.seconds,
                 refresh: self.refresh,
                 connection_string: self.connection_string()?,
-                multiplexed_connection: self.create_multiplexed_connection().await?,
+                #[cfg(not(feature = "redis_connection_manager"))]
+                connection: self.create_multiplexed_connection().await?,
+                #[cfg(feature = "redis_connection_manager")]
+                connection: self.create_connection_manager().await?,
                 namespace: self.namespace,
                 prefix: self.prefix,
                 _phantom: PhantomData::default(),
@@ -483,14 +502,18 @@ mod async_redis {
     /// Cache store backed by redis
     ///
     /// Values have a ttl applied and enforced by redis.
-    /// Uses a `redis::aio::MultiplexedConnection` under the hood.
+    /// Uses a `redis::aio::MultiplexedConnection` or `redis::aio::ConnectionManager`
+    /// under the hood depending if feature `redis_connection_manager` is used or not.
     pub struct AsyncRedisCache<K, V> {
         pub(super) seconds: u64,
         pub(super) refresh: bool,
         pub(super) namespace: String,
         pub(super) prefix: String,
         connection_string: String,
-        multiplexed_connection: redis::aio::MultiplexedConnection,
+        #[cfg(not(feature = "redis_connection_manager"))]
+        connection: redis::aio::MultiplexedConnection,
+        #[cfg(feature = "redis_connection_manager")]
+        connection: redis::aio::ConnectionManager,
         _phantom: PhantomData<(K, V)>,
     }
 
@@ -526,7 +549,7 @@ mod async_redis {
 
         /// Get a cached value
         async fn cache_get(&self, key: &K) -> Result<Option<V>, Self::Error> {
-            let mut conn = self.multiplexed_connection.clone();
+            let mut conn = self.connection.clone();
             let mut pipe = redis::pipe();
             let key = self.generate_key(key);
 
@@ -551,7 +574,7 @@ mod async_redis {
 
         /// Set a cached value
         async fn cache_set(&self, key: K, val: V) -> Result<Option<V>, Self::Error> {
-            let mut conn = self.multiplexed_connection.clone();
+            let mut conn = self.connection.clone();
             let mut pipe = redis::pipe();
             let key = self.generate_key(&key);
 
@@ -582,7 +605,7 @@ mod async_redis {
 
         /// Remove a cached value
         async fn cache_remove(&self, key: &K) -> Result<Option<V>, Self::Error> {
-            let mut conn = self.multiplexed_connection.clone();
+            let mut conn = self.connection.clone();
             let mut pipe = redis::pipe();
             let key = self.generate_key(key);