qail_redis/
pool.rs

1//! Connection pooling for Redis.
2//!
3//! Manages a pool of RedisDriver connections for concurrent access.
4
5use std::collections::VecDeque;
6use std::ops::{Deref, DerefMut};
7use std::sync::Arc;
8
9use tokio::sync::{Mutex, Semaphore};
10
11use crate::driver::RedisDriver;
12use crate::error::{RedisError, RedisResult};
13
14/// Pool configuration.
15#[derive(Debug, Clone)]
16pub struct PoolConfig {
17    /// Maximum number of connections.
18    pub max_connections: usize,
19    /// Redis host.
20    pub host: String,
21    /// Redis port.
22    pub port: u16,
23}
24
25impl Default for PoolConfig {
26    fn default() -> Self {
27        Self {
28            max_connections: 10,
29            host: "127.0.0.1".to_string(),
30            port: 6379,
31        }
32    }
33}
34
35impl PoolConfig {
36    /// Create a new pool configuration.
37    pub fn new(host: impl Into<String>, port: u16) -> Self {
38        Self {
39            max_connections: 10,
40            host: host.into(),
41            port,
42        }
43    }
44
45    /// Set max connections.
46    pub fn max_connections(mut self, n: usize) -> Self {
47        self.max_connections = n;
48        self
49    }
50
51    /// Create config from centralized `QailConfig`.
52    ///
53    /// Reads `[redis]` section; returns `None` if section is absent.
54    pub fn from_qail_config(qail: &qail_core::config::QailConfig) -> Option<Self> {
55        let redis = qail.redis.as_ref()?;
56        Some(Self {
57            max_connections: redis.max_connections,
58            host: redis.host.clone(),
59            port: redis.port,
60        })
61    }
62}
63
64/// Redis connection pool.
65pub struct RedisPool {
66    config: PoolConfig,
67    connections: Arc<Mutex<VecDeque<RedisDriver>>>,
68    semaphore: Arc<Semaphore>,
69}
70
71impl RedisPool {
72    /// Create a new connection pool.
73    pub fn new(config: PoolConfig) -> Self {
74        let semaphore = Arc::new(Semaphore::new(config.max_connections));
75        Self {
76            config,
77            connections: Arc::new(Mutex::new(VecDeque::new())),
78            semaphore,
79        }
80    }
81
82    /// Get a connection from the pool.
83    pub async fn get(&self) -> RedisResult<PooledConnection> {
84        // Acquire permit
85        let permit = self
86            .semaphore
87            .clone()
88            .acquire_owned()
89            .await
90            .map_err(|_| RedisError::Pool("Failed to acquire pool permit".into()))?;
91
92        // Try to get existing connection
93        let driver = {
94            let mut conns = self.connections.lock().await;
95            conns.pop_front()
96        };
97
98        let driver = match driver {
99            Some(d) => d,
100            None => {
101                // Create new connection
102                RedisDriver::connect(&self.config.host, self.config.port).await?
103            }
104        };
105
106        Ok(PooledConnection {
107            driver: Some(driver),
108            pool: self.connections.clone(),
109            _permit: permit,
110        })
111    }
112}
113
114/// A pooled connection that returns to the pool on drop.
115pub struct PooledConnection {
116    driver: Option<RedisDriver>,
117    pool: Arc<Mutex<VecDeque<RedisDriver>>>,
118    _permit: tokio::sync::OwnedSemaphorePermit,
119}
120
121impl Deref for PooledConnection {
122    type Target = RedisDriver;
123
124    fn deref(&self) -> &Self::Target {
125        self.driver.as_ref().unwrap()
126    }
127}
128
129impl DerefMut for PooledConnection {
130    fn deref_mut(&mut self) -> &mut Self::Target {
131        self.driver.as_mut().unwrap()
132    }
133}
134
135impl Drop for PooledConnection {
136    fn drop(&mut self) {
137        if let Some(driver) = self.driver.take() {
138            let pool = self.pool.clone();
139            tokio::spawn(async move {
140                let mut conns = pool.lock().await;
141                conns.push_back(driver);
142            });
143        }
144    }
145}
146
147#[cfg(test)]
148mod tests {
149    use super::*;
150
151    #[test]
152    fn test_pool_config_default() {
153        let config = PoolConfig::default();
154        assert_eq!(config.max_connections, 10);
155        assert_eq!(config.host, "127.0.0.1");
156        assert_eq!(config.port, 6379);
157    }
158}
qail_redis/pool.rs

qail_redis/
pool.rs
