#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(SHM_PORT)))==-1){ return false; } if((shmid=shmget(key,sizeof(AGBoxShm),IPC_CREAT|0666))==-1){ return false; } agBoxShm = static_cast(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(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)); }