zenith_http1/

connection.rs

//! HTTP/1.1 连接管理
//!
//! 管理单连接上的请求-响应循环（pipelining 支持可选）：
//! - 状态机：Waiting → Reading → Processing → Sending → Waiting/Closed
//! - keep-alive 超时
//! - 最大请求数限制
//! - 优雅关闭

use crate::parser::Http1Parser;
use crate::types::{Http1Config, Http1Error, HttpRequest};

/// HTTP/1.1 连接状态
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Http1ConnectionState {
    /// 等待请求
    Waiting,
    /// 读取请求
    ReadingRequest,
    /// 读取头部
    ReadingHeaders,
    /// 读取请求体
    ReadingBody,
    /// 处理请求
    Processing,
    /// 发送响应
    SendingResponse,
    /// 关闭中
    Closing,
    /// 已关闭
    Closed,
}

/// HTTP/1.1 连接
#[derive(Debug)]
pub struct Http1Connection {
    state: Http1ConnectionState,
    parser: Http1Parser,
    /// 当前请求
    current_request: Option<HttpRequest>,
    /// 是否启用 keep-alive
    keep_alive: bool,
    /// 已处理请求数
    requests_handled: u64,
    /// 是否关闭
    closed: bool,
    /// 上次活动时间戳（毫秒）
    last_activity_ms: u64,
    /// update_time 是否已被调用（区分"未初始化"与"update_time(0)"）
    time_initialized: bool,
}

impl Http1Connection {
    /// 创建连接
    #[inline]
    pub fn new(config: Http1Config) -> Self {
        Self {
            keep_alive: config.keep_alive,
            parser: Http1Parser::new(config),
            state: Http1ConnectionState::Waiting,
            current_request: None,
            requests_handled: 0,
            closed: false,
            last_activity_ms: 0,
            time_initialized: false,
        }
    }

    /// 当前状态
    #[inline]
    pub fn state(&self) -> Http1ConnectionState {
        self.state
    }

    /// 是否关闭
    #[inline]
    pub fn is_closed(&self) -> bool {
        self.closed
    }

    /// 输入数据，尝试解析请求
    ///
    /// # 返回
    /// - Ok(Some(req, body_consumed)): 已解析完整请求，可能包含 body 字节消耗
    /// - Ok((None, consumed)): 需要更多数据
    /// - Err(err): 协议错误（错误会自动关闭连接）
    pub fn on_data(&mut self, data: &[u8]) -> Result<(Option<HttpRequest>, usize), Http1Error> {
        if self.closed {
            return Err(Http1Error::ConnectionClosed);
        }

        // Slowloris 防护：每次收到数据刷新活跃时间戳
        // （含 body/chunked 阶段，M-9：body 中途 idle 超时同样生效）
        // 仅在 update_time 已初始化时间戳时刷新 parser 计时，
        // 否则跳过以避免 first_byte_ms 被误设为 0（struct 默认值）
        if !data.is_empty() && self.time_initialized {
            self.parser.note_activity(self.last_activity_ms);
        }

        let (req, consumed) = match self.parser.feed(data) {
            Ok(v) => v,
            Err(e) => {
                self.on_error(&e);
                return Err(e);
            }
        };

        if let Some(mut req) = req {
            // 检查 Connection: close
            if !req.keep_alive {
                self.keep_alive = false;
            }
            // Http1Parser 现已内置完整 chunked 请求体解析（RFC 7230 §4.1），
            // 当 req.chunked == true 时 req.body 已包含解码后的完整 body，
            // 无需上层通过 body_complete() 推送。
            // - Content-Length > 0 且 req.body 非空：Http1Parser 已同步读取 body，直接进入 Processing
            // - Content-Length > 0 且 req.body 为空：兼容模式，进入 ReadingBody 等待外部 body_complete()
            // - Content-Length == 0 或 None：无 body，直接处理
            if req.chunked {
                // chunked body 已由 parser 完整解码，直接进入 Processing
                self.state = Http1ConnectionState::Processing;
            } else if let Some(len) = req.content_length {
                if len == 0 {
                    self.state = Http1ConnectionState::Processing;
                } else if !req.body.is_empty() && req.body.len() as u64 == len {
                    // Http1Parser 已按 Content-Length 完整读取 body，无需再等待
                    self.state = Http1ConnectionState::Processing;
                } else {
                    // 兼容旧语义：parser 未读取 body，等待上层调用 body_complete()
                    self.state = Http1ConnectionState::ReadingBody;
                }
            } else if req.line.method.as_ref() == "POST" || req.line.method.as_ref() == "PUT" {
                // 方法无 content-length 时视为 body 长度为 0
                req.content_length = Some(0);
                self.state = Http1ConnectionState::Processing;
            } else {
                self.state = Http1ConnectionState::Processing;
            }
            self.current_request = Some(req);
            self.requests_handled += 1;
            Ok((self.current_request.clone(), consumed))
        } else {
            // parser 尚未返回完整请求（仍在读取 header / body / chunked 帧）
            // 连接层状态反映 parser 的活跃状态
            self.state = match self.parser.state() {
                crate::parser::ParserState::WaitingRequest
                | crate::parser::ParserState::ReadingRequest => Http1ConnectionState::ReadingRequest,
                crate::parser::ParserState::ReadingHeaders => Http1ConnectionState::ReadingHeaders,
                crate::parser::ParserState::ReadingBody
                | crate::parser::ParserState::ReadingChunkSize
                | crate::parser::ParserState::ReadingChunkData
                | crate::parser::ParserState::ReadingChunkTrailer => Http1ConnectionState::ReadingBody,
                crate::parser::ParserState::HeadersComplete => Http1ConnectionState::Processing,
                crate::parser::ParserState::Error => Http1ConnectionState::Closed,
            };
            Ok((None, consumed))
        }
    }

