server1/Server/ThreadManager/ThreadCompareItemManager.cpp

#include "ThreadCompareItemManager.h"

#include "GlobalInfo.h"
#include "SystemConfig.h"
#include "CompareItemThread.h"
#include "ThreadPool.h"
#include "FromMQTT.h"
#include "commonDefine.h"



/* 给对比项套一层壳，这个函数就是新的线程，在里面new出新的对比项实例，防止Qt报线程归属权错误
   在函数中将对比项实例插入到线程管理器中 */
void ThreadCompareItemManager::thread_compareItem(CalculateThreadInfo_t threadInfo)
{
    auto pThread = new CompareItemThread(threadInfo);
    if(pThread == nullptr)
    {
        SPDLOG_ERROR("ThreadCompareItemManager: 创建对比项线程失败");
        return;
    }
    CompareItemManager.addCompareItemThread(pThread);
    
    /* 启动线程，就会一直阻塞在这里了 */
    pThread->thread_task();
   
}




ThreadCompareItemManager::ThreadCompareItemManager()
{
    
}

ThreadCompareItemManager::~ThreadCompareItemManager()
{

}

/* 线程函数 */
void ThreadCompareItemManager::thread_CompareItemManager()
{
    m_logger = spdlog::get("CompareItemManager");
    if(m_logger == nullptr)
    {
        fmt::print("ThreadCompareItemManager: CompareItemManager Logger not found.\n");
        return;
    }

    /* 创建定时器和事件循环 */
    m_pEventLoop = new QEventLoop();
    m_pTimer = new QTimer();

    // m_pTimer->setInterval(10000);
    m_pTimer->setTimerType(Qt::PreciseTimer);
    m_pTimer->setSingleShot(false); // 设置为非单次定时器
    /* 这里要设置直接连接，才会在子线程中调用槽函数 */
    connect(m_pTimer, &QTimer::timeout, this, &ThreadCompareItemManager::do_task, Qt::DirectConnection);
    
    /* 初始化webapi */
    m_webAPIUrl = GInfo.webAPIUrl();
    m_webAPIID = GInfo.webAPIID();
    m_webAPIAppType = GInfo.webApiAppType();
    if(!m_fromWebAPI.initWebApi(m_webAPIUrl, m_webAPIID, m_webAPIAppType))
    {
        SPDLOG_LOGGER_ERROR(m_logger, "ThreadCompareItemManager: 初始化WebAPI失败");
        return;
    }
    /* 获取MQTT发布订阅 */
    m_pubTopic = GInfo.mqttPubTopicCompareItem();
    initMQTT();

    /* 获取基础配置，目前只获取一次 */
    updateBaseSettings();

    SPDLOG_LOGGER_INFO(m_logger, "开启对比项管理线程");

    m_pTimer->start(2000);
    /* 连接mqtt服务 */
    m_pFromMQTT->connectToServer();
    m_pEventLoop->exec();

    SPDLOG_LOGGER_INFO(m_logger, "ThreadCompareItemManager: 线程结束");

}



/* 通过对比项ID和通道ID获取声卡通道信息 */
OneSoundCardPCMInfo_t ThreadCompareItemManager::getSoundCardRoadInfo(int compareItemID, int roadNum)
{
    std::lock_guard<std::mutex> lock(m_mutexCompareItemThreads);
    auto it  = m_mapThreads.find(compareItemID);
    if(it == m_mapThreads.end())
    {
        SPDLOG_LOGGER_WARN(m_logger, "对比项线程不存在，ID: {}", compareItemID);
        return OneSoundCardPCMInfo_t();
    }
    auto compareInfo = it.value()->getThreadInfo().compareItemInfo;
    OneSoundCardPCMInfo_t roadInfo;
    for(const auto& road : compareInfo.mapRoad)
    {
        if(road.nCompareRoadNum == roadNum)
        {
            roadInfo = road.scRoadInfo;
            break;
        }
    }
    return roadInfo;
}

