lp-data-store/modules/TDengineClient/TDengineClient.cpp

#include "TDengineClient.h"
#include "taos.h"
#include <QtCore/QDebug>
#include <QtConcurrent/QtConcurrent>

#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include "taos.h"



int32_t TDengineClient::msgProcess(TAOS_RES* msg)
{
    char    buf[1024];
    int32_t rows = 0;

    const char* topicName = tmq_get_topic_name(msg);
    const char* dbName = tmq_get_db_name(msg);
    int32_t     vgroupId = tmq_get_vgroup_id(msg);

    //qDebug() << __FILE__ << __LINE__ << "db: " << dbName;
    //qDebug() << __FILE__ << __LINE__ << "topic: " << topicName;
    //qDebug() << __FILE__ << __LINE__ << "vgroup id:" << vgroupId;

    while (1)
    {
        TAOS_ROW row = taos_fetch_row(msg);
        if (row == NULL)
            break;

        TAOS_FIELD* fields = taos_fetch_fields(msg);
        int32_t     numOfFields = taos_field_count(msg);
        rows++;
        taos_print_row(buf, row, fields, numOfFields);

        QJsonObject jsonObject;
        for(int k = 0;k < numOfFields;k++)
        {
            QString fieldName = fields[k].name;
            if(fields[k].type == TSDB_DATA_TYPE_TIMESTAMP)
            {
                uint64_t value = *((int64_t *)row[k]);
                int intvalue = (int)value;
                jsonObject.insert(fieldName,intvalue);
            }
            else if(fields[k].type == TSDB_DATA_TYPE_VARCHAR)
            {
                QString value((char *)row[k]);
                jsonObject.insert(fieldName,value);
            }
            else if(fields[k].type == TSDB_DATA_TYPE_DOUBLE)
            {
                double value = *(double *)(row[k]);
                jsonObject.insert(fieldName,value);
            }
            else if(fields[k].type == TSDB_DATA_TYPE_FLOAT)
            {
                float value = *(float *)(row[k]);
                jsonObject.insert(fieldName,value);
            }
            else if(fields[k].type == TSDB_DATA_TYPE_INT)
            {
                int value = *(int *)(row[k]);
                jsonObject.insert(fieldName,value);
            }
            else if(fields[k].type == TSDB_DATA_TYPE_BIGINT)
            {
                int64_t value = *(int64_t *)(row[k]);
                int intvalue = (int)value;
                jsonObject.insert(fieldName,intvalue);
            }
        }

        std::string content = QJsonDocument(jsonObject).toJson(QJsonDocument::Compact).toStdString();

        //if( g_pSubCB != nullptr )
        {
            std::string topic = topicName;
            //std::string content = buf;
            callback((char*)topic.c_str(),(char*)content.c_str(), usrData);
        }
        //qDebug() << __FILE__ << __LINE__ << "row content: " << buf;
    }

    return rows;
}

//构建消费者
tmq_t* TDengineClient::buildConsumer()
{
    tmq_conf_res_t code;
    tmq_conf_t*    conf = tmq_conf_new();

    code = tmq_conf_set(conf, "td.connect.ip", host.toStdString().c_str());
    if (TMQ_CONF_OK != code) {
        tmq_conf_destroy(conf);
        return NULL;
    }

    //code = tmq_conf_set(conf, "enable.auto.commit", "true");
    code = tmq_conf_set(conf, "enable.auto.commit", "false"); //禁用提交回调
    if (TMQ_CONF_OK != code) {
        tmq_conf_destroy(conf);
        return NULL;
    }

    //    code = tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
    //    if (TMQ_CONF_OK != code) {
    //        tmq_conf_destroy(conf);
    //        return NULL;
    //    }

    code = tmq_conf_set(conf, "group.id", "cgrpName");
    if (TMQ_CONF_OK != code) {
        tmq_conf_destroy(conf);
        return NULL;
    }

    code = tmq_conf_set(conf, "client.id", "user defined name");
    if (TMQ_CONF_OK != code) {
        tmq_conf_destroy(conf);
        return NULL;
    }

    code = tmq_conf_set(conf, "td.connect.user", user.toStdString().c_str());
    if (TMQ_CONF_OK != code) {
        tmq_conf_destroy(conf);
        return NULL;
    }

    code = tmq_conf_set(conf, "td.connect.pass", password.toStdString().c_str());
    if (TMQ_CONF_OK != code) {
        tmq_conf_destroy(conf);
        return NULL;
    }

    //v3.2以后默认为最新的latest
//    code = tmq_conf_set(conf, "auto.offset.reset", "latest");
//    if (TMQ_CONF_OK != code) {
//        tmq_conf_destroy(conf);
//        return NULL;
//    }

    //tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL); //禁用提交回调

    tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
    //qDebug()<< __FILE__ << __LINE__ << "tmq" << tmq;
    tmq_conf_destroy(conf);
    return tmq;
}

