server1/RTPServer/RtpOneRoadThread.cpp

#include "RtpOneRoadThread.h"
#include "spdlog.h"





RTPOneRoadThread::RTPOneRoadThread(RecordThreadInfo_t& threadInfo)
    : BaseRecordThread(threadInfo),
      m_eventLoop()
{
    m_logger_local = spdlog::get("RTPServer");
    if(m_logger_local == nullptr)
    {
        fmt::print("RTPServer 日志记录器未初始化，请先初始化日志记录器");
        return;
    }
    m_logBase = fmt::format("录音通道: {}:{}", threadInfo.cardRoadInfo.strSoundCardName, threadInfo.cardRoadInfo.pcmInfo.strPCMName);
    
}

RTPOneRoadThread::~RTPOneRoadThread()
{

}


/**
 * @brief 停止线程
 */
void RTPOneRoadThread::thread_stop()
{
    m_isRunning.store(false);
    /* 等待事件循环停止 */
    while(m_isStoped.load())
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}

/* 设置数据 */
bool RTPOneRoadThread::setData(const AudioSrcData& srcData)
{
    /* UDP会话列表是空的，就不接收数据了 */
    if(!m_isRecvData.load())
    {
        return true;
    }

    AudioSrcData* pData = new AudioSrcData(srcData);
    if(pData == nullptr)
    {
        return false;
    }

    auto oldData = m_ringQueue.push_pop(pData);
    if(oldData != nullptr)
    {
        SPDLOG_LOGGER_ERROR(m_logger_local, "{} 环形队列已满，无法添加新的音频数据", m_logBase);
        delete oldData; // 删除被覆盖的数据
        oldData = nullptr;
    }

    return false;
}

/* 添加一个UDP会话 */
bool RTPOneRoadThread::addUdpSession(const RtpSendClientInfo_t& udpSession)
{
    /* 检查是否已有相同的会话，这里只需要检查IP和端口即可，会话ID和名称在这里没什么用 */
    for(const auto& session : m_listClients)
    {
        if(session.clientIP == udpSession.clientIP && session.clientPort == udpSession.clientPort)
        {
            SPDLOG_LOGGER_DEBUG(m_logger_local, "{} 已存在相同的UDP会话: {}:{}", m_logBase, udpSession.clientIP.toStdString(), udpSession.clientPort);
            return false;
        }
    }
    SPDLOG_LOGGER_INFO(m_logger_local, "{} 添加UDP会话: {}:{}", m_logBase, udpSession.clientIP.toStdString(), udpSession.clientPort);

    /* 绑定UDP本地端口 */
    if(m_udpState == eUDPState::eUDP_None)
    {
        m_udpState = eUDPState::eUDP_Init;
        m_localIP = udpSession.localIP;
    }
    
    std::lock_guard<std::mutex> lock(m_lockClients);
    m_listClients.append(udpSession);

    /* 触发一次定时器槽函数，执行处理UDP状态的函数 */
    emit signal_timerSendData();
    
    return true;
}

/* 删除一个会话 */
bool RTPOneRoadThread::removeUdpSession(QString clientIP, quint16 clientPort)
{
    std::lock_guard<std::mutex> lock(m_lockClients);
    for(auto it = m_listClients.begin(); it != m_listClients.end(); ++it)
    {
        if(it->clientIP == clientIP && it->clientPort == clientPort)
        {
            // if(it->udpSocket != nullptr)
            // {
            //     it->udpSocket->close();
            //     delete it->udpSocket;
            //     it->udpSocket = nullptr;
            // }
            SPDLOG_LOGGER_INFO(m_logger_local, "{} 删除UDP会话: {}:{}", m_logBase, it->clientIP.toStdString(), it->clientPort);
            // 从列表中删除该会话
            m_listClients.erase(it);
            break;
        }
    }
    /* 如果列表为空，就不再接收数据 */
    if(m_listClients.isEmpty())
    {
        /* 清空环形队列 */
        while(m_ringQueue.QueueSize() > 0)
        {
            auto data = m_ringQueue.front_pop_noBlock();
            if(data != nullptr)
            {
                delete data;
                data = nullptr;
            }
        }
        m_udpState = eUDPState::eUDP_Closed;
        
        SPDLOG_LOGGER_DEBUG(m_logger_local, "{} UDP会话列表为空，不再接收音频原始数据", m_logBase);

        /* 触发一次定时器槽函数，执行处理UDP状态的函数 */
        emit signal_timerSendData();
    }

    return true;
}

