server1/RTPServer/RtpServer.cpp

#include "RtpServer.h"

#include <QTcpSocket>
#include <QNetworkInterface>
#include <QHostAddress>

#include "ThreadManager.h"
#include "ThreadCompareItemManager.h"
#include "SystemConfig.h"


RTPServer::RTPServer(QObject *parent)
    : QObject(parent), m_tcpServer(nullptr), m_port(8808)
{
    
}

RTPServer::~RTPServer()
{
    if (m_tcpServer) 
    {
        m_tcpServer->close();
        delete m_tcpServer;
        m_tcpServer = nullptr;
    }
}

/**
 * @brief 启动RTP服务
 * @param port 监听端口
 * @return 成功返回true，失败返回false
 */
bool RTPServer::thread_task(int port)
{
    m_logger = spdlog::get("RTPServer");
    if (!m_logger) 
    {
        fmt::print("RTPServer: Failed to get logger instance.\n");
        return false;
    }

    SPDLOG_LOGGER_INFO(m_logger, "♫ 开启 RTPServer 服务线程");

    if(m_tcpServer == nullptr)
    {
        m_tcpServer = new QTcpServer(this);
    }

    m_port = port;
    if(!m_tcpServer->listen(QHostAddress::Any, m_port))
    {
        SPDLOG_LOGGER_ERROR(m_logger, "Failed to start RTP server on port {}", m_port);
        SPDLOG_LOGGER_ERROR(m_logger, "错误信息: {}", m_tcpServer->errorString().toStdString());
        return false;
    }

    /* 查找到出口IP */
    m_localIP = SysConfig.getBaseConfigSrc().strServerIP;
    if(m_localIP.isEmpty())
    {
        SPDLOG_LOGGER_ERROR(m_logger, "未设置服务器监听IP，无法监听客户端连接");
    }
    connect(m_tcpServer, &QTcpServer::newConnection, this, &RTPServer::do_newConnection);
    connect(m_tcpServer, &QTcpServer::acceptError, this, &RTPServer::do_error);

    /* 开启心跳定时器 */
    m_timerHeartbeat.setTimerType(Qt::PreciseTimer);
    m_timerHeartbeat.setSingleShot(false);
    m_timerHeartbeat.setInterval(1000);
    connect(&m_timerHeartbeat, &QTimer::timeout, this, &RTPServer::do_heartbeatTimeout);

    m_timerHeartbeat.start();

    /* 开启事件循环 */
    m_isStoped.store(false);
    m_eventLoop.exec();
    m_isStoped.store(true);
    SPDLOG_LOGGER_INFO(m_logger, "♫ RTPServer 线程结束运行");
    return true;
}

/* 停止线程 */
void RTPServer::stopThreadBlock()
{
    if(m_eventLoop.isRunning())
    {
        m_eventLoop.quit();
    }
    while(!m_isStoped.load())
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}

/* 处理新连接 */
void RTPServer::do_newConnection()
{
    QTcpSocket* clientSocket = m_tcpServer->nextPendingConnection();
    if (clientSocket == nullptr) 
    {
        SPDLOG_LOGGER_ERROR(m_logger, "Failed to get pending connection.");
        return;
    }

    connect(clientSocket, &QTcpSocket::disconnected, this, &RTPServer::do_disconnect);
    connect(clientSocket, &QTcpSocket::readyRead, this, &RTPServer::do_receiveMessage);
    SPDLOG_LOGGER_INFO(m_logger, "新的客户端连接: {}", clientSocket->peerAddress().toString().toStdString());
    m_listClients.append(clientSocket);  // 添加到客户端列表中
}

/* 监听服务错误 */
void RTPServer::do_error(QAbstractSocket::SocketError socketError)
{
    SPDLOG_LOGGER_ERROR(m_logger, "RTP Server error: {}, {}", static_cast<int>(socketError), m_tcpServer->errorString().toStdString());
}