    /// 设置请求体已接收完成
    pub fn body_complete(&mut self, body: Vec<u8>) {
        if let Some(req) = self.current_request.as_mut() {
            req.body = body;
            self.state = Http1ConnectionState::Processing;
        }
    }

    /// 获取当前请求
    #[inline]
    pub fn current_request(&self) -> Option<&HttpRequest> {
        self.current_request.as_ref()
    }

    /// 响应发送完成
    pub fn response_sent(&mut self) {
        self.current_request = None;
        if !self.keep_alive {
            self.state = Http1ConnectionState::Closing;
            self.closed = true;
        } else {
            // 准备下一个请求
            self.parser.reset();
            self.state = Http1ConnectionState::Waiting;
        }
    }

    /// 处理错误：关闭连接
    pub fn on_error(&mut self, _err: &Http1Error) {
        self.state = Http1ConnectionState::Closing;
        self.closed = true;
    }

    /// 获取已处理请求数
    #[inline]
    pub fn requests_handled(&self) -> u64 {
        self.requests_handled
    }

    /// 更新当前时间戳（由事件循环每周期调用）
    ///
    /// 必须在首次 [`Self::on_data`] 之前调用至少一次，否则 parser 的
    /// `first_byte_ms` 不会被设置（note_activity 被跳过），Slowloris
    /// 空闲超时检测在首字节阶段不生效。调用后 `time_initialized` 置 true，
    /// 后续 `on_data` 才会向 parser 刷新活跃时间戳。
    #[inline]
    pub fn update_time(&mut self, now_ms: u64) {
        self.last_activity_ms = now_ms;
        self.time_initialized = true;
    }

    /// 检查空闲超时（Slowloris 防护）
    ///
    /// 由事件循环在每个周期调用。如果请求解析超时，关闭连接。
    pub fn check_timeout(&mut self) -> Result<(), Http1Error> {
        if self.closed {
            return Ok(());
        }
        if let Err(e) = self.parser.check_idle_timeout(self.last_activity_ms) {
            self.on_error(&e);
            return Err(e);
        }
        Ok(())
    }

    /// 关闭连接
    pub fn close(&mut self) {
        self.state = Http1ConnectionState::Closed;
        self.closed = true;
    }
}

impl Default for Http1Connection {
    fn default() -> Self {
        Self::new(Http1Config::new())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_connection_basic() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let (req, _) = conn.on_data(input).unwrap();
        assert!(req.is_some());
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_lifecycle() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        conn.on_data(input).unwrap();
        assert_eq!(conn.state(), Http1ConnectionState::Processing);

        conn.response_sent();
        assert_eq!(conn.state(), Http1ConnectionState::Waiting);
        assert!(!conn.is_closed());
    }