/* 任务函数 */
void ThreadCompareItemManager::do_task()
{
    /* 如果定时间隔小于10秒，则设置成10秒，一开始小是为了线程开启后立马执行一次 */
    if(m_pTimer->interval() < 2000)
    {
        m_pTimer->setInterval(2000);
    }
    /* 判断MQTT是否连接成功，未连接则再次连接 */
    if(m_pFromMQTT->connectState() != QMQTT::ConnectionState::STATE_CONNECTED)
    {
        SPDLOG_LOGGER_WARN(m_logger, "MQTT未连接，尝试重新连接");
        m_pFromMQTT->connectToServer();
    }
    /* ------------------------------------------------------------------
     * 处理对比项信息
     * ------------------------------------------------------------------ */
    processCompareItemInfo();
    
    /* ------------------------------------------------------------------
     * 更新检测时段
     * ------------------------------------------------------------------ */
    processDetectPeriodInfo();

    /* ------------------------------------------------------------------
     * 更新对比项信息到MQTT
     * ------------------------------------------------------------------ */
    updateCompareItemInfoToMQTT();
}


/* 添加对比项实例 */
void ThreadCompareItemManager::addCompareItemThread(CompareItemThread* pThread)
{
    if(pThread == nullptr)
    {
        SPDLOG_LOGGER_ERROR(m_logger, "添加对比项线程失败，线程指针为空");
        return;
    }
    
    std::lock_guard<std::mutex> lock(m_mutexCompareItemThreads);
    int compareItemID = pThread->getThreadInfo().compareItemInfo.nID;
    if(m_mapThreads.contains(compareItemID))
    {
        SPDLOG_LOGGER_WARN(m_logger, "对比项线程已存在，ID: {}", compareItemID);
        return; // 对比项线程已存在
    }
    
    m_mapThreads.insert(compareItemID, pThread);
    SPDLOG_LOGGER_INFO(m_logger, "添加对比项线程成功，ID: {}", compareItemID);
}

/* 更新基础设置信息，如数据库设置，噪音参数等 */
bool ThreadCompareItemManager::updateBaseSettings()
{
    /* 更新基础数据 */
    QMap<std::string, std::string> baseSettings;
    if(!m_fromWebAPI.getSystemConfig(baseSettings))
    {
        SPDLOG_LOGGER_ERROR(m_logger, "获取系统配置失败");
        return false;
    }
    /* 将获取到的配置转换成结构体 */
    SysConfig.parseConfigFromDatabase(baseSettings);

    /* 检测时段单独获取 */
    QMap<int, DetectPeriodConfig_t> mapDetectConfig;
    if(!m_fromWebAPI.getDetectPeriodConfig(mapDetectConfig))
    {
        SPDLOG_ERROR("获取对比项检测时段配置失败");
        return false;
    }
    SysConfig.setDetectPeriodConfig(mapDetectConfig);

    return true;
}

/* 对比项信息处理函数 */
void ThreadCompareItemManager::processCompareItemInfo()
{
    /* 获取对比项信息 */
    QList<CompareItemInfo_t> listNewItems;
    if(!m_fromWebAPI.getCompareItemInfo(listNewItems))
    {
        SPDLOG_LOGGER_DEBUG(m_logger, "ThreadCompareItemManager: 获取对比项失败");
        return;
    }

    checkCompareItemInfo(listNewItems, m_listCreateItems, m_listDeleteItems);

    SPDLOG_LOGGER_DEBUG(m_logger, "要退出的对比项个数: {}, 要创建的对比项个数: {}",
        m_listDeleteItems.size(), m_listCreateItems.size());

    /* 先删除已消失的对比项信息 */
    processDeleteCompareItemThreads(m_listDeleteItems);
    /* 更新需要更新的线程 */
    // updateRunningThreads(m_listUpdateItems);
    /* 再创建新的对比项线程 */
    createNewCompareItemThreads(m_listCreateItems);

    SPDLOG_LOGGER_DEBUG(m_logger, "当前对比项个数: {}", m_mapNowCompareItem.size());
}


