提交漏掉的文件

CommRecognition/CommRecognition.pro

@@ -0,0 +1,22 @@
+QT = core
+QT -= gui
+
+CONFIG += c++17 cmdline
+
+TARGET = CommRecognition
+TEMPLATE = lib
+
+unix{
+}
+else{
+    DESTDIR = $$PWD/../bin/plugins
+}
+
+DEFINES += COMMRECOGNITION_LIBRARY
+INCLUDEPATH += ../include
+
+SOURCES += \
+        Module.cpp
+
+HEADERS += \
+    Module.h

CommRecognition/Module.cpp

@@ -0,0 +1,202 @@
+
+#include "Module.h"
+#include <QJsonArray>
+#include <QJsonObject>
+#include <QJsonDocument>
+#include <QtCore/QVariant>
+#include <QtCore/QDebug>
+#include <stdlib.h>
+#include <math.h>
+#include <thread>
+
+void split(const std::string source,const std::string delim,std::vector<std::string>& dest)
+{
+    long n = delim.length();
+    long lt = source.length();
+
+    long last = 0;
+    size_t index = source.find_first_of(delim,last);
+    while(index != std::string::npos)
+    {
+        dest.push_back(source.substr(last,index-last));
+        last = index + n;
+        index = source.find_first_of(delim,last);
+    }
+
+    long len = source.length();
+    if( len - last > 0)
+    {
+        dest.push_back(source.substr(last,len - last));
+    }
+}
+
+Module::Module()
+    :RunnableModule()
+{
+    m_pDataConsumer = nullptr;
+    m_szOutputs.clear();
+    m_mpParamConfig.clear();
+    m_mpDataValue.clear();
+    m_nLoopIntervalMS = UINT_FAST32_MAX;
+}
+
+bool Module::isInheritedFrom(std::string tp)
+{
+    bool bEqual = (strcasecmp(tp.c_str(), "BaseModule") == 0);
+    bEqual |= (strcasecmp(tp.c_str(), "RunnableModule") == 0);
+    return bEqual;
+}
+
+void Module::regConsumer(DataConsumer* pDC)
+{
+    m_pDataConsumer = pDC;
+}
+
+void Module::Setup(ModuleInfo mi)
+{
+    std::list<DataItem>::iterator itrO;
+    for( itrO = mi.Properties.begin(); itrO != mi.Properties.end(); ++itrO )
+    {
+        m_szOutputs.push_back(mi.Code + "." + itrO->Code);
+    }
+
+    std::vector<Setting>::iterator itr;
+    for( itr = mi.vSettings.begin(); itr != mi.vSettings.end(); ++itr )
+    {
+        QString szConfig = itr->qValue.toString();
+        QJsonDocument jsonDoc(QJsonDocument::fromJson(szConfig.toLocal8Bit().toStdString().c_str()));
+        QJsonArray ja = jsonDoc.array();
+        for(auto i : ja)
+        {
+            QString item = i.toString();
+            m_mpParamConfig[item.toLocal8Bit().toStdString()] = tagParam();
+        }
+    }
+}
+
+void Module::OnSubData(std::string name,std::string)
+{
+    if( m_mpParamConfig.find(name) == m_mpParamConfig.end())
+    {
+        return;
+    }
+
+    // add data
+    m_objDataLock.lockForWrite();
+    m_mpDataValue[name] = QDateTime::currentDateTime();
+    m_objDataLock.unlock();
+}
+
+void Module::Check()
+{
+    std::map<std::string,tagParam>::iterator itr;
+    for( itr = m_mpParamConfig.begin(); itr != m_mpParamConfig.end(); ++itr )
+    {
+        pubStatus(itr->first,EAS_Breaked);
+    }
+
+    while(1)
+    {
+        if( m_pDataConsumer == nullptr )
+        {
+            std::this_thread::sleep_for(std::chrono::milliseconds(m_nLoopIntervalMS));
+            continue;
+        }
+
+        // get first time
+        m_objDataLock.lockForRead();
+        std::map<std::string,QDateTime> mpCache = m_mpDataValue;
+        m_objDataLock.unlock();
+
+        // no data
+        QDateTime dtNow = QDateTime::currentDateTime();
+
+        // status merge
+        std::map<std::string,QDateTime>::iterator itrR;
+        for( itrR = mpCache.begin(); itrR != mpCache.end(); ++itrR )
+        {
+            if( m_mpParamConfig.find(itrR->first) == m_mpParamConfig.end())
+            {
+                continue;
+            }
+
+            uint nIntervalMS = m_mpParamConfig[itrR->first].uIntervalMS;
+            qint64 tDiff = itrR->second.msecsTo(dtNow);
+            if( tDiff > nIntervalMS )
+            {
+                if( m_mpParamConfig[itrR->first].nStatus != EAS_Breaked )
+                {
+                    m_mpParamConfig[itrR->first].nStatus = EAS_Breaked;
+                    m_mpParamConfig[itrR->first].bChanged = true;
+
+                    pubStatus(itrR->first,EAS_Breaked);
+                }
+                else
+                {
+                    m_mpParamConfig[itrR->first].bChanged = false;
+                }
+            }
+            else
+            {
+                if( m_mpParamConfig[itrR->first].nStatus == EAS_Breaked )
+                {
+                    m_mpParamConfig[itrR->first].nStatus = EAS_Recover;
+                    m_mpParamConfig[itrR->first].bChanged = true;
+
+                    pubStatus(itrR->first,EAS_Breaked);
+                }
+                else
+                {
+                    m_mpParamConfig[itrR->first].nStatus = EAS_Normal;
+                    m_mpParamConfig[itrR->first].bChanged = false;
+                }
+            }
+        }
+
+        std::this_thread::sleep_for(std::chrono::milliseconds(m_nLoopIntervalMS));
+    }
+}
+
+void Module::pubStatus(std::string name,int nStatus)
+{
+    std::vector<std::string>::iterator itr;
+    for( itr = m_szOutputs.begin(); itr != m_szOutputs.end(); ++itr )
+    {
+        QString szStatus;
+        switch( nStatus )
+        {
+        case EAS_Normal:
+            szStatus = "Normal";
+            break;
+        case EAS_Breaked:
+            szStatus = "Breaked";
+            break;
+        case EAS_Recover:
+            szStatus = "Recover";
+            break;
+        }
+
+        QString szMSG = QString("{\"%1\":\"%2\"}").arg(name.c_str()).arg(szStatus);
+        m_pDataConsumer->OnData(*itr,QVariant(szMSG));
+    }
+}
+
+void Module::Run()
+{
+    std::thread t(&Module::Check,this);
+    t.detach();
+}
+
+BaseModule* instance()
+{
+    return new Module();
+}
+
+void destroy(BaseModule* pInstance)
+{
+    if( pInstance )
+    {
+        delete pInstance;
+    }
+}
+

