rave/src/protocol/httpflv_server.rs

//! HTTP-FLV 拉流服务器（异步版本）
//!
//! 基于 tokio 异步运行时：
//! - 使用 `tokio::net::TcpListener` 异步 accept
//! - 每个连接通过 `tokio::spawn` 处理
//! - 异步读写 + 1ms 轮询订阅者队列

use std::sync::Arc;
use std::time::Duration;

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;

use crate::log_error;
use crate::log_info;

use crate::protocol::httpflv::{self, HttpRequestBuffer};
use crate::remux::rtmp2flv;
use crate::sdk::context::EngineContext;
use crate::sdk::types::MediaType;
use crate::stats::ServerStats;

/// HTTP-FLV 服务器配置
#[derive(Debug, Clone)]
pub struct HttpFlvServerConfig {
    /// 监听地址
    pub listen_addr: String,
    /// 监听端口（默认 8080）
    pub port: u16,
}

impl Default for HttpFlvServerConfig {
    fn default() -> Self {
        Self {
            listen_addr: "0.0.0.0".to_string(),
            port: 8080,
        }
    }
}

/// HTTP-FLV 服务器
pub struct HttpFlvServer {
    /// 服务器配置
    config: HttpFlvServerConfig,
    /// 引擎上下文
    context: Arc<EngineContext>,
}

impl HttpFlvServer {
    /// 创建 HTTP-FLV 服务器
    pub fn new(config: HttpFlvServerConfig, context: Arc<EngineContext>) -> Self {
        Self { config, context }
    }

    /// 异步监听并处理连接
    pub async fn listen(&self) -> Result<(), String> {
        let addr = format!("{}:{}", self.config.listen_addr, self.config.port);
        let listener = tokio::net::TcpListener::bind(&addr)
            .await
            .map_err(|e| format!("httpflv bind {} failed: {}", addr, e))?;

        loop {
            match listener.accept().await {
                Ok((tcp_stream, _addr)) => {
                    let context = self.context.clone();
                    tokio::spawn(async move {
                        handle_http_flv_client(tcp_stream, context).await;
                    });
                }
                Err(e) => {
                    log_error!("[httpflv] accept error: {}", e);
                }
            }
        }
    }
}

