lp-data-alarm/EngineAlarm/DataEngine.cpp

#include "DataEngine.h"
#include "DeviceManagerProxy.h"
#include <QJsonArray>
#include <QJsonObject>
#include <QJsonDocument>
#include <fstream>
#include <QtCore/QVariant>
#include <QtCore/QDebug>
#include <QtWidgets/QApplication>

extern std::string loadContent(std::string szFilename);

DataEngine::DataEngine()
    :QObject()
{
    m_pRedis = nullptr;
    m_pTDengine = nullptr;
    m_mpDevControler.clear();
}

void DataEngine::OnData(std::string name,QVariant val)
{
//    if( m_pRedis != nullptr )
//    {
//        m_pRedis->Publish(name, val);
//    }
//    else
//    {
//        qCritical() << __FILE__ << __LINE__ << " " << "no publisher.";
//    }
    if( m_pTDengine != nullptr )
    {
 //       m_pTDengine->Publish(name, val);
    }
    else
    {
        qCritical() << __FILE__ << __LINE__ << " " << "no publisher.";
    }
    if( m_pMqtt != nullptr )
    {
        m_pMqtt->Publish(name, val);
    }
    else
    {
        qCritical() << __FILE__ << __LINE__ << " " << "no publisher.";
    }
}

void DataEngine::sltSubData_td(const std::string topic, const std::string msg)
{
    if(m_mpDevControler.find(topic) == m_mpDevControler.end())
    {
        return;
    }

    std::list<DeviceController*>::iterator itr;
    for( itr = m_mpDevControler[topic].begin(); itr != m_mpDevControler[topic].end(); ++itr)
    {
        DeviceController* pControler = *itr;
        if( pControler != nullptr )
        {
            pControler->OnSubData(topic,msg);
        }
    }
}

void DataEngine::sltSubData_mqtt(const std::string topic, const std::string msg)
{
}

void loadConfig(std::string szFile,Config& config)
{
    std::string cfg = loadContent(szFile);
    QJsonDocument jsonDoc(QJsonDocument::fromJson(cfg.c_str()));
    QJsonObject json = jsonDoc.object();
    QStringList keys = json.keys();
    if( keys.contains("serverid",Qt::CaseInsensitive))
    {
        config.serverId = json["serverid"].toString().toLocal8Bit().toStdString();
    }
    if( keys.contains("module",Qt::CaseInsensitive))
    {
        config.appName = json["module"].toString().toLocal8Bit().toStdString();
    }
}

void DataEngine::startup_DataAlarm(DeviceInfo di)
{
    if( di.ModuleInfo.vSettings.begin() == di.ModuleInfo.vSettings.end() )
    {
        return;
    }

    std::list<std::string> lstTopics;
    std::vector<Setting>::iterator itr;
    for( itr = di.ModuleInfo.vSettings.begin(); itr != di.ModuleInfo.vSettings.end(); ++itr )
    {
        std::string table = itr->Name;

        DeviceInfo ditem = di;
        ditem.ModuleInfo.vSettings.clear();
        ditem.ModuleInfo.vSettings.push_back(*itr);
        DeviceController* pControler = new DeviceController();
        pControler->regConsumer(this);
        pControler->CreateDevice(ditem);

        lstTopics.push_back(table);
        m_mpDevControler[table].push_back(pControler);
    }
    if( lstTopics.begin() != lstTopics.end())
    {
        m_pTDengine->Subscribe(lstTopics);
    }
}

void DataEngine::startup_CommAlarm(DeviceInfo di)
{
    if( di.ModuleInfo.vSettings.begin() == di.ModuleInfo.vSettings.end() )
    {
        return;
    }

    DeviceController* pControler = new DeviceController();
    pControler->regConsumer(this);
    pControler->CreateDevice(di);

    std::list<std::string> lstTopics;
    std::vector<Setting>::iterator itr;
    for( itr = di.ModuleInfo.vSettings.begin(); itr != di.ModuleInfo.vSettings.end(); ++itr )
    {
        Setting oSet = *itr;
        QJsonDocument jsonDoc(QJsonDocument::fromJson(oSet.qValue.toString().toLocal8Bit()));
        QJsonArray json = jsonDoc.array();
        for(auto i : json)
        {
            std::string topic = i.toString().toLocal8Bit().toStdString();

            lstTopics.push_back(topic);
            m_mpDevControler[topic].push_back(pControler);
        }
    }
    if( lstTopics.begin() != lstTopics.end())
    {
        m_pTDengine->Subscribe(lstTopics);
    }
}

