nexus_common/db/connectors/
redis.rs

1use crate::types::DynError;
2use deadpool_redis::{Config, Connection, Pool, Runtime};
3use once_cell::sync::OnceCell;
4use std::fmt;
5use tracing::{debug, info};
6
7pub struct RedisConnector {
8    pool: Pool,
9}
10
11impl RedisConnector {
12    /// Initialize and register the global Redis connector
13    pub async fn init(redis_uri: &str) -> Result<(), DynError> {
14        let redis_connector = RedisConnector::new_connection(redis_uri)
15            .await
16            .expect("Failed to connect to Redis");
17
18        redis_connector.ping(redis_uri).await?;
19
20        match REDIS_CONNECTOR.set(redis_connector) {
21            Err(e) => debug!("RedisConnector was already set: {:?}", e),
22            Ok(()) => info!("RedisConnector successfully set up on {}", redis_uri),
23        }
24        Ok(())
25    }
26
27    /// Creates a new RedisConnector instance by building a connection pool using the provided URI.
28    async fn new_connection(uri: &str) -> Result<Self, DynError> {
29        // Create the deadpool-redis configuration from the URI.
30        let cfg = Config::from_url(uri.to_string());
31
32        // Create the connection pool. We use the Tokio runtime.
33        let pool = cfg.create_pool(Some(Runtime::Tokio1))?;
34        Ok(Self { pool })
35    }
36
37    /// Returns a reference to the underlying connection pool.
38    fn pool(&self) -> &Pool {
39        &self.pool
40    }
41
42    /// Perform a health-check PING against the Redis server
43    async fn ping(&self, redis_uri: &str) -> Result<(), DynError> {
44        let redis_conn = self.pool.get().await;
45        match redis_conn {
46            Ok(_) => info!(
47                "Redis health check PING succeeded; server at {} is reachable",
48                redis_uri
49            ),
50            Err(_) => return Err(format!("Failed to PING to Redis at {redis_uri}").into()),
51        }
52        Ok(())
53    }
54}
55
56impl fmt::Debug for RedisConnector {
57    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
58        f.debug_struct("RedisConnector")
59            .field("pool", &"deadpool_redis::Pool")
60            .finish()
61    }
62}
63
64/// Global RedisConnector instance.
65/// Make sure to initialize this once when your application starts.
66pub static REDIS_CONNECTOR: OnceCell<RedisConnector> = OnceCell::new();
67
68/// Retrieves a Redis connection from the pool.
69pub async fn get_redis_conn() -> Result<Connection, DynError> {
70    let connector = REDIS_CONNECTOR
71        .get()
72        .ok_or("RedisConnector not initialized")?;
73    let conn = connector.pool().get().await?;
74    Ok(conn)
75}
nexus_common/db/connectors/redis.rs

nexus_common/db/connectors/
redis.rs
