switch back to connection manager from cluster

lib/chirp/worker/src/manager.rs

@@ -110,9 +110,9 @@ where
 				// Create a dedicated connection to redis-chirp for blocking Redis requests
 				// that won't block other requests in the pool.
 				let url = std::env::var("REDIS_URL_PERSISTENT").expect("REDIS_URL_PERSISTENT");
-				let redis_chirp_conn = redis::cluster::ClusterClient::new(vec![url.as_str()])
+				let redis_chirp_conn = redis::Client::open(url)
 					.map_err(ManagerError::BuildRedis)?
-					.get_async_connection()
+					.get_tokio_connection_manager()
 					.await
 					.map_err(ManagerError::BuildRedis)?;
 

lib/pools/Cargo.toml

@@ -39,9 +39,9 @@ version = "0.23"
 default-features = false
 features = [
 	"aio",
-	"cluster-async",
 	"tokio-comp",
 	"tokio-native-tls-comp",
+	"connection-manager",
 	"script",
 	"streams",
 ]

lib/pools/src/lib.rs

@@ -201,14 +201,9 @@ async fn redis_from_env() -> Result<HashMap<String, RedisPool>, Error> {
 				.name("redis_from_env")
 				.spawn(async move {
 					tracing::info!(%url, "redis connecting");
-					let conn = redis::cluster::ClusterClient::builder(vec![url.as_str()])
-						// Keep trying to reconnect indefinitely
-						.retries(u32::MAX)
-						.min_retry_wait(250)
-						.max_retry_wait(30_000)
-						.build()
+					let conn = redis::Client::open(url.as_str())
 						.map_err(Error::BuildRedis)?
-						.get_async_connection()
+						.get_tokio_connection_manager()
 						.await
 						.map_err(Error::BuildRedis)?;
 

lib/pools/src/pools.rs

@@ -5,34 +5,7 @@ use crate::Error;
 
 pub type NatsPool = async_nats::Client;
 pub type CrdbPool = sqlx::PgPool;
-pub type RedisPool = redis::cluster_async::ClusterConnection;
-
-// #[derive(Clone)]
-// struct RedisPool {
-// 	connection: redis::cluster_async::ClusterConnection,
-// }
-
-// impl RedisPool {
-// 	pub async fn pipe<T, F>(&mut self, f: F) -> RedisResult<T> where F: FnOnce(redis::cluster::ClusterPipeline) -> redis::cluster::ClusterPipeline {
-// 		let mut pipe = redis::cluster::cluster_pipe();
-
-// 		f(pipe);
-
-// 		thread::spawn(|| async {
-// 			pipe.query(&mut self)
-// 		}).await
-// 	}
-
-// 	pub async fn cmd<T, F>(&mut self, cmd: &str, f: F) -> RedisResult<T> where F: FnOnce(redis::cluster::ClusterPipeline) -> redis::cluster::ClusterPipeline {
-// 		let mut cmd = redis::cluster::cmd(cmd);
-
-// 		f(cmd);
-
-// 		thread::spawn(|| async {
-// 			pipe.query(&mut self)
-// 		}).await
-// 	}
-// }
+pub type RedisPool = redis::aio::ConnectionManager;
 
 pub type Pools = Arc<PoolsInner>;