/**
 * @brief 处理对比项信息，新获取的和已有的对比，会在这里更新 m_mapNowCompareItem 内容
 * 
 * @param createList 创建列表
 * @param updateList 更新列表，根据对比项ID进行更新信息，没有更新列表了，对比项有更新就删除重建
 * @param deleteList 删除列表
 */
void ThreadCompareItemManager::checkCompareItemInfo(QList<CompareItemInfo_t>& newList, QList<CompareItemInfo_t>& createList, QList<int>& deleteList)
{
    createList.clear();
    // updateList.clear();
    deleteList.clear();

    m_mapNowCompareItem.clear();

    // QMap<int, CompareItemInfo_t> mapNowItems;
    /* 先从对比项线程中获取对比项信息 */
    for(auto it = m_mapThreads.begin(); it != m_mapThreads.end(); ++it)
    {
        BaseCalculateThread* pThread = it.value();
        if(pThread == nullptr)
        {
            continue;
        }
        
        /* 获取对比项信息 */
        CompareItemInfo_t itemInfo = pThread->getThreadInfo().compareItemInfo;
        m_mapNowCompareItem.insert(itemInfo.nID, itemInfo);
    }

    /* 遍历新获取的对比项信息，找出需要新增的对比项和需要更新的对比项 */
    for(const CompareItemInfo_t& item : newList)
    {
        if(!m_mapNowCompareItem.contains(item.nID))
        {
            /* 判断这个是否被启用 */
            if(item.isEnable == false)
            {
                continue;
            }
            /* 新对比项，添加到创建列表 */
            createList.append(item);
        } else
        {
            /* 已有对比项，检查是否需要更新 */
            const CompareItemInfo_t& existingItem = m_mapNowCompareItem.value(item.nID);
            /* 先对比基础信息 */
            if(!existingItem.isEqualBase(item))
            {
                /* 基础信息不同，需要更新，也可能是被禁用了导致的不同 */
                if(item.isEnable == false)
                {
                    /* 如果新对比项被禁用，则需要删除 */
                    deleteList.append(item.nID);
                    continue;
                }
                // updateList.append(item);
                deleteList.append(item.nID); // 先删除再创建
                createList.append(item); // 添加到创建列表
                continue;
            }
            /* 在对比对比项通道信息 */
            if(!existingItem.isEqualRoads(item))
            {
                /* 通道信息不同，需要更新 */
                // updateList.append(item);
                deleteList.append(item.nID); // 先删除再创建
                createList.append(item); // 添加到创建列表
                continue;
            }

        }
    }
    /* 遍历当前对比项信息，找出需要删除的对比项 */
    for(auto it : m_mapNowCompareItem)
    {
        bool isFound = false;
        for(const CompareItemInfo_t& newItem : newList)
        {
            if(it.nID == newItem.nID)
            {
                /* 找到对应的对比项，跳过 */
                isFound = true;
                break; 
            }
        }
        if(!isFound)
        {
            /* 当前对比项不在新获取的对比项中，说明需要删除 */
            deleteList.append(it.nID);
        }
        
    }


}


/**
 * @brief 处理需要删除的对比项线程
 *        1、先处理已经停止的线程，初始化失败的对比项会设置错误状态
 *        2、再将这次列表中的对比项ID对应的线程设置为停止状态，待到下次循环再删除已经停止完成的线程
 * 
 * @param deleteList 
 */
