Add redis sentinel support

omniqueue/Cargo.toml

@@ -39,6 +39,7 @@ serde = { version = "1.0.196", features = ["derive"] }
 tokio = { version = "1", features = ["macros"] }
 tokio-executor-trait = "2.1"
 tokio-reactor-trait = "1.1"
+rstest = "0.23.0"
 
 [features]
 default = ["in_memory", "gcp_pubsub", "rabbitmq", "redis", "redis_cluster", "sqs"]
@@ -49,6 +50,7 @@ rabbitmq = ["dep:futures-util", "dep:lapin"]
 rabbitmq-with-message-ids = ["rabbitmq", "dep:svix-ksuid"]
 redis = ["dep:bb8", "dep:bb8-redis", "dep:redis", "dep:svix-ksuid"]
 redis_cluster = ["dep:async-trait", "redis", "redis/cluster-async"]
+redis_sentinel = ["dep:async-trait", "redis", "redis/sentinel"]
 sqs = ["dep:aws-config", "dep:aws-sdk-sqs"]
 azure_queue_storage = ["dep:azure_storage", "dep:azure_storage_queues"]
 beta = []

omniqueue/src/backends/redis/mod.rs

@@ -41,7 +41,10 @@ use std::{
 
 use bb8::ManageConnection;
 pub use bb8_redis::RedisConnectionManager;
-use redis::{AsyncCommands, ExistenceCheck, SetExpiry, SetOptions};
+use redis::{
+    sentinel::SentinelNodeConnectionInfo, AsyncCommands, ExistenceCheck, ProtocolVersion,
+    RedisConnectionInfo, SetExpiry, SetOptions, TlsMode,
+};
 use serde::Serialize;
 use svix_ksuid::KsuidLike;
 use thiserror::Error;
@@ -58,30 +61,65 @@ use crate::{
 #[cfg(feature = "redis_cluster")]
 mod cluster;
 mod fallback;
+#[cfg(feature = "redis_sentinel")]
+mod sentinel;
 mod streams;
 
 #[cfg(feature = "redis_cluster")]
 pub use cluster::RedisClusterConnectionManager;
+#[cfg(feature = "redis_sentinel")]
+pub use sentinel::RedisSentinelConnectionManager;
 
 pub trait RedisConnection:
     ManageConnection<
     Connection: redis::aio::ConnectionLike + Send + Sync,
     Error: std::error::Error + Send + Sync + 'static,
 >
 {
-    fn from_dsn(dsn: &str) -> Result<Self>;
+    fn from_config(config: &RedisConfig) -> Result<Self>;
 }
 
 impl RedisConnection for RedisConnectionManager {
-    fn from_dsn(dsn: &str) -> Result<Self> {
-        Self::new(dsn).map_err(QueueError::generic)
+    fn from_config(config: &RedisConfig) -> Result<Self> {
+        Self::new(config.dsn.as_str()).map_err(QueueError::generic)
     }
 }
 
 #[cfg(feature = "redis_cluster")]
 impl RedisConnection for RedisClusterConnectionManager {
-    fn from_dsn(dsn: &str) -> Result<Self> {
-        Self::new(dsn).map_err(QueueError::generic)
+    fn from_config(config: &RedisConfig) -> Result<Self> {
+        Self::new(config.dsn.as_str()).map_err(QueueError::generic)
+    }
+}
+
+#[cfg(feature = "redis_sentinel")]
+impl RedisConnection for RedisSentinelConnectionManager {
+    fn from_config(config: &RedisConfig) -> Result<Self> {
+        let cfg = config
+            .sentinel_config
+            .clone()
+            .ok_or(QueueError::Unsupported("Missing sentinel configuration"))?;
+
+        let tls_mode = cfg.redis_tls_mode_secure.then_some(TlsMode::Secure);
+        let protocol = if cfg.redis_use_resp3 {
+            ProtocolVersion::RESP3
+        } else {
+            ProtocolVersion::default()
+        };
+        RedisSentinelConnectionManager::new(
+            vec![config.dsn.as_str()],
+            cfg.service_name.clone(),
+            Some(SentinelNodeConnectionInfo {
+                tls_mode,
+                redis_connection_info: Some(RedisConnectionInfo {
+                    db: cfg.redis_db.unwrap_or(0),
+                    username: cfg.redis_username.clone(),
+                    password: cfg.redis_password.clone(),
+                    protocol,
+                }),
+            }),
+        )
+        .map_err(QueueError::generic)
     }
 }
 
@@ -233,6 +271,17 @@ pub struct RedisConfig {
     pub payload_key: String,
     pub ack_deadline_ms: i64,
     pub dlq_config: Option<DeadLetterQueueConfig>,
+    pub sentinel_config: Option<SentinelConfig>,
+}
+
+#[derive(Clone)]
+pub struct SentinelConfig {
+    pub service_name: String,
+    pub redis_tls_mode_secure: bool,
+    pub redis_db: Option<i64>,
+    pub redis_username: Option<String>,
+    pub redis_password: Option<String>,
+    pub redis_use_resp3: bool,
 }
 
 #[derive(Clone)]
@@ -269,6 +318,12 @@ impl RedisBackend {
     pub fn cluster_builder(config: RedisConfig) -> RedisClusterBackendBuilder {
         RedisBackendBuilder::new(config)
     }
+
+    #[cfg(feature = "redis_sentinel")]
+    /// Creates a new redis sentinel queue builder with the given configuration.
+    pub fn sentinel_builder(config: RedisConfig) -> RedisSentinelBackendBuilder {
+        RedisBackendBuilder::new(config)
+    }
 }
 
 #[allow(deprecated)]
@@ -305,8 +360,11 @@ pub struct RedisBackendBuilder<R = RedisConnectionManager, S = Static> {
 #[cfg(feature = "redis_cluster")]
 pub type RedisClusterBackendBuilder = RedisBackendBuilder<RedisClusterConnectionManager>;
 
+#[cfg(feature = "redis_sentinel")]
+pub type RedisSentinelBackendBuilder = RedisBackendBuilder<RedisSentinelConnectionManager>;
+
 impl<R: RedisConnection> RedisBackendBuilder<R> {
-    fn new(config: RedisConfig) -> Self {
+    pub fn new(config: RedisConfig) -> Self {
         Self {
             config,
             use_redis_streams: true,
@@ -372,7 +430,7 @@ impl<R: RedisConnection> RedisBackendBuilder<R> {
     }
 
     pub async fn build_pair(self) -> Result<(RedisProducer<R>, RedisConsumer<R>)> {
-        let redis = R::from_dsn(&self.config.dsn)?;
+        let redis = R::from_config(&self.config)?;
         let redis = bb8::Pool::builder()
             .max_size(self.config.max_connections.into())
             .build(redis)
@@ -407,7 +465,7 @@ impl<R: RedisConnection> RedisBackendBuilder<R> {
     }
 
     pub async fn build_producer(self) -> Result<RedisProducer<R>> {
-        let redis = R::from_dsn(&self.config.dsn)?;
+        let redis = R::from_config(&self.config)?;
         let redis = bb8::Pool::builder()
             .max_size(self.config.max_connections.into())
             .build(redis)
@@ -427,7 +485,7 @@ impl<R: RedisConnection> RedisBackendBuilder<R> {
     }
 
     pub async fn build_consumer(self) -> Result<RedisConsumer<R>> {
-        let redis = R::from_dsn(&self.config.dsn)?;
+        let redis = R::from_config(&self.config)?;
         let redis = bb8::Pool::builder()
             .max_size(self.config.max_connections.into())
             .build(redis)

omniqueue/src/backends/redis/sentinel.rs

@@ -0,0 +1,56 @@
+use async_trait::async_trait;
+use redis::{
+    sentinel::{SentinelClient, SentinelNodeConnectionInfo, SentinelServerType},
+    ErrorKind, IntoConnectionInfo, RedisError,
+};
+use tokio::sync::Mutex;
+
+// The mutex here is needed b/c there's currently
+// no way to get connections in the redis sentinel client
+// without a mutable reference to the underlying client.
+struct LockedSentinelClient(pub(crate) Mutex<SentinelClient>);
+
+/// 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,
+            )?)),
+        })
+    }
+}
+
+#[async_trait]
+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::ResponseError, "ping request").into()),
+        }
+    }
+
+    fn has_broken(&self, _: &mut Self::Connection) -> bool {
+        false
+    }
+}

omniqueue/tests/it/main.rs

@@ -4,7 +4,7 @@ mod azure_queue_storage;
 mod gcp_pubsub;
 #[cfg(feature = "rabbitmq")]
 mod rabbitmq;
-#[cfg(feature = "redis")]
+#[cfg(any(feature = "redis", feature = "redis_sentinel"))]
 mod redis;
 #[cfg(feature = "redis_cluster")]
 mod redis_cluster;