charging-cabinet/software/device-server/src/tcp/server.rs

//! TCP 服务端：监听端口、接收设备连接、按 LF 分割 JSON 消息。

use futures::StreamExt;
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc;
use tokio::time;
use tokio_util::codec::{FramedRead, LinesCodec};

use super::connection::{ConnectionPool, DeviceConnection};
use super::handler::handle_message;
use super::protocol::DeviceMessage;

/// 心跳超时：5000ms 未收到消息则断开连接。
const HEARTBEAT_TIMEOUT: Duration = Duration::from_millis(5000);
/// 清理超时连接的间隔。
const CLEANUP_INTERVAL: Duration = Duration::from_secs(10);

/// 启动 TCP 监听循环。
pub async fn run_tcp_server(
    addr: String,
    pool: ConnectionPool,
    redis: redis::aio::ConnectionManager,
) {
    let listener = TcpListener::bind(&addr)
        .await
        .expect("TCP 端口绑定失败");
    tracing::info!("TCP 设备服务启动: {}", addr);

    // 后台清理过期连接
    let cleanup_pool = pool.clone();
    tokio::spawn(async move {
        let mut interval = time::interval(CLEANUP_INTERVAL);
        loop {
            interval.tick().await;
            let removed = cleanup_pool.cleanup_timeout(HEARTBEAT_TIMEOUT).await;
            for dev_id in &removed {
                tracing::warn!("[心跳超时] 断开 dev_id={}", dev_id);
            }
        }
    });

    loop {
        match listener.accept().await {
            Ok((stream, peer)) => {
                tracing::info!("[新连接] peer={}", peer);
                let pool = pool.clone();
                let redis = redis.clone();
                tokio::spawn(handle_connection(stream, peer, pool, redis));
            }
            Err(e) => {
                tracing::error!("accept 失败: {}", e);
            }
        }
    }
}

/// 处理单条 TCP 连接。
async fn handle_connection(
    stream: TcpStream,
    peer: std::net::SocketAddr,
    pool: ConnectionPool,
    redis: redis::aio::ConnectionManager,
) {
    let (reader, writer) = tokio::io::split(stream);
    let mut lines = FramedRead::new(reader, LinesCodec::new_with_max_length(4096));

    // 写通道：HTTP 命令下发通过此通道写入 TCP
    let (tx, mut rx) = mpsc::unbounded_channel::<String>();

    let mut current_dev_id: Option<String> = None;
    let mut writer = writer;

    loop {
        tokio::select! {
            // 读取设备消息
            line = lines.next() => {
                match line {
                    Some(Ok(text)) => {
                        if text.trim().is_empty() {
                            continue;
                        }

                        let msg: DeviceMessage = match serde_json::from_str(&text) {
                            Ok(m) => m,
                            Err(e) => {
                                tracing::warn!("[{}] JSON 解析失败: {} raw={}", peer, e, text);
                                continue;
                            }
                        };

                        // 如果是 login/auth_str，需要 dev_id 来关联连接
                        if (msg.act == "auth_str" || msg.act == "login")
                            && let Some(ref dev_id) = msg.dev_id
                                && current_dev_id.is_none() {
                                    let conn = DeviceConnection::new(dev_id.clone(), tx.clone());
                                    pool.register(dev_id.clone(), conn).await;
                                    current_dev_id = Some(dev_id.clone());
                                }

                        // 更新活跃时间
                        if let Some(ref dev_id) = current_dev_id {
                            pool.touch(dev_id).await;
                        }

                        // 处理消息
                        let resp = handle_message(msg, &pool, &redis).await;

                        if let Some(resp) = resp {
                            let json = resp.to_json();
                            if let Err(e) = tx.send(json) {
                                tracing::error!("[{}] 响应发送失败: {}", peer, e);
                                break;
                            }
                        }
                    }
                    Some(Err(e)) => {
                        tracing::warn!("[{}] 读取错误: {}", peer, e);
                        break;
                    }
                    None => {
                        tracing::info!("[{}] 连接关闭", peer);
                        break;
                    }
                }
            }
            // 向设备写指令
            Some(json) = rx.recv() => {
                use tokio::io::AsyncWriteExt;
                let framed = format!("{}\n", json);
                if let Err(e) = writer.write_all(framed.as_bytes()).await {
                    tracing::error!("[{}] 写入失败: {}", peer, e);
                    break;
                }
            }
        }
    }

    // 连接断开，清理
    if let Some(ref dev_id) = current_dev_id {
        pool.remove(dev_id).await;
        tracing::info!("[断连] dev_id={}", dev_id);
    }
}