    #[test]
    fn test_connection_close() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input =
            b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n";
        conn.on_data(input).unwrap();
        conn.response_sent();
        assert!(conn.is_closed());
    }

    #[test]
    fn test_connection_error_cases() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let r = conn.on_data(b"INV HTTP/1.1\r\nHost: x\r\n\r\n");
        assert!(r.is_err());
        assert!(conn.is_closed());
    }

    #[test]
    fn test_connection_body_post() {
        let mut conn = Http1Connection::new(Http1Config::new());
        // Http1Parser 在 Content-Length 模式下会同步读取完整 body，因此此处需要同时提供 headers + body
        //（不再走 ReadingBody → body_complete 路径，而是直接进入 Processing）
        let input =
            b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: 5\r\n\r\nhello";
        let (req, _) = conn.on_data(input).unwrap();
        let req = req.unwrap();
        assert_eq!(req.content_length, Some(5));
        assert_eq!(req.body, b"hello");
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_state_variants() {
        let states = [
            Http1ConnectionState::Waiting,
            Http1ConnectionState::ReadingRequest,
            Http1ConnectionState::ReadingHeaders,
            Http1ConnectionState::ReadingBody,
            Http1ConnectionState::Processing,
            Http1ConnectionState::SendingResponse,
            Http1ConnectionState::Closing,
            Http1ConnectionState::Closed,
        ];
        for (i, s) in states.iter().enumerate() {
            assert_eq!(*s, states[i]);
        }
        assert_ne!(Http1ConnectionState::Waiting, Http1ConnectionState::Closed);
    }

    #[test]
    fn test_connection_default() {
        let conn = Http1Connection::default();
        assert_eq!(conn.state(), Http1ConnectionState::Waiting);
        assert!(!conn.is_closed());
        assert_eq!(conn.requests_handled(), 0);
    }

    #[test]
    fn test_connection_keep_alive_multiple_requests() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input1 = b"GET /1 HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let (req1, _) = conn.on_data(input1).unwrap();
        assert!(req1.is_some());
        assert_eq!(conn.requests_handled(), 1);

        conn.response_sent();
        assert_eq!(conn.state(), Http1ConnectionState::Waiting);

        let input2 = b"GET /2 HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let (req2, _) = conn.on_data(input2).unwrap();
        assert!(req2.is_some());
        assert_eq!(conn.requests_handled(), 2);
    }

    #[test]
    fn test_connection_closed_rejects_data() {
        let mut conn = Http1Connection::new(Http1Config::new());
        conn.close();
        assert!(conn.is_closed());
        assert_eq!(conn.state(), Http1ConnectionState::Closed);

        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let r = conn.on_data(input);
        assert!(r.is_err());
        assert!(matches!(r.unwrap_err(), Http1Error::ConnectionClosed));
    }

    #[test]
    fn test_connection_current_request() {
        let mut conn = Http1Connection::new(Http1Config::new());
        assert!(conn.current_request().is_none());

        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        conn.on_data(input).unwrap();
        assert!(conn.current_request().is_some());
        assert_eq!(conn.current_request().unwrap().line.method.as_ref(), "GET");
    }

    #[test]
    fn test_connection_chunked_body_state() {
        let mut conn = Http1Connection::new(Http1Config::new());
        // 完整 chunked 请求（RFC 7230 §4.1）：headers + chunk + last-chunk + trailer CRLF
        let input = b"POST /upload HTTP/1.1\r\nHost: example.com\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n0\r\n\r\n";
        let (req, _) = conn.on_data(input).unwrap();
        assert!(req.is_some());
        let req = req.unwrap();
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
        assert_eq!(req.body, b"Hello");
    }

    #[test]
    fn test_connection_chunked_body_partial() {
        let mut conn = Http1Connection::new(Http1Config::new());
        // 仅 headers，未包含 chunk 数据 → parser 返回 None，连接进入 ReadingBody
        let part1 = b"POST /upload HTTP/1.1\r\nHost: example.com\r\nTransfer-Encoding: chunked\r\n\r\n";
        let (req, _) = conn.on_data(part1).unwrap();
        assert!(req.is_none());
        assert_eq!(conn.state(), Http1ConnectionState::ReadingBody);

        // 后续 chunk 数据到达 → parser 完成解析，连接进入 Processing
        let part2 = b"5\r\nHello\r\n0\r\n\r\n";
        let (req, _) = conn.on_data(part2).unwrap();
        assert!(req.is_some());
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
        assert_eq!(req.unwrap().body, b"Hello");
    }

    #[test]
    fn test_connection_zero_content_length() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: 0\r\n\r\n";
        let (req, _) = conn.on_data(input).unwrap();
        assert!(req.is_some());
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_put_method() {
        let mut conn = Http1Connection::new(Http1Config::new());
        // Http1Parser 同步读取 Content-Length body，提供完整 headers+body 直接进入 Processing
        let input =
            b"PUT /resource HTTP/1.1\r\nHost: example.com\r\nContent-Length: 5\r\n\r\nworld";
        let (req, _) = conn.on_data(input).unwrap();
        let req = req.unwrap();
        assert_eq!(req.line.method.as_ref(), "PUT");
        assert_eq!(req.body, b"world");
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_head_method() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"HEAD / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let (req, _) = conn.on_data(input).unwrap();
        assert!(req.is_some());
        assert_eq!(req.unwrap().line.method.as_ref(), "HEAD");
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_on_error_closes_connection() {
        let mut conn = Http1Connection::new(Http1Config::new());
        conn.on_error(&Http1Error::SyntaxError("test".into()));
        assert!(conn.is_closed());
        assert_eq!(conn.state(), Http1ConnectionState::Closing);
    }

    #[test]
    fn test_connection_body_complete_no_current_request() {
        let mut conn = Http1Connection::new(Http1Config::new());
        conn.body_complete(b"test".to_vec());
        assert_eq!(conn.state(), Http1ConnectionState::Waiting);
    }

    #[test]
    fn test_connection_response_sent_clears_request() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        conn.on_data(input).unwrap();
        assert!(conn.current_request().is_some());
        conn.response_sent();
        assert!(conn.current_request().is_none());
    }

    #[test]
    fn test_connection_closing_state() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n";
        conn.on_data(input).unwrap();
        conn.response_sent();
        assert_eq!(conn.state(), Http1ConnectionState::Closing);
        assert!(conn.is_closed());
    }

    #[test]
    fn test_connection_debug_format() {
        let conn = Http1Connection::new(Http1Config::new());
        let s = format!("{:?}", conn);
        assert!(!s.is_empty());
    }

    #[test]
    fn test_connection_idle_timeout_during_body_read() {
        // M-9：body 中途静默超过 idle_timeout → 连接层 IdleTimeout 并关闭
        let config = Http1Config::new().with_idle_timeout_ms(30_000);
        let mut conn = Http1Connection::new(config);
        conn.update_time(0);
        // 仅头部到达（CL=10，body 未到）→ ReadingBody
        let (req, _) = conn
            .on_data(b"POST / HTTP/1.1\r\nHost: example.com\r\nContent-Length: 10\r\n\r\n")
            .unwrap();
        assert!(req.is_none());
        assert_eq!(conn.state(), Http1ConnectionState::ReadingBody);
        // 30s+ 未再收到 body → IdleTimeout，连接 fail-closed
        conn.update_time(30_001);
        let r = conn.check_timeout();
        assert!(matches!(r, Err(Http1Error::IdleTimeout)), "实际 {r:?}");
        assert!(conn.is_closed());
    }

    #[test]
    fn test_connection_body_activity_prevents_timeout() {
        // M-9 对照：body 分片持续到达刷新活跃，不得误判超时
        let config = Http1Config::new().with_idle_timeout_ms(30_000);
        let mut conn = Http1Connection::new(config);
        conn.update_time(0);
        let (req, _) = conn
            .on_data(b"POST / HTTP/1.1\r\nHost: example.com\r\nContent-Length: 6\r\n\r\nhe")
            .unwrap();
        assert!(req.is_none());
        // 20s 后 body 第二片到达
        conn.update_time(20_000);
        let (req, _) = conn.on_data(b"ll").unwrap();
        assert!(req.is_none());
        // 活跃后 25s：不得超时
        conn.update_time(45_000);
        assert!(conn.check_timeout().is_ok(), "活跃刷新后不得超时");
        // 最后一片凑满 6 字节 body 完成请求
        let (req, _) = conn.on_data(b"lo").unwrap();
        assert!(req.is_some());
    }
}