void DataEngine::startup_AlarmMgr(DeviceInfo di)
{
    DeviceController* pControler = new DeviceController();
    pControler->regConsumer(this);
    pControler->CreateDevice(di);

    std::list<std::string> lstTopics;
    std::vector<Setting>::iterator itr;
    for( itr = di.ModuleInfo.vSettings.begin(); itr != di.ModuleInfo.vSettings.end(); ++itr )
    {
        Setting oSet = *itr;
        QJsonDocument jsonDoc(QJsonDocument::fromJson(oSet.qValue.toString().toLocal8Bit()));
        QJsonArray json = jsonDoc.array();
        for(auto i : json)
        {
            std::string topic = i.toString().toLocal8Bit().toStdString();

            lstTopics.push_back(topic);
            m_mpDevControler[topic].push_back(pControler);
        }
    }
    if( lstTopics.begin() != lstTopics.end())
    {
        m_pMqtt->Subscribe(lstTopics);
    }
}

void DataEngine::Startup()
{
    // merge file name
    QString szWorkDir = qApp->applicationDirPath();
    if( szWorkDir.right(1).contains("/") == false && szWorkDir.right(1).contains("\\") == false)
    {
        szWorkDir += "/";
    }
    std::string szFile = szWorkDir.toLocal8Bit().toStdString() + "config/server.json";

    // load server config
    std::map<std::string,tagSetup> mpConfig;
    std::string cfg = loadContent(szFile);
    QJsonDocument jsonDoc(QJsonDocument::fromJson(cfg.c_str()));
    QJsonArray ja = jsonDoc.array();
    for(auto i : ja)
    {
        QJsonObject json = i.toObject();
        QStringList items = json.keys();
        if( items.contains("ignore",Qt::CaseInsensitive) && json["ignore"].type() == QJsonValue::Bool && json["ignore"].toBool())
        {
            continue;
        }

        tagSetup ts;
        ts.addr = json["addr"].toString().toLocal8Bit().toStdString();
        ts.user = json["user"].toString().toLocal8Bit().toStdString();
        ts.password = json["password"].toString().toLocal8Bit().toStdString();
        ts.db = json["database"].toString().toLocal8Bit().toStdString();
        ts.port = json["port"].toInt();

        std::string szName = json["name"].toString().toLocal8Bit().toStdString();
        mpConfig[szName] = ts;
    }

    m_pRedis = new RedisAgent();
    m_pRedis->Connect(mpConfig["redis"]);           // redis

    m_pTDengine = new TDAgent();
    m_pTDengine->Connect(mpConfig["tdengine"]);     // td
    qRegisterMetaType<std::string>("std::string");
    connect(m_pTDengine,SIGNAL(sigEvent(const std::string, const std::string)),this,SLOT(sltSubData_td(const std::string, const std::string)));

    m_pMqtt = new MQTTAgent();
    m_pMqtt->Connect(mpConfig["mqtt"]);             // mqtt
    connect(m_pMqtt,SIGNAL(sigEvent(const std::string, const std::string)),this,SLOT(sltSubData_mqtt(const std::string, const std::string)));

    // server.id
    DeviceManagerProxy dm(m_pRedis);
    DeviceInfo da = dm.loadDataAlarm("alarmrule", "plugins/AlarmRecognition.dll");
    startup_DataAlarm(da);

    DeviceInfo ca = dm.loadCommAlarm("alarmruledevicestatus", "plugins/CommRecognition.dll");
    startup_CommAlarm(ca);

    /* 这里是想向mqtt订阅，在另外的模块里，实现对数据库操作
    std::list<std::string> lstTopics;
    DeviceInfo am = dm.loadAlarmMgr("alarmrule", "plugins/AlarmMgr.dll");
    std::list<DataItem>::iterator itr;
    for( itr = da.ModuleInfo.Properties.begin(); itr != da.ModuleInfo.Properties.end(); ++itr )
    {
        lstTopics.push_back(da.ModuleInfo.Code + "_" + itr->Code);
    }
    if( lstTopics.begin() != lstTopics.end())
    {
        m_pMqtt->Subscribe(lstTopics);
    }
    */
}