CommRecognition/Module.h

@@ -0,0 +1,54 @@
+#pragma once
+#include "RunnableModule.h"
+#include "libaray_symbols.h"
+#include "LockFreeQueue.h"
+#include <QtCore/QDateTime>
+#include <QtCore/QReadWriteLock>
+
+enum EAlarmStatus
+{
+    EAS_Normal = 0,
+    EAS_Breaked = 1,
+    EAS_Recover = 2
+};
+
+// 全局唯一实例
+// 数据通断告警
+class COMMRECOGNITION_EXPORT Module : public RunnableModule
+{
+public:
+    Module();
+
+private:
+    struct tagParam{
+        uint uWinSize = 3;
+        uint uIntervalMS = 1000;
+        int nStatus = EAS_Breaked;
+        bool bChanged = false;
+    };
+
+private:
+    DataConsumer*                       m_pDataConsumer = nullptr;
+    std::vector<std::string>            m_szOutputs;         // 数据名称
+    std::map<std::string,tagParam>      m_mpParamConfig;     // 数据参数
+    uint                                m_nLoopIntervalMS;   // 检测周期
+
+private:
+    std::map<std::string,QDateTime>     m_mpDataValue;       // 测量数据
+    QReadWriteLock                      m_objDataLock;
+
+private:
+    void pubStatus(std::string,int);
+    void Check();
+    virtual void Run();
+public:
+    virtual void Setup(ModuleInfo mi);
+    virtual void regConsumer(DataConsumer* pDC) ;
+    virtual bool isInheritedFrom(std::string tp);
+    virtual void OnSubData(std::string name,std::string val);
+};
+
+extern "C" {//一定要添加上
+COMMRECOGNITION_EXPORT BaseModule* instance();
+COMMRECOGNITION_EXPORT void destroy(BaseModule*);
+}