/// 异步处理单个 HTTP-FLV 客户端连接
async fn handle_http_flv_client(mut stream: TcpStream, context: Arc<EngineContext>) {
    let peer = stream.peer_addr().map(|a| a.to_string()).unwrap_or_default();
    log_info!("[httpflv] client connected from {}", peer);

    // 1. 读取 HTTP 请求头
    let mut req_buf = HttpRequestBuffer::new();
    let mut read_buf = [0u8; 4096];
    loop {
        let n = match tokio::time::timeout(Duration::from_secs(5), stream.read(&mut read_buf)).await
        {
            Ok(Ok(0)) => return,
            Ok(Ok(n)) => n,
            Ok(Err(_)) | Err(_) => return,
        };
        if req_buf.feed(&read_buf[..n]) {
            break;
        }
    }

    let request = match req_buf.parse() {
        Ok(r) => r,
        Err(_) => {
            let _ = stream.write_all(httpflv::build_http_404_response().as_bytes()).await;
            return;
        }
    };

    // API 端点
    if request.path == "/api/streams" {
        handle_api_streams(&mut stream, &context).await;
        return;
    }
    if request.path == "/api/stats" {
        handle_api_stats(&mut stream, &context).await;
        return;
    }

    // 验证 FLV 请求
    if !httpflv::is_http_flv_request(&request) {
        let _ = stream.write_all(httpflv::build_http_404_response().as_bytes()).await;
        return;
    }

    let (app, stream_name) = match httpflv::parse_stream_path(&request.path) {
        Some(pair) => pair,
        None => {
            let _ = stream.write_all(httpflv::build_http_404_response().as_bytes()).await;
            return;
        }
    };

    let stream_path = crate::sdk::types::StreamPath::new(&app, &stream_name);
    let sm = context.stream_manager();

    let subscriber = match sm.subscribe(&stream_path) {
        Ok(sub) => {
            log_info!("[httpflv:{}] pull start {}", peer, stream_path);
            sub
        }
        Err(_) => {
            let _ = stream.write_all(httpflv::build_http_404_response().as_bytes()).await;
            return;
        }
    };

    let stats = context.get_service::<ServerStats>();
    if let Some(ref s) = stats {
        s.on_pull_connect();
    }

    // 发送 HTTP 200 + FLV Header
    let http_resp = httpflv::build_http_flv_response();
    if stream.write_all(http_resp.as_bytes()).await.is_err() {
        if let Some(ref s) = stats { s.on_pull_disconnect(); }
        return;
    }
    if let Some(ref s) = stats { s.add_bytes_out(http_resp.len() as u64); }

    let flv_header = rtmp2flv::flv_header_with_audio_video();
    if stream.write_all(&flv_header).await.is_err() {
        if let Some(ref s) = stats { s.on_pull_disconnect(); }
        return;
    }
    if let Some(ref s) = stats { s.add_bytes_out(flv_header.len() as u64); }

    // 循环读取帧 → FLV Tag → 发送
    let mut last_frame_time = tokio::time::Instant::now();
    let poll_interval = Duration::from_millis(1);

    loop {
        match subscriber.read_frame() {
            Some(frame) => {
                last_frame_time = tokio::time::Instant::now();
                if let Some(tag) = avframe_to_flv_tag(&frame) {
                    let encoded = tag.encode();
                    if stream.write_all(&encoded).await.is_err() {
                        break;
                    }
                    if let Some(ref s) = stats {
                        s.add_bytes_out(encoded.len() as u64);
                        if frame.is_video() {
                            s.inc_video_frames_out();
                        } else if frame.is_audio() {
                            s.inc_audio_frames_out();
                        }
                    }
                }
            }
            None => {
                // 3 秒无帧且流已移除则退出
                if last_frame_time.elapsed() > Duration::from_secs(3) {
                    if !sm.has_stream(&stream_path) {
                        break;
                    }
                    last_frame_time = tokio::time::Instant::now();
                }
                tokio::time::sleep(poll_interval).await;
            }
        }

        if !subscriber.is_active() {
            break;
        }
    }

    subscriber.close();
    log_info!("[httpflv:{}] pull stop {}", peer, stream_path);
    if let Some(ref s) = stats { s.on_pull_disconnect(); }
}

/// 将 AVFrame 转为 FLV Tag
fn avframe_to_flv_tag(frame: &crate::sdk::types::AVFrame) -> Option<crate::codec::flv::FlvTag> {
    let msg_type = match frame.media_type {
        MediaType::Audio => crate::protocol::rtmp::chunk::msg_type_id::AUDIO,
        MediaType::Video => crate::protocol::rtmp::chunk::msg_type_id::VIDEO,
        MediaType::Data => crate::protocol::rtmp::chunk::msg_type_id::DATA_AMF0,
    };
    rtmp2flv::rtmp_msg_to_flv_tag(msg_type, frame.timestamp_ms as u32, &frame.data)
}