/* 发送数据的线程函数 */
void RTPOneRoadThread::task()
{
    SPDLOG_LOGGER_INFO(m_logger_local, "➢ {} 开启RTP发送数据线程 ", m_logBase);
    if(!initData())
    {
        SPDLOG_LOGGER_ERROR(m_logger_local, "{} 初始化数据失败", m_logBase);
        return;
    }

    m_sendTimer.setTimerType(Qt::PreciseTimer);
    m_sendTimer.setSingleShot(false);
    m_sendTimer.setInterval(25);
    QEventLoop::connect(&m_sendTimer, &QTimer::timeout, this, &RTPOneRoadThread::do_timerSendData);
    connect(this, &RTPOneRoadThread::signal_timerSendData, this, &RTPOneRoadThread::do_timerSendData);
    m_sendTimer.start();

    m_isRunning.store(true); // 设置为运行状态
    m_isStoped.store(false);
    /* 将线程控制权交给Qt的事件循环 */
    m_eventLoop.exec();
    /* 清空数据 */
    clearData();
    m_isRunning.store(false);
    m_isStoped.store(true);
    SPDLOG_LOGGER_WARN(m_logger_local, "➢ {} RTP发送数据线程结束 ", m_logBase);
}


/* 初始化数据 */
bool RTPOneRoadThread::initData()
{
    m_listClients.clear();
    m_isRecvData.store(false);

    m_ringQueue.setQueueCapacity(60);
    m_ringQueue.setDefaultValue(nullptr);

    return true;
}

/* 清除数据*/
void RTPOneRoadThread::clearData()
{
    m_isRecvData.store(false); // 设置为不接收数据状态
    /* 清空UDP会话列表 */
    std::lock_guard<std::mutex> lock(m_lockClients);
    m_listClients.clear();
    /* 关闭UDP占用的本地端口 */
    if(m_udpSocket != nullptr)
    {
        m_udpSocket->close();
        m_udpSocket->deleteLater();
    }
    m_udpState = eUDPState::eUDP_None;
    
    /* 清空环形队列 */
    while(m_ringQueue.QueueSize() > 0)
    {
        auto data = m_ringQueue.front_pop_noBlock();
        if(data != nullptr)
        {
            delete data;
            data = nullptr;
        }
    }
}


/* 处理UDP状态 */
bool RTPOneRoadThread::processUdpState()
{
    if(m_udpSocket == nullptr)
    {
        m_udpSocket = new QUdpSocket();
        if(m_udpSocket == nullptr)
        {
            SPDLOG_LOGGER_ERROR(m_logger_local, "{} 创建UDP套接字失败", m_logBase);
            return false;
        }
        m_udpSocket->setSocketOption(QAbstractSocket::LowDelayOption, 1); // 设置低延迟选项
        connect(m_udpSocket, SIGNAL(error(QAbstractSocket::SocketError)), this, SLOT(do_udpError(QAbstractSocket::SocketError)));
    }

    if(m_udpState == eUDPState::eUDP_Init)
    {
        if(!m_udpSocket->bind(QHostAddress(m_localIP)))
        {
            SPDLOG_LOGGER_ERROR(m_logger_local, "{} 绑定UDP套接字失败: {}:{}", m_logBase, m_localIP.toStdString(), m_udpSocket->localPort());
            return false;
        }
        SPDLOG_LOGGER_INFO(m_logger_local, "{} 绑定UDP套接字到本地IP: {}:{}", m_logBase, 
                            m_udpSocket->localAddress().toString().toStdString(), m_udpSocket->localPort());
        /* 设置为接收数据状态 */
        m_isRecvData.store(true);
        m_udpState = eUDPState::eUDP_Opened;
    }
    else if(m_udpState == eUDPState::eUDP_Closed || m_udpState == eUDPState::eUDP_Error)
    {
        /* 关闭接收数据 */
        m_isRecvData.store(false);
        /* 关闭UDP占用的本地端口 */
        if(m_udpSocket != nullptr)
        {
            m_udpSocket->close();
        }
        m_udpState = eUDPState::eUDP_None;
        /* 清空连接信息 */
        m_listClients.clear();
        m_localIP.clear();
        // m_localPort = -1;
        emit signal_udpClosed(m_threadInfo.cardRoadInfo.pcmInfo.strPCMName);
        /* 清空环形队列中的数据 */
        while(m_ringQueue.QueueSize() > 0)
        {
            auto data = m_ringQueue.front_pop_noBlock();
            if(data != nullptr)
            {
                delete data;
                data = nullptr;
            }
        }
    }

    return true;
}