CommRecognition/libaray_symbols.h

@@ -0,0 +1,8 @@
+#pragma once
+#include <QtCore/QGlobal.h>
+
+#if defined(COMMRECOGNITION_LIBRARY)
+#  define COMMRECOGNITION_EXPORT Q_DECL_EXPORT
+#else
+#  define COMMMRECOGNITION_EXPORT Q_DECL_IMPORT
+#endif

CommRecognition/main.cpp

@@ -0,0 +1,22 @@
+
+#include <QCoreApplication>
+
+#include <QLocale>
+#include <QTranslator>
+
+int main(int argc, char *argv[])
+{
+    QCoreApplication a(argc, argv);
+
+    QTranslator translator;
+    const QStringList uiLanguages = QLocale::system().uiLanguages();
+    for (const QString &locale : uiLanguages) {
+        const QString baseName = "SampleTemperature_" + QLocale(locale).name();
+        if (translator.load(":/i18n/" + baseName)) {
+            a.installTranslator(&translator);
+            break;
+        }
+    }
+
+    return a.exec();
+}

MQTTAgent/qmqtt_global.h

@@ -0,0 +1,48 @@
+/*
+ * qmqtt_global.h - qmqtt libray global
+ *
+ * Copyright (c) 2013  Ery Lee <ery.lee at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of mqttc nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+#ifndef QMQTT_GLOBAL_H
+#define QMQTT_GLOBAL_H
+
+#include <QtGlobal>
+
+#if !defined(QT_STATIC) && !defined(MQTT_PROJECT_INCLUDE_SRC)
+#  if defined(QT_BUILD_QMQTT_LIB)
+#    define Q_MQTT_EXPORT Q_DECL_EXPORT
+#  else
+#    define Q_MQTT_EXPORT Q_DECL_IMPORT
+#  endif
+#else
+#  define Q_MQTT_EXPORT
+#endif
+
+#endif // QMQTT_GLOBAL_H
+

MQTTAgent/widget.cpp

@@ -0,0 +1,67 @@
+#include "widget.h"
+#include "ui_widget.h"
+
+Widget::Widget(QWidget *parent)
+    : QWidget(parent)
+    , ui(new Ui::Widget)
+{
+    ui->setupUi(this);
+
+    mqtt = new QMQTT::Client();
+
+    connect(mqtt, &QMQTT::Client::connected, []{
+        qDebug()<< "connected";
+    });
+    connect(mqtt, &QMQTT::Client::disconnected, []{
+        qDebug()<< "disconnect";
+    });
+    connect(mqtt, &QMQTT::Client::received, this, &Widget::onReceived);
+
+    mqtt->setHostName("y.kjxry.cn");
+    //mqtt->setHostName("mq.tongxinmao.com"); //18830
+    //mqtt->setHostName("www.kjxry.cn");
+    mqtt->setPort(port);
+    mqtt->setKeepAlive(60);
+    mqtt->setClientId("C001"); //唯一id, 相同id不能同时连接
+    mqtt->setUsername(userName);
+    mqtt->setPassword(password.toUtf8());
+
+    mqtt->setAutoReconnect(true); //开启自动重连
+    mqtt->setCleanSession(true); //非持久化连接，上线时，将不再关心之前所有的订阅关系以及离线消息
+
+    mqtt->setVersion(QMQTT::V3_1_1);
+    qDebug()<< "ver" << mqtt->version();
+
+    mqtt->connectToHost();
+}
+
+Widget::~Widget()
+{
+    delete ui;
+}
+
+//接收到的消息
+void Widget::onReceived(const QMQTT::Message& message)
+{
+    qDebug() << "recv" << message.topic() << message.payload();
+}
+
+void Widget::on_pushButton_2_clicked()
+{
+    //不同的订阅方式
+    //mqtt->subscribe("alarm/led", 1);
+    //mqtt->subscribe("+/led", 1);
+    mqtt->subscribe("alarm/#", 1);
+}
+
+void Widget::on_pushButton_3_clicked()
+{
+    //QMQTT::Message message(i, EXAMPLE_TOPIC, QString("Number %1").arg(i).toUtf8());
+    QMQTT::Message message;
+    message.setTopic(topic);
+    message.setPayload("test");
+    message.setRetain(true); //保留最后一条数据
+
+    mqtt->publish(message);
+}
+

MQTTAgent/widget.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include <QWidget>
+#include <qmqtt.h>
+
+QT_BEGIN_NAMESPACE
+namespace Ui { class Widget; }
+QT_END_NAMESPACE
+
+class Widget : public QWidget
+{
+    Q_OBJECT
+
+public:
+    Widget(QWidget *parent = nullptr);
+    ~Widget();
+
+private slots:
+    void on_pushButton_2_clicked();
+    void onReceived(const QMQTT::Message& message);
+    void on_pushButton_3_clicked();
+
+private:
+    Ui::Widget *ui;
+
+    QMQTT::Client *mqtt;
+    const quint16 port = 1883;
+    const QString userName = "admin";
+    const QString password = "N6pNXbZjspDRqNGnxMmc";
+    const QString topic = "alarm/led";
+};

MQTTAgent/widget.ui

@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ui version="4.0">
+ <class>Widget</class>
+ <widget class="QWidget" name="Widget">
+  <property name="geometry">
+   <rect>
+    <x>0</x>
+    <y>0</y>
+    <width>432</width>
+    <height>357</height>
+   </rect>
+  </property>
+  <property name="windowTitle">
+   <string>Widget</string>
+  </property>
+  <widget class="QPushButton" name="pushButton_2">
+   <property name="geometry">
+    <rect>
+     <x>50</x>
+     <y>80</y>
+     <width>75</width>
+     <height>23</height>
+    </rect>
+   </property>
+   <property name="text">
+    <string>Sub</string>
+   </property>
+  </widget>
+  <widget class="QPushButton" name="pushButton_3">
+   <property name="geometry">
+    <rect>
+     <x>50</x>
+     <y>130</y>
+     <width>75</width>
+     <height>23</height>
+    </rect>
+   </property>
+   <property name="text">
+    <string>Pub</string>
+   </property>
+  </widget>
+ </widget>
+ <resources/>
+ <connections/>
+</ui>

TDAgent/TDengine.cpp

@@ -125,6 +125,7 @@ tmq_list_t* TDengine::buildTopicList()
     for( itr = g_lstTopics.begin(); itr != g_lstTopics.end(); ++itr )
     {
         std::string szTopic = *itr;
+        qDebug() << __FILE__ << __LINE__ << " topic:" << szTopic.c_str();
         int32_t code = tmq_list_append(topicList, szTopic.c_str());
         if (code) {
             return NULL;
@@ -234,6 +235,7 @@ void TDengine::start()
                          dbName.toStdString().c_str(),
                          m_nPort);
     if (pConn == NULL) {
+        qDebug() << __FILE__ << __LINE__ << "taos_connect failed:" << taos_errstr(pConn);
         qDebug()<< __FILE__ << __LINE__ << "td conn err.";
         return;
     }

TDAgent/include/taos.h

@@ -51,8 +51,7 @@ typedef void   TAOS_SUB;
 #define TSDB_DATA_TYPE_BLOB       18  // binary
 #define TSDB_DATA_TYPE_MEDIUMBLOB 19
 #define TSDB_DATA_TYPE_BINARY     TSDB_DATA_TYPE_VARCHAR  // string
-#define TSDB_DATA_TYPE_GEOMETRY   20  // geometry
-#define TSDB_DATA_TYPE_MAX        21
+#define TSDB_DATA_TYPE_MAX        20
 
 typedef enum {
   TSDB_OPTION_LOCALE,
@@ -102,7 +101,6 @@ typedef struct TAOS_FIELD_E {
 #endif
 
 typedef void (*__taos_async_fn_t)(void *param, TAOS_RES *res, int code);
-typedef void (*__taos_notify_fn_t)(void *param, void *ext, int type);
 
 typedef struct TAOS_MULTI_BIND {
   int       buffer_type;
@@ -123,10 +121,6 @@ typedef enum {
   SET_CONF_RET_ERR_TOO_LONG = -6
 } SET_CONF_RET_CODE;
 
-typedef enum {
-  TAOS_NOTIFY_PASSVER = 0,
-} TAOS_NOTIFY_TYPE;
-
 #define RET_MSG_LENGTH 1024
 typedef struct setConfRet {
   SET_CONF_RET_CODE retCode;
@@ -168,7 +162,7 @@ DLL_EXPORT int        taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name
 DLL_EXPORT int        taos_stmt_get_tag_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields);
 DLL_EXPORT int        taos_stmt_get_col_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields);
 // let stmt to reclaim TAOS_FIELD_E that was allocated by `taos_stmt_get_tag_fields`/`taos_stmt_get_col_fields`
-DLL_EXPORT void taos_stmt_reclaim_fields(TAOS_STMT *stmt, TAOS_FIELD_E *fields);
+DLL_EXPORT void       taos_stmt_reclaim_fields(TAOS_STMT *stmt, TAOS_FIELD_E *fields);
 
 DLL_EXPORT int       taos_stmt_is_insert(TAOS_STMT *stmt, int *insert);
 DLL_EXPORT int       taos_stmt_num_params(TAOS_STMT *stmt, int *nums);
@@ -231,11 +225,6 @@ DLL_EXPORT int taos_get_tables_vgId(TAOS *taos, const char *db, const char *tabl
 
 DLL_EXPORT int taos_load_table_info(TAOS *taos, const char *tableNameList);
 
-// set heart beat thread quit mode , if quicByKill 1 then kill thread else quit from inner
-DLL_EXPORT void taos_set_hb_quit(int8_t quitByKill);
-
-DLL_EXPORT int taos_set_notify_cb(TAOS *taos, __taos_notify_fn_t fp, void *param, int type);
-
 /*  --------------------------schemaless INTERFACE------------------------------- */
 
 DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision);