/* 断开连接 */
void RTPServer::do_disconnect()
{
    QTcpSocket* clientSocket = qobject_cast<QTcpSocket*>(sender());
    if (clientSocket) 
    {
        SPDLOG_LOGGER_INFO(m_logger, "客户端断开连接: {}", clientSocket->peerAddress().toString().toStdString());
        clientSocket->deleteLater();
        m_listClients.removeAll(clientSocket);
    } else 
    {
        SPDLOG_LOGGER_ERROR(m_logger, "Failed to cast sender to QTcpSocket.");
    }
}

/* 接收消息 */
void RTPServer::do_receiveMessage()
{
    QTcpSocket* clientSocket = qobject_cast<QTcpSocket*>(sender());
    if (clientSocket == nullptr) 
    {
        SPDLOG_LOGGER_ERROR(m_logger, "Failed to cast sender to QTcpSocket.");
        return;
    }
    QString recvIP = clientSocket->peerAddress().toString();
    quint16 recvPort = clientSocket->peerPort();

    QByteArray data = clientSocket->readAll();
    if (data.isEmpty()) 
    {
        SPDLOG_LOGGER_WARN(m_logger, "Received empty data from client: {}", recvIP.toStdString());
        return;
    }

    SPDLOG_LOGGER_TRACE(m_logger, "RTPServer从: {}:{} 接收到原始数据: {}, 大小: {}", 
                        recvIP.toStdString(), recvPort, data.toHex().toStdString(), data.size());
    // SPDLOG_LOGGER_TRACE(m_logger, "RTPServer从: {}:{} 接收到数据: {}", 
    //                     clientSocket->peerAddress().toString().toStdString(), clientSocket->peerPort(), 
    //                     QString(data).toStdString());
    // return;

    if(static_cast<unsigned long>(data.size()) != sizeof(RtpRecvClientInfo_t)) 
    {
        SPDLOG_LOGGER_WARN(m_logger, "从客户端接收到的数据大小: {} 不等于应有的大小 {}", data.size(), sizeof(RtpRecvClientInfo_t));
        return;
    }


    /* 处理接收到的数据 */
    RtpSendClientInfo_t sendInfo;
    RtpRecvClientInfo_t* recvInfo = reinterpret_cast<RtpRecvClientInfo_t*>(data.data());

    SPDLOG_LOGGER_DEBUG(m_logger, "客户端信息: ");
    SPDLOG_LOGGER_DEBUG(m_logger, "    客户端IP: {}", recvInfo->clientIP);
    SPDLOG_LOGGER_DEBUG(m_logger, "    客户端端口: {}", recvInfo->clientPort);
    SPDLOG_LOGGER_DEBUG(m_logger, "    对比项ID: {}", recvInfo->compareItemID);
    SPDLOG_LOGGER_DEBUG(m_logger, "    对比项通道号: {}", recvInfo->compareItemRoadNum);
    SPDLOG_LOGGER_DEBUG(m_logger, "    包类型: {}", recvInfo->type);


    /* 解析客户端发送的信息 */
    sendInfo.clientIP = recvInfo->clientIP;
    sendInfo.clientPort = recvInfo->clientPort;
    // sendInfo.compareItemID = recvInfo->compareItemID;
    // sendInfo.compareItemRoadNum = recvInfo->compareItemRoadNum;
    /* 根据对比项ID和通道编号查找到使用到的声卡通道编号 */
    SoundCardRoadInfo_t roadInfo = CompareItemManager.getSoundCardRoadInfo(recvInfo->compareItemID, recvInfo->compareItemRoadNum);
    if(roadInfo.nSoundCardNum < 0 || roadInfo.roadInfo.nRoadNum < 0)
    {
        SPDLOG_LOGGER_ERROR(m_logger, "无法获取对比项 {} 通道 {} 的声卡信息", 
                            recvInfo->compareItemID, recvInfo->compareItemRoadNum);
        return;
    }
    sendInfo.SoundCardNum = roadInfo.nSoundCardNum;
    sendInfo.SoundCardRoadNum = roadInfo.roadInfo.nRoadNum;

    switch(recvInfo->type)
    {
        case 0: // 登录请求
            handleLogin(sendInfo);
            break;
        case 1: // 心跳请求
            handleHeartbeat(sendInfo);
            break;
        case 2: // 注销请求
            handleLogout(sendInfo);
            break;
        default:
            SPDLOG_LOGGER_WARN(m_logger, "来自客户端未知的请求:{}, {}:{}", recvInfo->type, sendInfo.clientIP.toStdString(), sendInfo.clientPort);
            break;
    }
}