void ThreadCompareItemManager::processDeleteCompareItemThreads(const QList<int>& deleteList)
{
    /* 先处理已经停止运行的线程，是上次循环停止的线程 */
    for(auto it = m_mapThreads.begin(); it != m_mapThreads.end();)
    {
        BaseCalculateThread* pThread = it.value();
        if(pThread == nullptr)
        {
            SPDLOG_LOGGER_WARN(m_logger, "对比项线程指针为空，即将删除该线程指针");
            it = m_mapThreads.erase(it);
            continue;
        }
        if(pThread->getThreadInfo().threadState == EThreadState::State_Stopped ||
            pThread->getThreadInfo().threadState == EThreadState::State_Error )
        {
            /* 确保线程已经停止 */
            pThread->thread_stop_block();
            SPDLOG_LOGGER_INFO(m_logger, "对比项线程 {} 已经停止，准备删除", pThread->getThreadInfo().compareItemInfo.strName.toStdString());
            delete pThread;
            it = m_mapThreads.erase(it);
            continue;
        }
        ++it;

    }

    /* 停止本次需要停止的线程 */
    for(auto it = m_mapThreads.begin(); it != m_mapThreads.end(); )
    {
        int compareItemID = it.value()->getThreadInfo().compareItemInfo.nID;
        if(deleteList.contains(compareItemID))
        {
            std::string compareItemName = it.value()->getThreadInfo().compareItemInfo.strName.toStdString();
            /* 设置线程停止标志，阻塞等待线程停止，然后再删除 */
            it.value()->thread_stop_block();
            delete it.value();
            it.value() = nullptr; // 设置为nullptr，防止悬空指针
            it = m_mapThreads.erase(it); // 从列表中删除
            SPDLOG_LOGGER_INFO(m_logger, "对比项线程 {} 已删除", compareItemName);
        }else {
            ++it;
        }
    }
}

/* 更新正在运行的线程信息 */
void ThreadCompareItemManager::updateRunningThreads(const QList<CompareItemInfo_t>& updateList)
{
    if(updateList.isEmpty())
    {
        return;
    }
    for(const CompareItemInfo_t& item : updateList)
    {
        auto it = m_mapThreads.find(item.nID);
        if(it == m_mapThreads.end())
        {
            SPDLOG_LOGGER_WARN(m_logger, "对比项线程 {} 不存在，无法更新信息", item.strName.toStdString());
            continue;
        }
        BaseCalculateThread* pThread = it.value();
        if(pThread == nullptr)
        {
            continue;
        }
        CalculateThreadInfo_t threadInfo;
        threadInfo.compareItemInfo = item;
        pThread->updateThreadInfo(threadInfo);
    }
}

/* 创建新的线程 */
bool ThreadCompareItemManager::createNewCompareItemThreads(const QList<CompareItemInfo_t>& createList)
{
    if(createList.isEmpty())
    {
        // SPDLOG_LOGGER_DEBUG(m_logger, "没有新的对比项需要创建");
        return true;
    }

    for(auto& it : createList)
    {
        /* 创建新的对比项线程 */
        CalculateThreadInfo_t threadInfo;
        threadInfo.compareItemInfo = it;
        threadInfo.threadType = EThreadType::Type_CompareItem;
        threadInfo.threadState.store(EThreadState::State_Inited);

        CPPTP.add_task(&ThreadCompareItemManager::thread_compareItem, threadInfo);

        /* 创建线程对象 */
        // CompareItemThread* pThread = new CompareItemThread(threadInfo);
        // if(pThread == nullptr)
        // {
        //     SPDLOG_LOGGER_ERROR(m_logger, "创建对比项线程 {} 失败", it.strName.toStdString());
        //     return false;
        // }
        
        /* 启动线程 */
        // CPPTP.add_task(&CompareItemThread::threadTask, pThread);
        
        /* 添加到线程列表中 */
        // m_mapThreads.insert(it.nID, pThread);
    }

    return true;
}


/* 处理检测时段信息 */
void ThreadCompareItemManager::processDetectPeriodInfo()
{
    /* 获取计划信息 */
    QMap<int, DetectPeriodConfig_t> mapNewDetectConfig;
    if(!m_fromWebAPI.getDetectPeriodConfig(mapNewDetectConfig))
    {
        SPDLOG_LOGGER_ERROR(m_logger, "获取检测时段配置失败");
        return;
    }

    QMap<int, DetectPeriodConfig_t> mapUpdateDetectConfig;
    checkDetectPeriodInfo(m_mapDetectPeriod, mapNewDetectConfig, mapUpdateDetectConfig);

    /* 更新检测时段 */
    for(const auto& it : mapUpdateDetectConfig)
    {
        auto threadIt = m_mapThreads.find(it.nID);
        if(threadIt != m_mapThreads.end())
        {
            /* 找到对应的对比项线程，更新检测时段 */
            CompareItemThread* pThread = dynamic_cast<CompareItemThread*>(threadIt.value());
            if(pThread != nullptr)
            {
                pThread->setDetectPeriod(it);
                SPDLOG_LOGGER_TRACE(m_logger, "更新对比项 {} 的检测时段", pThread->getThreadInfo().compareItemInfo.strName.toStdString());
            }
        }
    }
    /* 更新当前检测时段信息 */
    m_mapDetectPeriod = mapNewDetectConfig;
}