/// 处理 /api/stats
async fn handle_api_stats(stream: &mut TcpStream, context: &EngineContext) {
    let stats = context.get_service::<ServerStats>();
    let sm = context.stream_manager();

    let (snap, stream_count) = match (stats, sm) {
        (Some(s), sm) => (s.snapshot(), sm.stream_count()),
        (None, sm) => {
            let json = format!(
                "{{\"uptime_secs\":0,\"streams\":{},\"push_connections\":0,\"pull_connections\":0}}",
                sm.stream_count()
            );
            let resp = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\nCache-Control: no-cache\r\n\r\n{}",
                json.len(), json
            );
            let _ = stream.write_all(resp.as_bytes()).await;
            return;
        }
    };

    let uptime = if snap.uptime_secs > 0 { snap.uptime_secs } else { 1 };
    let in_kbps = snap.bytes_in / uptime * 8 / 1000;
    let out_kbps = snap.bytes_out / uptime * 8 / 1000;
    let video_fps_in = snap.video_frames_in / uptime;
    let audio_fps_in = snap.audio_frames_in / uptime;
    let video_fps_out = snap.video_frames_out / uptime;
    let audio_fps_out = snap.audio_frames_out / uptime;

    let json = format!(
        "{{\"uptime_secs\":{},\"streams\":{},\"bytes_in\":{},\"bytes_out\":{},\"in_bitrate_kbps\":{},\"out_bitrate_kbps\":{},\"video_frames_in\":{},\"audio_frames_in\":{},\"video_frames_out\":{},\"audio_frames_out\":{},\"video_fps_in\":{},\"audio_fps_in\":{},\"video_fps_out\":{},\"audio_fps_out\":{},\"push_connections\":{},\"pull_connections\":{},\"total_push_connections\":{},\"total_pull_connections\":{}}}",
        snap.uptime_secs, stream_count, snap.bytes_in, snap.bytes_out,
        in_kbps, out_kbps, snap.video_frames_in, snap.audio_frames_in,
        snap.video_frames_out, snap.audio_frames_out,
        video_fps_in, audio_fps_in, video_fps_out, audio_fps_out,
        snap.push_connections, snap.pull_connections,
        snap.total_push_connections, snap.total_pull_connections,
    );

    let resp = format!(
        "HTTP/1.1 200 OK\r\nContent-Type: application/json; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\nCache-Control: no-cache\r\n\r\n{}",
        json.len(), json
    );
    let _ = stream.write_all(resp.as_bytes()).await;
}

/// 处理 /api/streams
async fn handle_api_streams(stream: &mut TcpStream, context: &EngineContext) {
    let sm = context.stream_manager();
    let summaries = sm.stream_summaries();

    let mut json = String::from("{\"streams\":[");
    for (i, s) in summaries.iter().enumerate() {
        if i > 0 { json.push(','); }
        let vcodec = codec_name_video(s.video_codec);
        let acodec = codec_name_audio(s.audio_codec);
        let vts = match s.last_video_timestamp_ms {
            Some(t) => format!("{}", t),
            None => "null".to_string(),
        };
        let ats = match s.last_audio_timestamp_ms {
            Some(t) => format!("{}", t),
            None => "null".to_string(),
        };
        json.push_str(&format!(
            "{{\"path\":\"{}/{}\",\"video_codec\":\"{}\",\"audio_codec\":\"{}\",\"subscribers\":{},\"last_video_ts\":{},\"last_audio_ts\":{},\"gop_cache_size\":{},\"total_video_frames\":{},\"total_audio_frames\":{}}}",
            s.path.app_name, s.path.stream_name, vcodec, acodec,
            s.subscriber_count, vts, ats, s.gop_cache_size,
            s.total_video_frames, s.total_audio_frames,
        ));
    }
    json.push_str("]}");

    let resp = format!(
        "HTTP/1.1 200 OK\r\nContent-Type: application/json; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\nCache-Control: no-cache\r\n\r\n{}",
        json.len(), json
    );
    let _ = stream.write_all(resp.as_bytes()).await;
}

fn codec_name_video(codec: crate::sdk::types::VideoCodec) -> &'static str {
    match codec {
        crate::sdk::types::VideoCodec::H264 => "H264",
        crate::sdk::types::VideoCodec::H265 => "H265",
        _ => "Unknown",
    }
}

fn codec_name_audio(codec: crate::sdk::types::AudioCodec) -> &'static str {
    match codec {
        crate::sdk::types::AudioCodec::Aac => "AAC",
        crate::sdk::types::AudioCodec::G711A => "G711A",
        crate::sdk::types::AudioCodec::G711U => "G711U",
        crate::sdk::types::AudioCodec::Opus => "Opus",
        _ => "Unknown",
    }
}

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

    #[test]
    fn test_http_flv_server_config_default() {
        let config = HttpFlvServerConfig::default();
        assert_eq!(config.port, 8080);
    }
}