/* 一个通道UDP关闭的信号 */
void RTPServer::do_udpClosed(int soundCardNum, int roadNum)
{
    SPDLOG_LOGGER_DEBUG(m_logger, "接收到释放本地端口的信号: 声卡编号: {}, 通道编号: {}", 
                        soundCardNum, roadNum);
    
    // for(auto it = m_mapSoundCardRoadPorts.begin(); it != m_mapSoundCardRoadPorts.end(); ++it)
    // {
    //     if(it.key().nSoundCardNum == soundCardNum && it.key().nRoadNum == roadNum)
    //     {
    //         // SPDLOG_LOGGER_DEBUG(m_logger, "从声卡通道列表中删除: 声卡编号: {}, 通道编号: {}, 本地端口: {}", 
    //         //                     soundCardNum, roadNum, localPort);
    //         m_mapSoundCardRoadPorts.remove(it.key());
    //         break;
    //     }
    // }
    
}

/* 处理心跳超时 */
void RTPServer::do_heartbeatTimeout()
{
    const QDateTime currentTime = QDateTime::currentDateTime();
    QList<RtpClientKey_t> keysToRemove;

    for (auto it = m_mapClientHeartbeats.begin(); it != m_mapClientHeartbeats.end(); ++it) 
    {
        if (it.value().secsTo(currentTime) > 20) // 超过30秒未收到心跳
        {
            SPDLOG_LOGGER_WARN(m_logger, "客户端 {}:{} 心跳超时，准备移除", 
                               it.key().clientIP.toStdString(), it.key().clientPort);
            keysToRemove.append(it.key());
        }
    }

    /* 注销超时的会话 */
    for (const auto& key : keysToRemove) 
    {
        RtpSendClientInfo_t clientInfo;
        clientInfo.clientIP = key.clientIP;
        clientInfo.clientPort = key.clientPort;
        clientInfo.SoundCardNum = key.SoundCardNum;
        clientInfo.SoundCardRoadNum = key.SoundCardRoadNum;
        handleLogout(clientInfo);
    }
}

/* 处理登录请求 */
void RTPServer::handleLogin(RtpSendClientInfo_t& clientInfo)
{
    SPDLOG_LOGGER_TRACE(m_logger, "处理登录请求: {}:{}", clientInfo.clientIP.toStdString(), clientInfo.clientPort);
    /* 查找本地可用的UDP端口，这里先使用10010测试 */
    // clientInfo.localPort = 10010;
    clientInfo.localIP = m_localIP;
    /* 查找RTP发送线程 */
    auto pThread = ThreadMan.getRtpSendThread(clientInfo.SoundCardNum, clientInfo.SoundCardRoadNum);
    if(pThread == nullptr)
    {
        SPDLOG_LOGGER_ERROR(m_logger, "没有找到对应的RTP发送线程，声卡编号: {}, 通道编号: {}", 
                            clientInfo.SoundCardNum, clientInfo.SoundCardRoadNum);
        return;
    }
    /* 设置客户端信息 */
    if(!pThread->addUdpSession(clientInfo))
    {
        return;
    }
    connect(pThread, &RTPOneRoadThread::signal_udpClosed, this, &RTPServer::do_udpClosed);
    /* 将本地端口添加到心跳列表中 */
    RtpClientKey_t clientKey;
    clientKey.clientIP = clientInfo.clientIP;
    clientKey.clientPort = clientInfo.clientPort;
    clientKey.SoundCardNum = clientInfo.SoundCardNum;
    clientKey.SoundCardRoadNum = clientInfo.SoundCardRoadNum;
    m_mapClientHeartbeats[clientKey] = QDateTime::currentDateTime();
}

