data-agbox/agEnvironmentProcess/mqttsub.cpp

#include "mqttsub.h"
#include "../AGBoxDog/boxshm.h"

MqttSub::MqttSub(QObject *parent) : QThread(parent)
{

    timer = new QTimer(this);
    connect(timer,&QTimer::timeout,this,&MqttSub::time_out);
}

void MqttSub::run()
{

}

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

    if((m_client->connectionState()==QMQTT::STATE_DISCONNECTED)||(m_client->connectionState()==QMQTT::STATE_INIT)){
        m_client->connectToHost();
    }
}

void MqttSub::mqtt_conf(QString ip,QString port,QString username,QString password)
{
    mqtt_ip = ip;
    mqtt_port = port;
    mqtt_username = username;
    mqtt_passwd = password;


    m_client = new QMQTT::Client(QHostAddress(mqtt_ip),static_cast<quint16>(mqtt_port.toInt()),this);
    m_client->setUsername(mqtt_username);
    m_client->setPassword(mqtt_passwd.toLatin1());
    m_client->setCleanSession(true);
    connect(m_client,&QMQTT::Client::connected,this,&MqttSub::onConnected);
    connect(m_client,&QMQTT::Client::received,this,&MqttSub::onReceived);
    m_client->connectToHost();


    timer->start(1000);
}

void MqttSub::onConnected()
{
    m_client->subscribe("/edge/bainfo/72B4884AE5CB4407B4D5D0BACB509D79/rtg",0);
    m_client->subscribe("/edge/bainfo/271ACE89CB504642950FC7B3C2505205/rtg",0);
    m_client->subscribe("/edge/bainfo/9849B7BE3D204EF984D851EC067BF9C1/rtg",0);
    m_client->subscribe("/edge/bainfo/CE1A787947914261B2ADAF5AA9FC1EBC/rtg",0);

}

void MqttSub::onReceived(const QMQTT::Message &message)
{

    QString topic = message.topic();
    QByteArray data = message.payload();

    emit sendMqttData(topic,data);
}