data-server/ytDMPDataServer/dmpdatacore.cpp

#include "dmpdatacore.h"
#include "../ytDMPServerDog/dogshm.h"

DmpDataCore::DmpDataCore(QObject *parent) : QObject(parent)
{
    isSending1 = true;

    bstrlist.clear();
    pointstrlist.clear();

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

    dataserver= new DMPDataServer(this);
    connect(dataserver,&DMPDataServer::CommData,this,&DmpDataCore::CommData);
    connect(dataserver,&DMPDataServer::DatabaseData,this,&DmpDataCore::DatabaseData);
    connect(dataserver,&DMPDataServer::getERealtimeValue,this,&DmpDataCore::getRealtimeValue);

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

    dbthread = new DatabaseThread(this);
    dbthread->start();

    wsserver = new WebsocketServer(this);

    networkmanager1 = new QNetworkAccessManager(this);
    connect(networkmanager1,&QNetworkAccessManager::finished,this,&DmpDataCore::finishedSlot1);

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

DmpDataCore::~DmpDataCore()
{
    logthread->stop();
    dbthread->stop();
}

void DmpDataCore::start()
{
    timer->start(1000);
    dataserver->start();
}

void DmpDataCore::onConnected()
{
    printf("mqtt connected\n");
}

void DmpDataCore::time_out()
{
    uint t = QDateTime::currentDateTime().toTime_t();
    dogshm->data_feedtime = t;

    if((bstrlist.length()>0)&&(isSending1)){
        isSending1 = false;
        QString data = bstrlist.first();

        QByteArray Report = data.toUtf8();
        QNetworkRequest *req = new QNetworkRequest();
        req->setUrl(QUrl("http://172.16.120.69:8086/write?db=USKTSDB_A&u=root&p=root"));
        req->setHeader(QNetworkRequest::ContentTypeHeader,"application/json; charset=UTF-8");
        req->setHeader(QNetworkRequest::ContentLengthHeader,QString("%1").arg(Report.length()).toUtf8());
        QNetworkReply *reply = networkmanager1->post(*req,Report);
        QReplyTimeout *pTimeout = new QReplyTimeout(reply,10000);
        connect(pTimeout, SIGNAL(net_timeout()),this,SLOT(reply_timeout1()));
    }

    if(pointstrlist.length()>0){
        QString data = pointstrlist.first();
        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.toUtf8()));
        if(mqttIdx > 9999)
            mqttIdx = 1;
        pointstrlist.removeFirst();
    }
}

void DmpDataCore::reply_timeout1()
{
    logthread->appendData(QString("reply_timeout1"));

    if(bstrlist.length()>0){
        bstrlist.removeFirst();
    }
    isSending1 = true;
}

void DmpDataCore::finishedSlot1(QNetworkReply *reply)
{
    if(bstrlist.length()>0){
        if(reply->error()==QNetworkReply::NoError){
            QString bak_info = QString::fromUtf8(reply->readAll());
            logthread->appendData(bak_info);

        }else{
            QString bak_info = QString::fromUtf8(reply->readAll());
            logthread->appendData(QString("QNetworkReply1::Error [%1]").arg(bak_info));
        }

        bstrlist.removeFirst();
    }
    reply->deleteLater();
    isSending1 = true;
}

void DmpDataCore::CommData(QString data)
{
    logthread->appendData(data);
}

void DmpDataCore::DatabaseData(QString sql)
{
    dbthread->appendSql(sql);
}

void DmpDataCore::getRealtimeValue(QString DeviceCode, QString bstr, QString pointstr, QDateTime t)
{
    bstrlist.append(bstr);
    pointstrlist.append(pointstr);

    QString tmp = QString("[%1]  bstr: [%2], pointstr: [%3]").arg(QDateTime::currentDateTime().toString("yyyy-mm-dd HH:mm:ss")).arg(bstr).arg(pointstr);
    logthread->appendData(tmp);
}