data-writer/ytDataWriteProcess/mqttdatathread.cpp

#include "mqttdatathread.h"
#include "../DataWriteDog/writeshm.h"

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

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

void MqttDataThread::run()
{

}

void MqttDataThread::time_out()
{
    dataWriteShm->processStatus[1].t_time = QDateTime::currentDateTime().toTime_t();
    if((m_client->connectionState()==QMQTT::STATE_DISCONNECTED)||(m_client->connectionState()==QMQTT::STATE_INIT)){
        m_client->connectToHost();
    }
}

void MqttDataThread::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(false);
    m_client->setAutoReconnect(true);
    m_client->setAutoReconnectInterval(60000);
    connect(m_client,&QMQTT::Client::connected,this,&MqttDataThread::onConnected);
    connect(m_client,&QMQTT::Client::received,this,&MqttDataThread::onReceived);
    m_client->connectToHost();


    timer->start(1000);
}

void MqttDataThread::onConnected()
{
    emit mq_log(QString("[%1] WriteProcess mqttdatathread onConnected").arg(QDateTime::currentDateTime().toString("yyyy-MM-dd HH:mm:ss.zzz")));

    m_client->subscribe("data-collector",0);

}

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

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


    emit sendMqttData(topic,data);
}