@@ -273,12 +262,6 @@ DLL_EXPORT tmq_t *tmq_consumer_new(tmq_conf_t *conf, char *errstr, int32_t errst
 DLL_EXPORT const char *tmq_err2str(int32_t code);
 
 /* ------------------------TMQ CONSUMER INTERFACE------------------------ */
-typedef struct tmq_topic_assignment {
-  int32_t vgId;
-  int64_t currentOffset;
-  int64_t begin;
-  int64_t end;
-} tmq_topic_assignment;
 
 DLL_EXPORT int32_t   tmq_subscribe(tmq_t *tmq, const tmq_list_t *topic_list);
 DLL_EXPORT int32_t   tmq_unsubscribe(tmq_t *tmq);
@@ -287,10 +270,6 @@ DLL_EXPORT TAOS_RES *tmq_consumer_poll(tmq_t *tmq, int64_t timeout);
 DLL_EXPORT int32_t   tmq_consumer_close(tmq_t *tmq);
 DLL_EXPORT int32_t   tmq_commit_sync(tmq_t *tmq, const TAOS_RES *msg);
 DLL_EXPORT void      tmq_commit_async(tmq_t *tmq, const TAOS_RES *msg, tmq_commit_cb *cb, void *param);
-DLL_EXPORT int32_t   tmq_get_topic_assignment(tmq_t *tmq, const char *pTopicName, tmq_topic_assignment **assignment,
-                                              int32_t *numOfAssignment);
-DLL_EXPORT void      tmq_free_assignment(tmq_topic_assignment* pAssignment);
-DLL_EXPORT int32_t   tmq_offset_seek(tmq_t *tmq, const char *pTopicName, int32_t vgId, int64_t offset);
 
 /* ----------------------TMQ CONFIGURATION INTERFACE---------------------- */
 
@@ -312,7 +291,6 @@ DLL_EXPORT void           tmq_conf_set_auto_commit_cb(tmq_conf_t *conf, tmq_comm
 DLL_EXPORT const char *tmq_get_topic_name(TAOS_RES *res);
 DLL_EXPORT const char *tmq_get_db_name(TAOS_RES *res);
 DLL_EXPORT int32_t     tmq_get_vgroup_id(TAOS_RES *res);
-DLL_EXPORT int64_t     tmq_get_vgroup_offset(TAOS_RES* res);
 
 /* ------------------------------ TAOSX -----------------------------------*/
 // note: following apis are unstable