/* 检查获取出更新的对比项信息 */
void ThreadCompareItemManager::checkDetectPeriodInfo(QMap<int, DetectPeriodConfig_t>& nowDetectInfo, QMap<int, DetectPeriodConfig_t> newDetectInfo, QMap<int, DetectPeriodConfig_t>& updateList)
{
    for(const auto& it : newDetectInfo)
    {
        int compareItemID = it.nID;
        bool isFound = false;
        for(const auto& existingItem : nowDetectInfo)
        {
            if(existingItem.nID == compareItemID)
            {
                isFound = true;
                /* 已经存在的对比项，检查是否需要更新 */
                if(existingItem == it)
                {
                    /* 对比项信息相同，不需要更新 */
                    continue;
                } else
                {
                    /* 对比项信息不同，需要更新 */
                    updateList.insert(compareItemID, it);
                }
                return;
            }
        }
        if(isFound == false)
        {
            /* 新对比项，添加到更新列表 */
            updateList.insert(compareItemID, it);
        }
    }
}

/* 更新对比项信息到MQTT */
void ThreadCompareItemManager::updateCompareItemInfoToMQTT()
{
    QMap<int, CompareItemInfo_t> newMap;
    /* 获取当前的对比项信息 */
    for(auto it = m_mapThreads.begin(); it != m_mapThreads.end(); ++it)
    {
        BaseCalculateThread* pThread = it.value();
        if(pThread == nullptr)
        {
            continue;
        }
        
        /* 获取对比项信息 */
        CompareItemInfo_t itemInfo = pThread->getThreadInfo().compareItemInfo;
        newMap.insert(itemInfo.nID, itemInfo);
    }

    /* 和之前的对比，对比项是否有更新，有更新则发送 */
    bool isUpdated = false;
    if(newMap.size() != m_mapMQTTItemInfo.size())
    {
        isUpdated = true;
    } else
    {
        /* 进一步对比相信信息 */
        for(const auto& newItem : newMap)
        {
            auto nowIt = m_mapMQTTItemInfo.find(newItem.nID);
            if(nowIt == m_mapMQTTItemInfo.end())
            {
                isUpdated = true;
                break;
            }
            /* 比较对比项通道数量和对比项名称 */
            if(nowIt.value().strName != newItem.strName || nowIt.value().mapRoad.size() != newItem.mapRoad.size())
            {
                isUpdated = true;
                break;
            }
            /* 挨个比较对比项通道信息 */
            for(const auto& road : newItem.mapRoad)
            {
                auto roadIt = nowIt.value().mapRoad.find(road.nCompareRoadNum);
                if(roadIt == nowIt.value().mapRoad.end() || roadIt.value().strCompareRoadName != road.strCompareRoadName)
                {
                    isUpdated = true;
                    break;
                }
            }
        }
    }

    /* 服务启动，可能对比项是空的，上面检查结果为false，但是也需要发送一次清空对比项 */
    if(isUpdated || m_isOneSend.load())
    {
        sendCompareItemInfoToMQTT(newMap);
        m_mapMQTTItemInfo = newMap;
    }
    m_isOneSend.store(false);
}