/* 处理心跳请求 */
void RTPServer::handleHeartbeat(RtpSendClientInfo_t& clientInfo)
{
    SPDLOG_LOGGER_TRACE(m_logger, "处理心跳请求: {}:{}", clientInfo.clientIP.toStdString(), clientInfo.clientPort);
    RtpClientKey_t clientKey;
    clientKey.clientIP = clientInfo.clientIP;
    clientKey.clientPort = clientInfo.clientPort;
    clientKey.SoundCardNum = clientInfo.SoundCardNum;
    clientKey.SoundCardRoadNum = clientInfo.SoundCardRoadNum;
    /* 更新心跳时间 */
    if(m_mapClientHeartbeats.contains(clientKey))
    {
        m_mapClientHeartbeats[clientKey] = QDateTime::currentDateTime();
        SPDLOG_LOGGER_TRACE(m_logger, "更新心跳时间: {}:{}", clientInfo.clientIP.toStdString(), clientInfo.clientPort);
    }
}

/* 处理注销请求 */
void RTPServer::handleLogout(RtpSendClientInfo_t& clientInfo)
{
    SPDLOG_LOGGER_TRACE(m_logger, "处理注销请求: {}:{}", clientInfo.clientIP.toStdString(), clientInfo.clientPort);

    /* 查找RTP发送线程 */
    auto pThread = ThreadMan.getRtpSendThread(clientInfo.SoundCardNum, clientInfo.SoundCardRoadNum);
    if(pThread == nullptr)
    {
        SPDLOG_LOGGER_ERROR(m_logger, "没有找到对应的RTP发送线程，声卡编号: {}, 通道编号: {}", 
                            clientInfo.SoundCardNum, clientInfo.SoundCardRoadNum);
        return;
    }
    /* 删除UDP会话 */
    if(!pThread->removeUdpSession(clientInfo.clientIP, clientInfo.clientPort))
    {
        SPDLOG_LOGGER_ERROR(m_logger, "删除UDP会话失败: {}:{}", clientInfo.clientIP.toStdString(), clientInfo.clientPort);
        return;
    }

    /* 将客户端信息从心跳列表中移除 */
    RtpClientKey_t clientKey;
    clientKey.clientIP = clientInfo.clientIP;
    clientKey.clientPort = clientInfo.clientPort;
    clientKey.SoundCardNum = clientInfo.SoundCardNum;
    clientKey.SoundCardRoadNum = clientInfo.SoundCardRoadNum;
    if(m_mapClientHeartbeats.contains(clientKey))
    {
        m_mapClientHeartbeats.remove(clientKey);
        SPDLOG_LOGGER_DEBUG(m_logger, "从心跳列表中移除客户端: {}:{}", 
                            clientInfo.clientIP.toStdString(), clientInfo.clientPort);
    }
}



/* 查找本地出口IP，适用于单个出口IP，如果电脑上有多个网口，需要手动指定IP */
QString RTPServer::findLocalIP()
{
    QString localIP;
    QList<QHostAddress> addresses = QNetworkInterface::allAddresses();
    for (const QHostAddress& address : addresses) 
    {
        if (address.protocol() == QAbstractSocket::IPv4Protocol && !address.isLoopback()) 
        {
            if(address == QHostAddress::AnyIPv4)
            {
                continue; // 跳过
            }
            localIP = address.toString();
            SPDLOG_LOGGER_INFO(m_logger, "找到本地出口IP: {}", localIP.toStdString());
            return localIP;
        }
    }

    SPDLOG_LOGGER_ERROR(m_logger, "未找到有效的本地出口IP");
    return localIP;
}