feat(cli): cache

1 files changed, 66 insertions, 0 deletions

diff --git a/crates/sh-util/src/cache/sentinel.rs b/crates/sh-util/src/cache/sentinel.rs
new file mode 100644
index 0000000..e52b043
--- /dev/null
+++ b/crates/sh-util/src/cache/sentinel.rs
@@ -0,0 +1,66 @@
+use futures_util::lock::Mutex;
+use redis::{
+    ErrorKind, IntoConnectionInfo, RedisError,
+    sentinel::{SentinelClient, SentinelNodeConnectionInfo, SentinelServerType},
+};
+use serde::Deserialize;
+
+struct LockedSentinelClient(pub(crate) Mutex<SentinelClient>);
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SentinelConfig {
+    pub service_name: String,
+    pub redis_tls_mode_secure: bool,
+    pub redis_db: Option<i64>,
+    pub redis_username: String,
+    pub redis_password: String,
+    pub redis_use_resp3: bool,
+}
+
+/// ConnectionManager that implements `bb8::ManageConnection` and supports
+/// asynchronous Sentinel connections via `redis::sentinel::SentinelClient`
+pub struct RedisSentinelConnectionManager {
+    client: LockedSentinelClient,
+}
+
+impl RedisSentinelConnectionManager {
+    pub fn new<T: IntoConnectionInfo>(
+        info: Vec<T>,
+        service_name: String,
+        node_connection_info: Option<SentinelNodeConnectionInfo>,
+    ) -> Result<RedisSentinelConnectionManager, RedisError> {
+        Ok(RedisSentinelConnectionManager {
+            client: LockedSentinelClient(Mutex::new(SentinelClient::build(
+                info,
+                service_name,
+                node_connection_info,
+                SentinelServerType::Master,
+            )?)),
+        })
+    }
+}
+
+impl bb8::ManageConnection for RedisSentinelConnectionManager {
+    type Connection = redis::aio::MultiplexedConnection;
+    type Error = RedisError;
+
+    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+        self.client.0.lock().await.get_async_connection().await
+    }
+
+    async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+        let pong: String = redis::cmd("PING").query_async(conn).await?;
+        match pong.as_str() {
+            "PONG" => Ok(()),
+            _ => Err((
+                ErrorKind::Server(redis::ServerErrorKind::ResponseError),
+                "ping request",
+            )
+                .into()),
+        }
+    }
+
+    fn has_broken(&self, _: &mut Self::Connection) -> bool {
+        false
+    }
+}