/* 发送到MQTT */
void ThreadCompareItemManager::sendCompareItemInfoToMQTT(const QMap<int, CompareItemInfo_t>& mapCompareItem)
{
    /* 生成发送的json文件 */
    nJson jsonArray = nJson::array();
    for(const auto& it : mapCompareItem)
    {
        /* 判断这个对比项是否启用，不启用就跳过 */
        if(it.isEnable == false)
        {
            continue;
        }

        nJson jsonItem;
        /* 对比项ID */
        jsonItem["compareItem_id"] = it.nID;
        /* 对比项名称 */
        jsonItem["compareItem_name"] = it.strName.toStdString();
        nJson josnItemRoads = nJson::array();
        for(const auto& road : it.mapRoad)
        {
            nJson jsonRoad;
            /* 对比项通道编号和名称 */
            jsonRoad["road_num"] = road.nCompareRoadNum;
            jsonRoad["road_name"] = road.strCompareRoadName.toStdString();
            /* 通道使用的声卡编号 */
            // jsonRoad["sound_card_num"] = road.scRoadInfo.nSoundCardNum;
            // jsonRoad["sound_card_road_num"] = road.scRoadInfo.roadInfo.nRoadNum;

            josnItemRoads.push_back(jsonRoad);
        }
        jsonItem["compareItem_roads"] = josnItemRoads;
        // /* 静音条件 */
        // jsonItem["silence_switch"] = it.paramMute.isEnable;
        // jsonItem["silence_threshold"] = it.paramMute.threshold.nThreshold;
        // jsonItem["silence_duration"] = it.paramMute.nLen;
        // jsonItem["silence_sensitivity"] = it.paramMute.nSensitivity;
        // /* 过载条件 */
        // jsonItem["overload_switch"] = it.paramOverload.isEnable;
        // jsonItem["overload_threshold"] = it.paramOverload.threshold.nThreshold;
        // jsonItem["overload_duration"] = it.paramOverload.nLen;
        // jsonItem["overload_sensitivity"] = it.paramOverload.nSensitivity;
        // /* 反相条件 */
        // jsonItem["reverse_switch"] = it.paramPhase.isEnable;
        // jsonItem["reverse_threshold"] = it.paramPhase.threshold.dThreshold;
        // jsonItem["reverse_duration"] = it.paramPhase.nLen;
        // jsonItem["reverse_sensitivity"] = it.paramPhase.nSensitivity;

        jsonArray.push_back(jsonItem);
    }

    /* 发送到MQTT */
    if(m_pFromMQTT->connectState() == QMQTT::ConnectionState::STATE_CONNECTED)
    {
        QByteArray jsonData = QByteArray::fromStdString(jsonArray.dump());
        if(!m_pFromMQTT->sendMessage(m_pubTopic, jsonData, 0, true))
        {
            SPDLOG_LOGGER_WARN(m_logger, "ThreadCompareItemManager: 发送对比项信息到MQTT失败");
        }else
        {
            SPDLOG_LOGGER_TRACE(m_logger, "ThreadCompareItemManager: 发送对比项信息到MQTT成功");
        }
    } else
    {
        SPDLOG_LOGGER_WARN(m_logger, "m_pFromMQTT 未连接到服务器，无法发送");
    }
}

/* 初始化MQTT */
void ThreadCompareItemManager::initMQTT()
{
    if(m_pFromMQTT == nullptr)
    {
        m_pFromMQTT = new FromMQTT();
        if(m_pFromMQTT == nullptr)
        {
            SPDLOG_LOGGER_ERROR(m_logger, "创建MQTT对象失败");
            return;
        }
    }
    /* 登陆MQTT */
    m_pFromMQTT->setIPAndPort(GInfo.mqttIP(), GInfo.mqttPort());
    // m_pFromMQTT->addSubcribe("LH_WEBINFO");
    m_pFromMQTT->setAutoReconnect(true);
    // m_pFromMQTT->connectToServer();

    // connect(m_pFromMQTT, &FromMQTT::signal_recvMessage, [this](const QMQTT::Message& message) {
    //     SPDLOG_LOGGER_WARN(m_logger, "--------------------- 接收到MQTT消息: {}", message.topic().toStdString());
    //     SPDLOG_LOGGER_WARN(m_logger, "消息内容: {}", message.payload().toStdString());
    // });

    SPDLOG_LOGGER_INFO(m_logger, "☆ 连接MQTT服务器: {}:{}, 对比项信息订阅主题: {}", GInfo.mqttIP().toStdString(), GInfo.mqttPort(), m_pubTopic.toStdString());
}