/**
 * @brief 发送数据，先获取一个缓冲区的数据，获取到后就遍历所有的UDP会话，将这个缓冲区所有的数据都发送出去
 *        然后紧接着获取第二个缓冲区的数据，也一起发送出去
 * 
 * @return true 
 * @return false 
 */
bool RTPOneRoadThread::sendData()
{
    while(m_ringQueue.QueueSize() > 0)
    {
        // SPDLOG_LOGGER_TRACE(m_logger_local, "{} 发送音频数据", m_logBase);
        auto pData = m_ringQueue.front_pop_noBlock();
        if(pData == nullptr)
        {
            continue;
        }
        if(pData->isEmpty())
        {
            delete pData;
            continue;
        }
        std::lock_guard<std::mutex> lock(m_lockClients);
        /* 遍历所有的UDP会话，发送数据 */
        for(const auto& session : m_listClients)
        {

            // qint64 bytesSent = session.udpSocket->writeDatagram(data->pData, data->dataSize,
                                                                // QHostAddress(session.clientIP), session.clientPort);
            qint64 bytesSent = m_udpSocket->writeDatagram(QByteArray(reinterpret_cast<const char*>(pData->pData), pData->dataSize),
                                                                QHostAddress(session.clientIP), session.clientPort);
            if(bytesSent == -1)
            {
                SPDLOG_LOGGER_ERROR(m_logger_local, "{} 发送数据失败: {}:{}", m_logBase, session.clientIP.toStdString(), session.clientPort);
            } else
            {
            //     SPDLOG_LOGGER_TRACE(m_logger_local, "{} 发送数据成功: {}:{}, 本地IP: {}:{}, 大小: {}", m_logBase, 
            //             session.clientIP.toStdString(), session.clientPort, 
            //             m_udpSocket->localAddress().toString().toStdString(), m_udpSocket->localPort(), bytesSent);
            }
        }
        /* 发送完数据后，删除这个数据 */
        delete pData;
        pData = nullptr;
    }

    return true;
}

/* 停止线程 */
void RTPOneRoadThread::stop_thread()
{
    if(m_sendTimer.isActive())
    {
        m_sendTimer.stop();
    }
    if(m_eventLoop.isRunning())
    {
        m_eventLoop.quit();
        m_eventLoop.processEvents(QEventLoop::AllEvents, 10);
    }
}

/* 定时发送数据 */
void RTPOneRoadThread::do_timerSendData()
{
    if(m_isRunning.load() == false)
    {
        stop_thread();
    }
    if(m_udpState == eUDPState::eUDP_None)
    {
        return;
    }
    
    processUdpState();
    
    
    /* 发送数据 */
    sendData();
}

/* UDP错误槽函数 */
void RTPOneRoadThread::do_udpError(QAbstractSocket::SocketError socketError)
{
    auto senderSocket = qobject_cast<QUdpSocket*>(sender());
    // SPDLOG_LOGGER_ERROR(m_logger_local, "{} UDP套接字错误: {}", m_logBase, static_cast<int>(socketError));
    /* 这里可以根据需要处理不同的错误 */
    if(socketError == QAbstractSocket::SocketError::SocketTimeoutError)
    {
        SPDLOG_LOGGER_WARN(m_logger_local, "{} UDP连接超时", m_logBase);
    } else
    {
        SPDLOG_LOGGER_ERROR(m_logger_local, "{} UDP套接字发生错误: {}", m_logBase, static_cast<int>(socketError));
        SPDLOG_LOGGER_ERROR(m_logger_local, "{} 错误信息: {}", m_logBase, senderSocket->errorString().toStdString());
    }
    
    m_udpState = eUDPState::eUDP_Error;
}