//构建主题
tmq_list_t* TDengineClient::buildTopicList()
{
    tmq_list_t* tmqList = tmq_list_new();

    std::set<std::string>::iterator itr;
    for( itr = topicList.begin(); itr != topicList.end(); ++itr )
    {
        std::string szTopic = *itr;
        int32_t code = tmq_list_append(tmqList, szTopic.c_str());
        if (code) {
            return NULL;
        }
    }
    return tmqList;
}

//轮询主题
void TDengineClient::topicLoop()
{
    int32_t code;
    tmq_t* tmq = buildConsumer();
    if (NULL == tmq) {
        qDebug()<< __FILE__ << __LINE__ << "err" << stderr;
        return;
    }

    tmq_list_t* topic_list = buildTopicList();
    code = tmq_subscribe(tmq, topic_list);
    if ( code ) {
        qDebug() << __FILE__ << __LINE__ << "Failed to tmq_subscribe(): " << tmq_err2str(code);
    }

    tmq_list_destroy(topic_list);

    int32_t totalRows = 0;
    int32_t msgCnt = 0;
    int32_t timeout = 100; //多个主题共用1个线程轮询，超时时间太长会影响其它主题实时性
    while (true) {
        TAOS_RES* tmqmsg = tmq_consumer_poll(tmq, timeout);
        if (tmqmsg) {
            msgCnt++;
            totalRows += msgProcess(tmqmsg);
            taos_free_result(tmqmsg);
        } else {
            //超时
            //break;
        }
    }

    code = tmq_consumer_close(tmq);
    if (code) {
        fprintf(stderr, "%% Failed to close consumer: %s\n", tmq_err2str(code));
    } else {
        fprintf(stderr, "%% Consumer closed\n");
    }
    qDebug()<< __FILE__ << __LINE__ << "stderr" << stderr;
}

TDengineClient::TDengineClient(QObject *parent)
    : QObject(parent)
{
    pConn = nullptr;
    //g_pSubCB = nullptr;
    topicList.clear();
}

TDengineClient::~TDengineClient()
{
    taos_close(pConn);
    if(future.isRunning()){
        future.waitForFinished();
    }
}

void TDengineClient::exec(QString sql)
{
    if (pConn)
    {
        TAOS_RES* pRes = taos_query(pConn, sql.toStdString().c_str());
        if (taos_errno(pRes) != 0) {
            qDebug() << __FILE__ << __LINE__ << "failed to insert into tab, reason:" << taos_errstr(pRes);
            qDebug()<< __FILE__ << __LINE__ << "err sql" << sql;
        }
        else
        {
            //qDebug()<< "exec ok";
        }
        taos_free_result(pRes);
    }
}

void TDengineClient::subscribe(QString ch, std::function<void(const char* topic, const char* data, void*usr)>& fn , void* usrdata)
{
    callback = fn;
    usrData = usrdata;
    topicList.insert(ch.toLocal8Bit().toStdString());
}

void TDengineClient::psubscribe(QString ch, std::function<void(const char* topic, const char* data, void*usr)>& fn, void*usrdata)
{
    callback = fn;
    usrData = usrdata;
    topicList.insert(ch.toLocal8Bit().toStdString());
}

void TDengineClient::Setup(const char* _host, const char* _user, const char* _passwd, uint16_t _port)
{
    // host = ts.addr.c_str();
    // user = ts.user.c_str();
    // dbName = ts.db.c_str();
    // password = ts.password.c_str();
    // port = ts.port;
    host = _host;
    user = _user;
    password = _passwd;
    port = _port;
}

void TDengineClient::start()
{
    pConn = taos_connect(host.toStdString().c_str(),
                         user.toStdString().c_str(),
                         password.toStdString().c_str(),
                         dbName.toStdString().c_str(),
                         port);
    if (pConn == NULL) {
        qDebug()<< __FILE__ << __LINE__ << "td conn err.";
        return;
    }

    //开启一个线程轮询订阅的主题
    future = QtConcurrent::run(this, &TDengineClient::topicLoop);

}