data-agbox/agEnvironmentProcess/environmentcore.cpp

#include "environmentcore.h"
#include "../AGBoxDog/boxshm.h"
AGBoxShm *agBoxShm;

void EnvironmentCore::shm_init(){
    QSqlQuery qry;
    QString sql = QString("select mqtt_ip,mqtt_port,user_name,pass_word from yt_t_mqtt where item_name = 'data-agbox'");
    qry = db.exec(sql);
    while (qry.next()) {
        ip = qry.value(0).toString();
        port = qry.value(1).toString();
        username = qry.value(2).toString();
        password = qry.value(3).toString();
    }
    qry.clear();
    sql.clear();

}

bool EnvironmentCore::shm_load(){
    key_t key;
    int shmid;
    if((key=ftok(SHM_PATH,static_cast<int>(SHM_PORT)))==-1){
        return false;
    }
    if((shmid=shmget(key,sizeof(AGBoxShm),IPC_CREAT|0666))==-1){
        return false;
    }
    agBoxShm = static_cast<AGBoxShm *>(shmat(shmid,nullptr,0));

    shm_init();
    return true;
}

EnvironmentCore::EnvironmentCore(QObject *parent) : QObject(parent)
{
    logthread = new LogThread(this);
    logthread->start();

    db = QSqlDatabase::addDatabase("QSQLITE","conf_db");
    db.setDatabaseName(QString("/opt/db/yt_conf.db"));
    if(!db.open()){
        logthread->appendData(QString("[agenvironment] open yt_conf.db failed"));
    }
    if(shm_load()){
        logthread->appendData(QString("[agenvironment] shm load success"));
    }

    mqttIdx = 1;
    m_client = new QMQTT::Client(QHostAddress(ip),static_cast<quint16>(port.toInt()),this);
    connect(m_client,&QMQTT::Client::connected,this,&EnvironmentCore::onConnected);
    m_client->setUsername(username);
    m_client->setPassword(password.toLatin1());
    m_client->setCleanSession(true);
    m_client->connectToHost();

    mqttsub = new MqttSub(this);
    connect(mqttsub,&MqttSub::sendMqttData,this,&EnvironmentCore::receiveDevData);

    mqttpub = new MqttPub(this);
    connect(mqttpub,&MqttPub::mqttData,this,&EnvironmentCore::mqtt_data);
    connect(mqttpub,&MqttPub::dataListLog,this,&EnvironmentCore::dataLog);

    timer = new QTimer(this);
    connect(timer,&QTimer::timeout,this,&EnvironmentCore::time_out);
    timer->start(1000);

}


EnvironmentCore::~EnvironmentCore()
{
    logthread->stop();
    mqttpub->stop();

}

void EnvironmentCore::start()
{
    mqttsub->start();
    mqttsub->mqtt_conf(ip,port,username,password);

    mqttpub->start();
}

void EnvironmentCore::time_out()
{
    agBoxShm->processStatus[5].t_time=QDateTime::currentDateTime().toTime_t();

    for(int i=0;i<1024;i++){
        if((agBoxShm->device[i].Enabled == 0x01)&&(agBoxShm->device[i].device_type == 509)){
            uint curTime = QDateTime::currentDateTime().toTime_t();
            if((curTime - agBoxShm->device[i].lastTime)>7200){
                agBoxShm->device[i].lastTime = QDateTime::currentDateTime().toTime_t();
                this->mqtt_data(QString("{\"device_id\":\"%1\",\"device_code\":\"%2\",\"product_id\":\"%3\",\"timestamp\":%4,\"tags\":{\"conn_type\":\"\",\"type\":\"\"},\"metrics\":{\"device_status\":0},\"device_type\":\"%5-ev\"}")
                              .arg(agBoxShm->device[i].device_id).arg(agBoxShm->device[i].device_code).arg(agBoxShm->device[i].product_code).arg(curTime).arg(agBoxShm->device[i].device_type));
            }
        }
    }
}

void EnvironmentCore::onConnected()
{
    logthread->appendData(QString("mqtt onConnected"));
}

void EnvironmentCore::dataLog(QString log)
{
    logthread->appendData(log);
}

void EnvironmentCore::mqtt_data(QString mqtt_msg)
{
    printf("test5555 [%s]\n",mqtt_msg.toUtf8().data());
    logthread->appendData(QString("[%1] %2").arg(QDateTime::currentDateTime().toString("yyyy-MM-dd HH:mm:ss")).arg(mqtt_msg));
    if((m_client->connectionState()==QMQTT::STATE_INIT)||(m_client->connectionState()==QMQTT::STATE_DISCONNECTED)){
        m_client->connectToHost();
    }
    m_client->publish(QMQTT::Message(mqttIdx++,"data-collector",mqtt_msg.toUtf8()));
    if(mqttIdx > 9999){
        mqttIdx = 1;
    }
}

void EnvironmentCore::receiveDevData(QString topic,QByteArray data)
{
    mqttpub->devMessage(MqttData(topic,data));
}