feox_server/network/
connection.rs

1use crate::protocol::resp::write_resp_value;
2use crate::protocol::{Command, CommandExecutor, RespParser};
3use feoxdb::FeoxStore;
4use std::os::fd::RawFd;
5use std::sync::Arc;
6
7/// Manages a client connection with RESP protocol handling
8///
9/// Handles command parsing, execution, and response buffering.
10pub struct Connection {
11    fd: RawFd,
12
13    // Protocol parser
14    parser: RespParser,
15    executor: CommandExecutor,
16
17    // Single consolidated write buffer for better performance
18    write_buffer: Vec<u8>,
19    write_position: usize,
20
21    // Pipeline tracking
22    pipeline_depth: usize,
23
24    // Connection state
25    closed: bool,
26}
27
28impl Connection {
29    /// Create a new connection handler
30    pub fn new(fd: RawFd, buffer_size: usize, store: Arc<FeoxStore>) -> Self {
31        let executor = CommandExecutor::new(store);
32        Self {
33            fd,
34            parser: RespParser::new(),
35            executor,
36            write_buffer: Vec::with_capacity(buffer_size),
37            write_position: 0,
38            pipeline_depth: 0,
39            closed: false,
40        }
41    }
42
43    pub fn fd(&self) -> RawFd {
44        self.fd
45    }
46
47    pub fn is_closed(&self) -> bool {
48        self.closed
49    }
50
51    pub fn close(&mut self) {
52        if !self.closed {
53            self.closed = true;
54            unsafe {
55                libc::close(self.fd);
56            }
57        }
58    }
59
60    /// Process incoming data with inline execution
61    pub fn process_read(&mut self, data: &[u8]) -> crate::error::Result<()> {
62        // Feed data to parser
63        self.parser.feed(data);
64
65        // Clear buffer for new batch of responses
66        self.write_buffer.clear();
67        self.write_position = 0;
68
69        // Parse and execute commands inline
70        while let Some(resp_value) = self
71            .parser
72            .parse_next()
73            .map_err(crate::error::Error::Protocol)?
74        {
75            // Parse command
76            let command = Command::from_resp(resp_value).map_err(crate::error::Error::Protocol)?;
77
78            // Check for quit
79            if matches!(command, Command::Quit) {
80                self.closed = true;
81                self.write_buffer.extend_from_slice(b"+OK\r\n");
82                return Ok(());
83            }
84
85            // Execute command and write response directly to buffer
86            let response = self.executor.execute(command);
87            write_resp_value(&mut self.write_buffer, &response);
88
89            self.pipeline_depth += 1;
90        }
91
92        Ok(())
93    }
94
95    /// Get pending write data as a single buffer slice
96    pub fn pending_writes(&mut self) -> Option<&[u8]> {
97        if self.write_position < self.write_buffer.len() {
98            Some(&self.write_buffer[self.write_position..])
99        } else {
100            None
101        }
102    }
103
104    /// Mark bytes as written
105    pub fn consume_writes(&mut self, n: usize) {
106        self.write_position += n;
107    }
108}
109
110impl Drop for Connection {
111    fn drop(&mut self) {
112        self.close();
113    }
114}
feox_server/network/connection.rs

feox_server/network/
connection.rs
