data-transfer/ytDataCollectorCore/datacollectorcore.cpp

#include "datacollectorcore.h"

DataCollectorCore::DataCollectorCore(QObject *parent) : QObject(parent)
{
    mqttIdx = 1;

    logthread = new logThread(this);
    logthread->start();

    dataSub = new DataCollectorSub(this);
    connect(dataSub,&DataCollectorSub::sendDevData,this,&DataCollectorCore::receiveDevData);

    dataPub = new DataCollectorPub(this);
    connect(dataPub,&DataCollectorPub::publishDevData,this,&DataCollectorCore::publishDevData);
    connect(dataPub,&DataCollectorPub::dataListLog,this,&DataCollectorCore::dataListLog);

    m_client = new QMQTT::Client(QHostAddress("47.98.201.73"),1883,this);
    connect(m_client,&QMQTT::Client::connected,this,&DataCollectorCore::onConnected);
    m_client->setUsername("usky");
    m_client->setPassword("usky");
    m_client->setCleanSession(true);
    m_client->connectToHost();

}

DataCollectorCore::~DataCollectorCore()
{
    logthread->stop();
    dataPub->stop();

}

void DataCollectorCore::start()
{
    dataPub->start();
    dataSub->start();
}

void DataCollectorCore::dataListLog(QString log)
{
    logthread->appendData(log);
}

void DataCollectorCore::publishDevData(QByteArray data)
{
    if((m_client->connectionState()==QMQTT::STATE_DISCONNECTED)||(m_client->connectionState()==QMQTT::STATE_INIT)){
        m_client->connectToHost();
    }
    m_client->publish(QMQTT::Message(mqttIdx++,"data-collector",data));
    if(mqttIdx > 9999)
    {
        mqttIdx = 1;
    }
}

void DataCollectorCore::onConnected()
{
    printf("onConnected\n");
}

void DataCollectorCore::receiveDevData(QString topic,QByteArray data)
{
    dataPub->devMessage(DevData(topic,data));
}