redis_oxide/
connection.rs

1//! Connection management and topology detection
2//!
3//! This module handles low-level TCP connections to Redis servers,
4//! automatic topology detection, and connection lifecycle management.
5
6use crate::core::{
7    config::{ConnectionConfig, TopologyMode},
8    error::{RedisError, RedisResult},
9    value::RespValue,
10};
11use crate::protocol::{ProtocolConnection, RespDecoder, RespEncoder};
12use bytes::{Buf, BytesMut};
13use std::io::Cursor;
14use tokio::io::{AsyncReadExt, AsyncWriteExt};
15use tokio::net::TcpStream;
16use tokio::time::timeout;
17use tracing::{debug, info, warn};
18
19/// Type of Redis topology
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum TopologyType {
22    /// Standalone Redis server
23    Standalone,
24    /// Redis Cluster
25    Cluster,
26}
27
28/// A connection to a Redis server
29pub struct RedisConnection {
30    stream: TcpStream,
31    read_buffer: BytesMut,
32    config: ConnectionConfig,
33}
34
35impl RedisConnection {
36    fn is_cluster_info(info: &str) -> bool {
37        info.contains("cluster_enabled:1") || info.contains("cluster_state:ok")
38    }
39
40    /// Connect to a Redis server
41    pub async fn connect(host: &str, port: u16, config: ConnectionConfig) -> RedisResult<Self> {
42        let addr = format!("{}:{}", host, port);
43        debug!("Connecting to Redis at {}", addr);
44
45        let stream = timeout(config.connect_timeout, TcpStream::connect(&addr))
46            .await
47            .map_err(|_| RedisError::Timeout)?
48            .map_err(|e| RedisError::Connection(format!("Failed to connect to {}: {}", addr, e)))?;
49
50        // Set TCP keepalive if configured
51        if let Some(keepalive_duration) = config.tcp_keepalive {
52            let socket = socket2::Socket::from(stream.into_std()?);
53            let keepalive = socket2::TcpKeepalive::new().with_time(keepalive_duration);
54            socket.set_tcp_keepalive(&keepalive).map_err(|e| {
55                RedisError::Connection(format!("Failed to set TCP keepalive: {}", e))
56            })?;
57            let stream = TcpStream::from_std(socket.into())?;
58
59            let mut conn = Self {
60                stream,
61                read_buffer: BytesMut::with_capacity(8192),
62                config: config.clone(),
63            };
64
65            // Authenticate if password is provided
66            if let Some(ref password) = config.password {
67                conn.authenticate(password).await?;
68            }
69
70            Ok(conn)
71        } else {
72            let mut conn = Self {
73                stream,
74                read_buffer: BytesMut::with_capacity(8192),
75                config: config.clone(),
76            };
77
78            // Authenticate if password is provided
79            if let Some(ref password) = config.password {
80                conn.authenticate(password).await?;
81            }
82
83            Ok(conn)
84        }
85    }
86
87    /// Authenticate with the Redis server
88    async fn authenticate(&mut self, password: &str) -> RedisResult<()> {
89        debug!("Authenticating with Redis server");
90        let response = self
91            .execute_command("AUTH", &[RespValue::from(password)])
92            .await?;
93
94        match response {
95            RespValue::SimpleString(ref s) if s == "OK" => Ok(()),
96            RespValue::Error(e) => Err(RedisError::Auth(e)),
97            _ => Err(RedisError::Auth(
98                "Unexpected authentication response".to_string(),
99            )),
100        }
101    }
102
103    /// Send a command to the server
104    pub async fn send_command(&mut self, command: &RespValue) -> RedisResult<()> {
105        let mut buffer = BytesMut::new();
106        RespEncoder::encode(command, &mut buffer)?;
107        self.stream.write_all(&buffer).await?;
108        Ok(())
109    }
110
111    /// Execute a command and return the response
112    pub async fn execute_command(
113        &mut self,
114        command: &str,
115        args: &[RespValue],
116    ) -> RedisResult<RespValue> {
117        // Encode command
118        let encoded = RespEncoder::encode_command(command, args)?;
119
120        // Send command with timeout
121        timeout(
122            self.config.operation_timeout,
123            self.stream.write_all(&encoded),
124        )
125        .await
126        .map_err(|_| RedisError::Timeout)?
127        .map_err(RedisError::Io)?;
128
129        // Read response with timeout
130        let response = timeout(self.config.operation_timeout, self.read_response())
131            .await
132            .map_err(|_| RedisError::Timeout)??;
133
134        // Check if response is an error and parse redirects
135        if let RespValue::Error(ref msg) = response {
136            if let Some(redirect_error) = RedisError::parse_redirect(msg) {
137                return Err(redirect_error);
138            }
139            return Err(RedisError::Server(msg.clone()));
140        }
141
142        Ok(response)
143    }
144
145    /// Read a complete RESP response from the connection
146    pub async fn read_response(&mut self) -> RedisResult<RespValue> {
147        loop {
148            // Try to decode from existing buffer
149            let mut cursor = Cursor::new(&self.read_buffer[..]);
150            if let Some(value) = RespDecoder::decode(&mut cursor)? {
151                let pos = cursor.position() as usize;
152                self.read_buffer.advance(pos);
153                return Ok(value);
154            }
155
156            // Need more data - read from socket
157            let n = self.stream.read_buf(&mut self.read_buffer).await?;
158            if n == 0 {
159                return Err(RedisError::Connection(
160                    "Connection closed by server".to_string(),
161                ));
162            }
163        }
164    }
165
166    /// Detect the topology type of the Redis server
167    pub async fn detect_topology(&mut self) -> RedisResult<TopologyType> {
168        info!("Detecting Redis topology");
169
170        // Try CLUSTER INFO command
171        match self
172            .execute_command("CLUSTER", &[RespValue::from("INFO")])
173            .await
174        {
175            Ok(RespValue::BulkString(data)) => {
176                let info_str = String::from_utf8(data.to_vec())
177                    .map_err(|e| RedisError::Protocol(format!("Invalid UTF-8: {}", e)))?;
178
179                // Parse cluster_state
180                if Self::is_cluster_info(&info_str) {
181                    info!("Detected Redis Cluster");
182                    return Ok(TopologyType::Cluster);
183                }
184            }
185            Ok(RespValue::SimpleString(info_str)) if Self::is_cluster_info(&info_str) => {
186                // Parse cluster_state
187                info!("Detected Redis Cluster");
188                return Ok(TopologyType::Cluster);
189            }
190            Ok(RespValue::Error(ref e))
191                if e.contains("command not supported")
192                    || e.contains("unknown command")
193                    || e.contains("disabled") =>
194            {
195                // Cluster commands not available - this is standalone
196                info!("Detected Standalone Redis (CLUSTER command not available)");
197                return Ok(TopologyType::Standalone);
198            }
199            Err(RedisError::Server(ref e))
200                if e.contains("command not supported")
201                    || e.contains("unknown command")
202                    || e.contains("disabled") =>
203            {
204                info!("Detected Standalone Redis (CLUSTER command not available)");
205                return Ok(TopologyType::Standalone);
206            }
207            Err(e) => {
208                warn!("Error detecting topology: {:?}, assuming standalone", e);
209                return Ok(TopologyType::Standalone);
210            }
211            _ => {}
212        }
213
214        info!("Detected Standalone Redis");
215        Ok(TopologyType::Standalone)
216    }
217
218    /// Select a database (only works in standalone mode)
219    pub async fn select_database(&mut self, db: u8) -> RedisResult<()> {
220        let response = self
221            .execute_command("SELECT", &[RespValue::from(db as i64)])
222            .await?;
223
224        match response {
225            RespValue::SimpleString(ref s) if s == "OK" => Ok(()),
226            RespValue::Error(e) => Err(RedisError::Server(e)),
227            _ => Err(RedisError::UnexpectedResponse(format!("{:?}", response))),
228        }
229    }
230}
231
232#[async_trait::async_trait]
233impl ProtocolConnection for RedisConnection {
234    async fn send_command(&mut self, command: &RespValue) -> RedisResult<()> {
235        self.send_command(command).await
236    }
237
238    async fn read_response(&mut self) -> RedisResult<RespValue> {
239        self.read_response().await
240    }
241}
242
243/// Connection manager that handles topology detection and connection creation
244pub struct ConnectionManager {
245    config: ConnectionConfig,
246    topology: Option<TopologyType>,
247}
248
249impl ConnectionManager {
250    /// Create a new connection manager
251    pub fn new(config: ConnectionConfig) -> Self {
252        Self {
253            config,
254            topology: None,
255        }
256    }
257
258    /// Get or detect the topology type
259    pub async fn get_topology(&mut self) -> RedisResult<TopologyType> {
260        if let Some(topology) = self.topology {
261            return Ok(topology);
262        }
263
264        // Check if topology mode is forced
265        match self.config.topology_mode {
266            TopologyMode::Standalone => {
267                self.topology = Some(TopologyType::Standalone);
268                Ok(TopologyType::Standalone)
269            }
270            TopologyMode::Cluster => {
271                self.topology = Some(TopologyType::Cluster);
272                Ok(TopologyType::Cluster)
273            }
274            TopologyMode::Auto => {
275                // Auto-detect
276                let endpoints = self.config.parse_endpoints();
277                if endpoints.is_empty() {
278                    return Err(RedisError::Config("No endpoints specified".to_string()));
279                }
280
281                let (host, port) = &endpoints[0];
282                let mut conn = RedisConnection::connect(host, *port, self.config.clone()).await?;
283                let topology = conn.detect_topology().await?;
284                self.topology = Some(topology);
285                Ok(topology)
286            }
287        }
288    }
289
290    /// Create a new connection to the specified host and port
291    pub async fn create_connection(&self, host: &str, port: u16) -> RedisResult<RedisConnection> {
292        RedisConnection::connect(host, port, self.config.clone()).await
293    }
294
295    /// Get the configuration
296    pub fn config(&self) -> &ConnectionConfig {
297        &self.config
298    }
299}
300
301#[cfg(test)]
302mod tests {
303    use super::*;
304
305    #[test]
306    fn test_connection_manager_creation() {
307        let config = ConnectionConfig::new("redis://localhost:6379");
308        let manager = ConnectionManager::new(config);
309        assert!(manager.topology.is_none());
310    }
311
312    #[test]
313    fn test_forced_topology() {
314        let config = ConnectionConfig::new("redis://localhost:6379")
315            .with_topology_mode(TopologyMode::Standalone);
316        let manager = ConnectionManager::new(config);
317
318        // This would normally require async, but we can test the logic
319        assert_eq!(manager.config.topology_mode, TopologyMode::Standalone);
320    }
321}
redis_oxide/connection.rs

redis_oxide/
connection.rs
