增加mqtt模块

改造接口，使接口统一

Alarm.pro

@@ -5,6 +5,7 @@ CONFIG += ordered
 SUBDIRS += \
     RedisAgent/RedisAgent.pro \
     TDAgent/TDAgent.pro \
+    MQTTAgent/MQTTAgent.pro \
 #    AlarmVerification/AlarmVerification.pro \
     AlarmRecognition/AlarmRecognition.pro \
     EngineAlarm/EngineAlarm.pro \

AlarmRecognition/Module.cpp

@@ -1,5 +1,8 @@
 
 #include "Module.h"
+#include <QJsonArray>
+#include <QJsonObject>
+#include <QJsonDocument>
 #include <QtCore/QVariant>
 #include <QtCore/QDebug>
 #include <stdlib.h>
@@ -33,8 +36,49 @@ void Module::Setup(ModuleInfo mi)
         m_szOutputs.push_back(mi.Code + "." + itrO->Code);
     }
 
-    tagRange tr;
-    m_mpRangeConfig[""] = tr;
+    std::vector<Setting>::iterator itr;
+    for( itr = mi.vSettings.begin(); itr != mi.vSettings.end(); ++itr )
+    {
+        QJsonDocument jsonDoc(QJsonDocument::fromJson(itr->qValue.toString().toLocal8Bit().toStdString().c_str()));
+        QJsonObject json = jsonDoc.object();
+        QStringList keys = json.keys();
+
+        tagRange tr;
+        if( keys.contains("KeepTimes",Qt::CaseInsensitive) )
+        {
+            tr.uWinSize = abs(json["KeepTimes"].toInt());
+        }
+        if( keys.contains("Operator",Qt::CaseInsensitive) )
+        {
+            double dbOperator1 = 0;
+            if( keys.contains("Operator1",Qt::CaseInsensitive) )
+            {
+                dbOperator1 = json["Operator1"].toDouble();
+            }
+
+            double dbOperator2 = 0;
+            if( keys.contains("Operator2",Qt::CaseInsensitive) )
+            {
+                dbOperator2 = json["Operator2"].toDouble();
+            }
+
+            std::string szOp = json["Operator"].toString().toLocal8Bit().toStdString();
+            if( strcasecmp(szOp.c_str(), "rang") == 0)
+            {
+                tr.dbMin = dbOperator1;
+                tr.dbMax = dbOperator2;
+            }
+            else if( strcasecmp(szOp.c_str(), "above") == 0)
+            {
+                tr.dbMax = dbOperator1;
+            }
+            else if( strcasecmp(szOp.c_str(), "below") == 0)
+            {
+                tr.dbMin = dbOperator1;
+            }
+            m_mpRangeConfig["topic_" + itr->Name] = tr;
+        }
+    }
 }
 
 void Module::OnSubData(std::string name,std::string val)

EngineAlarm/DataEngine.cpp

@@ -37,6 +37,14 @@ void DataEngine::OnData(std::string name,QVariant val)
     {
         qCritical() << __FILE__ << __LINE__ << " " << "no publisher.";
     }
+    if( m_pMqtt != nullptr )
+    {
+        m_pMqtt->Publish(name, val);
+    }
+    else
+    {
+        qCritical() << __FILE__ << __LINE__ << " " << "no publisher.";
+    }
 }
 
 void DataEngine::sltSubData(char* topic, char* msg)
@@ -111,22 +119,29 @@ void DataEngine::Startup()
     loadConfig(szWorkDir.toLocal8Bit().toStdString() + "config/alarm.json",config);
 
     DeviceManagerProxy dm(m_pRedis);
-    std::list<AlarmRuler> lstRulers = dm.loadDevices(config.serverId, config.appName);
-
-    DeviceController* pControler = new DeviceController();
-    pControler->regConsumer(this);
-    pControler->CreateDevice(config.appName,lstRulers);
-
-    std::list<std::string> topics;
-    std::list<AlarmRuler>::iterator itr;
-    for( itr = lstRulers.begin(); itr != lstRulers.end(); ++itr )
+    std::list<DeviceInfo> lstDevices = dm.loadDevices(config.serverId, config.appName);
+    std::list<DeviceInfo>::iterator itr;
+    for( itr = lstDevices.begin();itr!= lstDevices.end();++itr)
     {
-        std::string szTopic = itr->NodeCode + "." + itr->DataItemCode;
-        m_mpDevControler[szTopic] = pControler;
-        topics.push_back(szTopic);
-    }
+        DeviceInfo di = *itr;
+        DeviceController* pControler = new DeviceController();
+        pControler->regConsumer(this);
+        pControler->CreateDevice(di);
+
+        std::list<std::string> lstTopics;
+        std::vector<Setting>::iterator itrT;
+        for( itrT = di.ModuleInfo.vSettings.begin(); itrT != di.ModuleInfo.vSettings.end(); ++itrT )
+        {
+            lstTopics.push_back("topic_" + itrT->Name);
+        }
+        m_pTDengine->Subscribe(lstTopics);
 
-    m_pTDengine->Subscribe(topics);
-    connect(m_pTDengine,SIGNAL(sigEvent(char*, char*)),this,SLOT(sltSubData(char*, char*)));
+        connect(m_pTDengine,SIGNAL(sigEvent(char*, char*)),this,SLOT(sltSubData(char*, char*)));
+        std::list<std::string>::iterator itrI;
+        for( itrI = di.ModuleInfo.lstInputs.begin(); itrI != di.ModuleInfo.lstInputs.end(); ++itrI )
+        {
+            m_mpDevControler[*itrI] = pControler;
+        }
+    }
 }
 

EngineAlarm/DataEngine.h

@@ -1,6 +1,7 @@
 #pragma once
 #include "../RedisAgent/RedisAgent.h"
 #include "../TDAgent/TDAgent.h"
+#include "../MQTTAgent/MQTTAgent.h"
 #include "BaseModule.h"
 #include "DeviceController.h"
 #include <QtCore/QObject>
@@ -16,6 +17,8 @@ public:
 private:
     RedisAgent*               m_pRedis;          // 读取配置
     TDAgent*                  m_pTDengine;       // 订阅发布
+    MQTTAgent*                m_pMqtt;           // 订阅发布
+
     std::map<std::string,
              DeviceController*>  m_mpDevControler;
 

EngineAlarm/DeviceController.cpp

@@ -34,9 +34,10 @@ void DeviceController::OnSubData(std::string name,std::string val)
     }
 }
 
-void DeviceController::CreateDevice(const std::string name,const std::list<AlarmRuler>& lst)
+void DeviceController::CreateDevice(const DeviceInfo& di)
 {
-    std::string assemblyName = name;
+    std::string assemblyName = di.ModuleInfo.AssemblyName;
+    std::string className = di.ModuleInfo.ClassName;
 
     BaseModule* pModule = LibraryLoader::load<BaseModule>(assemblyName);
     if( pModule == nullptr )
@@ -45,17 +46,8 @@ void DeviceController::CreateDevice(const std::string name,const std::list<Alarm
         return;
     }
 
-    // 组装一下数据,要不就得改接口,填充到Properties
-    ModuleInfo mi;
-    std::list<AlarmRuler>::iterator itr;
-    for(itr = lst.begin(); itr != lst.end(); ++itr)
-    {
-        DataItem data;
-        mi.Properties.push_back( data);
-    }
-
     m_pModule = pModule;
-    m_pModule->Setup(mi);
+    m_pModule->Setup(di.ModuleInfo);
     m_pModule->regConsumer(this);
     if( m_pModule->isInheritedFrom("RunnableModule"))
     {

EngineAlarm/DeviceController.h

@@ -15,6 +15,6 @@ public:
     virtual void OnData(std::string name,QVariant val);
     virtual void OnSubData(std::string name,std::string val);
     virtual void regConsumer(DataConsumer* pDC);
-    void CreateDevice(const std::string name,const std::list<AlarmRuler>& lst);
+    void CreateDevice(const DeviceInfo& di);
 };
 

EngineAlarm/DeviceManagerProxy.cpp

@@ -35,9 +35,9 @@ bool DeviceManagerProxy::parse(QString cfg,AlarmRuler& ar)
     QStringList keys = jsonObject.keys();
     foreach (QString key, keys)
     {
-        if( key.compare("nodeCode",Qt::CaseInsensitive) == 0)
+        if( key.compare("code",Qt::CaseInsensitive) == 0)
         {
-            ar.NodeCode = jsonObject["nodeCode"].toString().toLocal8Bit().toStdString();
+            ar.NodeCode = jsonObject["code"].toString().toLocal8Bit().toStdString();
         }
         else if( key.compare("dataItemCode",Qt::CaseInsensitive) == 0)
         {
@@ -64,11 +64,19 @@ bool DeviceManagerProxy::parse(QString cfg,AlarmRuler& ar)
     return true;
 }
 
-std::list<AlarmRuler> DeviceManagerProxy::loadDevices(std::string id,std::string app)
+std::list<DeviceInfo> DeviceManagerProxy::loadDevices(std::string id,std::string app)
 {
-    std::list<AlarmRuler> lstRulers;
+    std::list<DeviceInfo> lstDevices;
     if( m_pRedis != nullptr)
     {
+        DeviceInfo di;
+        ModuleInfo mi;
+        mi.Code = "alarm";
+        mi.AssemblyName = "plugins/AlarmRecognition.dll";
+
+        DataItem oDT;
+        oDT.Code = "data";
+        mi.Properties.push_back(oDT);
         QStringList lst = m_pRedis->hvals(id.c_str());
         foreach (QString str, lst)
         {
@@ -80,9 +88,23 @@ std::list<AlarmRuler> DeviceManagerProxy::loadDevices(std::string id,std::string
             AlarmRuler ar;
             if( parse(str,ar) )
             {
-                lstRulers.push_back(ar);
+                QString szJson;
+                szJson += "{";
+                szJson += QString("\"Operator\":\"%1\",").arg(ar.Operator.c_str());
+                szJson += QString("\"Operator1\":%1,").arg(ar.dbOperator1);
+                szJson += QString("\"Operator2\":%1,").arg(ar.dbOperator1);
+                szJson += QString("\"KeepTimes\":%1").arg(ar.nKeepTimes);
+                szJson += "}";
+                Setting oSet;
+                oSet.Name = ar.NodeCode;
+                oSet.qValue = szJson;
+
+                mi.vSettings.push_back(oSet);
             }
         }
+        di.ModuleInfo = mi;
+        lstDevices.push_back(di);
     }
-    return lstRulers;
+
+    return lstDevices;
 }

EngineAlarm/DeviceManagerProxy.h

@@ -15,5 +15,5 @@ private:
     bool parse(QString cfg,AlarmRuler& ar);
 
 public:
-    std::list<AlarmRuler> loadDevices(std::string id,std::string app);
+    std::list<DeviceInfo> loadDevices(std::string id,std::string app);
 };

EngineAlarm/EngineAlarm.pro

@@ -17,6 +17,7 @@ INCLUDEPATH += ../include
 
 LIBS += -L$$PWD/../lib/ -lRedisAgent
 LIBS += -L$$PWD/../lib/ -lTDAgent
+LIBS += -L$$PWD/../lib/ -lMQTTAgent
 
 SOURCES += \
         DataEngine.cpp \

MQTTAgent/MQTTAgent.cpp

@@ -0,0 +1,110 @@
+#include "MQTTAgent.h"
+#include <QMqtt.h>
+
+MQTTAgent* g_pThis;
+
+QMQTT::Client*  m_pMqtt;
+MQTTAgent::MQTTAgent()
+	: QObject()
+{
+    m_pMqtt = nullptr;
+    g_pThis = this;
+}
+
+MQTTAgent::~MQTTAgent()
+{
+    if( m_pMqtt != nullptr )
+    {
+        delete m_pMqtt;
+    }
+}
+
+void onReceived(const QMQTT::Message& message)
+{
+    emit g_pThis->sigEvent((char*)message.topic().toLocal8Bit().toStdString().c_str(), (char*)message.payload().toStdString().c_str());
+    //qDebug() << "recv" << message.topic() << message.payload();
+}
+
+bool MQTTAgent::Connect(tagSetup ts)
+{
+    m_pMqtt = new QMQTT::Client();
+
+    connect(m_pMqtt, &QMQTT::Client::connected, []{
+        qDebug()<< __FILE__ << __LINE__ << "connected";
+    });
+    connect(m_pMqtt, &QMQTT::Client::disconnected, []{
+        qDebug()<< __FILE__ << __LINE__ << "disconnect";
+    });
+    connect(m_pMqtt, &QMQTT::Client::received, this, &onReceived);
+
+    m_pMqtt->setHostName(ts.addr.c_str());
+    //m_pMqtt->setHostName("mq.tongxinmao.com"); //18830
+    //m_pMqtt->setHostName("www.kjxry.cn");
+    m_pMqtt->setPort(ts.port);
+    m_pMqtt->setKeepAlive(60);
+    m_pMqtt->setClientId("C001"); //唯一id, 相同id不能同时连接
+    m_pMqtt->setUsername(ts.user.c_str());
+    m_pMqtt->setPassword(ts.password.c_str());
+
+    m_pMqtt->setAutoReconnect(true); //开启自动重连
+    m_pMqtt->setCleanSession(true); //非持久化连接，上线时，将不再关心之前所有的订阅关系以及离线消息
+
+    m_pMqtt->setVersion(QMQTT::V3_1_1);
+    qDebug()<< __FILE__ << __LINE__<< "ver" << m_pMqtt->version();
+
+    m_pMqtt->connectToHost();
+
+    return true;
+}
+
+bool MQTTAgent::Subscribe(std::list<std::string> topics)
+{
+    bool ret = true;
+    std::list<std::string>::iterator itr;
+    for (itr = topics.begin(); itr != topics.end(); ++itr)
+    {
+        m_pMqtt->subscribe(itr->c_str(), 1);
+    }
+
+	return ret;
+}
+
+void MQTTAgent::Publish(std::string szKey,QVariant v)
+{
+    std::string szContent;
+    switch( v.type() )
+    {
+    case QMetaType::Bool:
+        szContent = v.toBool() ? "true":"false";
+        break;
+    case QMetaType::Int:
+        szContent = std::to_string(v.toInt());
+        break;
+    case QMetaType::UInt:
+        szContent = std::to_string(v.toUInt());
+        break;
+    case QMetaType::LongLong:
+        szContent = std::to_string(v.toLongLong());
+        break;
+    case QMetaType::ULongLong:
+        szContent = std::to_string(v.toULongLong());
+        break;
+    case QMetaType::Double:
+        szContent = std::to_string(v.toDouble());
+        break;
+    case QMetaType::QString:
+        szContent = v.toString().toLocal8Bit().toStdString();
+        break;
+    }
+
+    QMQTT::Message message;
+    message.setTopic(szKey.c_str());
+    message.setPayload(szContent.c_str());
+    message.setRetain(true); //保留最后一条数据
+    m_pMqtt->publish(message);
+}
+
+QStringList MQTTAgent::hvals(QString k)
+{
+    return QStringList();
+}

MQTTAgent/MQTTAgent.h

@@ -0,0 +1,41 @@
+#pragma once
+
+//#include "BaseModule.h"
+//#include "libaray_symbols.h"
+#include "MWareInterface.h"
+#include <QtNetwork/QUdpSocket>
+#include <QtCore/QVariant>
+#include <QtCore/QObject>
+
+class MQTTAgent : public QObject,public EventSubInterface//,public BaseModule
+{
+	Q_OBJECT
+public:
+    MQTTAgent();
+    ~MQTTAgent();
+
+private:
+//    virtual void Setup(ModuleInfo mi) {};
+//    virtual void regConsumer(DataConsumer* pDC) {};
+//    virtual bool isInheritedFrom(std::string tp) {return false;};
+//    virtual void OnSubData(std::string name,std::string val) {};
+
+private:
+
+signals:
+    void sigEvent(char* topic, char* msg);
+
+private:
+    //void SubCB(char* topic, char* msg);
+
+public:
+    bool Connect(tagSetup ts);
+    bool Subscribe(std::list<std::string> lstTopic);
+    void Publish(std::string,QVariant v);
+    QStringList hvals(QString k);
+};
+
+//extern "C" {//一定要添加上
+//MQTTAgent_EXPORT BaseModule* instance();
+//MQTTAgent_EXPORT void destroy(BaseModule*);
+//}

MQTTAgent/MQTTAgent.pro

@@ -0,0 +1,38 @@
+QT       += core gui network
+
+greaterThan(QT_MAJOR_VERSION, 4): QT += widgets
+
+CONFIG += c++17
+
+TARGET = MQTTAgent
+TEMPLATE = lib
+
+unix{
+}
+else{
+#DESTDIR = $$PWD/../bin/plugins
+DESTDIR = $$PWD/../lib
+}
+
+INCLUDEPATH += ../include
+#LIBS += -L$$PWD/lib/ -lQt5Qmqtt
+
+SOURCES += \
+    MQTTAgent.cpp \
+#    main.cpp \
+#    widget.cpp
+
+HEADERS += \
+    MQTTAgent.h \
+#    qmqtt.h \
+#    widget.h
+
+FORMS += \
+#    widget.ui
+
+LIBS += $$PWD/../bin/Qt5Qmqtt.dll
+
+# Default rules for deployment.
+#qnx: target.path = /tmp/$${TARGET}/bin
+#else: unix:!android: target.path = /opt/$${TARGET}/bin
+#!isEmpty(target.path): INSTALLS += target

MQTTAgent/main.cpp

@@ -0,0 +1,11 @@
+#include "widget.h"
+
+#include <QApplication>
+
+int main(int argc, char *argv[])
+{
+    QApplication a(argc, argv);
+    Widget w;
+    w.show();
+    return a.exec();
+}

MQTTAgent/qmqtt.h

@@ -0,0 +1,38 @@
+/*
+ * qmqtt.h - qmqtt library heaer
+ *
+ * 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_H
+#define QMQTT_H
+
+#include <qmqtt_message.h>
+#include <qmqtt_client.h>
+
+#endif // QMQTT_H

MQTTAgent/qmqtt_client.h

@@ -0,0 +1,286 @@
+/*
+ * qmqtt_client.h - qmqtt client header
+ *
+ * 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_CLIENT_H
+#define QMQTT_CLIENT_H
+
+#include <qmqtt_global.h>
+
+#include <QObject>
+#include <QString>
+#include <QHostAddress>
+#include <QByteArray>
+#include <QAbstractSocket>
+#include <QScopedPointer>
+#include <QList>
+
+#ifdef QT_WEBSOCKETS_LIB
+#include <QWebSocket>
+#endif // QT_WEBSOCKETS_LIB
+
+#ifndef QT_NO_SSL
+#include <QSslConfiguration>
+QT_FORWARD_DECLARE_CLASS(QSslError)
+#endif // QT_NO_SSL
+
+#ifndef Q_ENUM_NS
+#define Q_ENUM_NS(x)
+#endif // Q_ENUM_NS
+
+namespace QMQTT {
+#if (QT_VERSION >= QT_VERSION_CHECK(5, 8, 0))
+Q_MQTT_EXPORT Q_NAMESPACE
+#endif
+
+static const quint8 LIBRARY_VERSION_MAJOR = 0;
+static const quint8 LIBRARY_VERSION_MINOR = 3;
+static const quint8 LIBRARY_VERSION_REVISION = 1;
+//static const char* LIBRARY_VERSION = "0.3.1";
+
+enum MQTTVersion
+{
+    V3_1_0 = 3,
+    V3_1_1 = 4
+};
+Q_ENUM_NS(MQTTVersion)
+
+enum ConnectionState
+{
+    STATE_INIT = 0,
+    STATE_CONNECTING,
+    STATE_CONNECTED,
+    STATE_DISCONNECTED
+};
+Q_ENUM_NS(ConnectionState)
+
+enum ClientError
+{
+    UnknownError = 0,
+    SocketConnectionRefusedError,
+    SocketRemoteHostClosedError,
+    SocketHostNotFoundError,
+    SocketAccessError,
+    SocketResourceError,
+    SocketTimeoutError,
+    SocketDatagramTooLargeError,
+    SocketNetworkError,
+    SocketAddressInUseError,
+    SocketAddressNotAvailableError,
+    SocketUnsupportedSocketOperationError,
+    SocketUnfinishedSocketOperationError,
+    SocketProxyAuthenticationRequiredError,
+    SocketSslHandshakeFailedError,
+    SocketProxyConnectionRefusedError,
+    SocketProxyConnectionClosedError,
+    SocketProxyConnectionTimeoutError,
+    SocketProxyNotFoundError,
+    SocketProxyProtocolError,
+    SocketOperationError,
+    SocketSslInternalError,
+    SocketSslInvalidUserDataError,
+    SocketTemporaryError,
+    MqttUnacceptableProtocolVersionError=1<<16,
+    MqttIdentifierRejectedError,
+    MqttServerUnavailableError,
+    MqttBadUserNameOrPasswordError,
+    MqttNotAuthorizedError,
+    MqttNoPingResponse
+};
+Q_ENUM_NS(ClientError)
+
+class ClientPrivate;
+class Message;
+class Frame;
+class NetworkInterface;
+
+class Q_MQTT_EXPORT Client : public QObject
+{
+    Q_OBJECT
+    Q_PROPERTY(quint16 _port READ port WRITE setPort)
+    Q_PROPERTY(QHostAddress _host READ host WRITE setHost)
+    Q_PROPERTY(QString _hostName READ hostName WRITE setHostName)
+    Q_PROPERTY(QString _clientId READ clientId WRITE setClientId)
+    Q_PROPERTY(QString _username READ username WRITE setUsername)
+    Q_PROPERTY(QByteArray _password READ password WRITE setPassword)
+    Q_PROPERTY(quint16 _keepAlive READ keepAlive WRITE setKeepAlive)
+    Q_PROPERTY(MQTTVersion _version READ version WRITE setVersion)
+    Q_PROPERTY(bool _autoReconnect READ autoReconnect WRITE setAutoReconnect)
+    Q_PROPERTY(int _autoReconnectInterval READ autoReconnectInterval WRITE setAutoReconnectInterval)
+    Q_PROPERTY(bool _cleanSession READ cleanSession WRITE setCleanSession)
+    Q_PROPERTY(QString _willTopic READ willTopic WRITE setWillTopic)
+    Q_PROPERTY(quint8 _willQos READ willQos WRITE setWillQos)
+    Q_PROPERTY(bool _willRetain READ willRetain WRITE setWillRetain)
+    Q_PROPERTY(QByteArray _willMessage READ willMessage WRITE setWillMessage)
+    Q_PROPERTY(ConnectionState _connectionState READ connectionState)
+#ifndef QT_NO_SSL
+    Q_PROPERTY(QSslConfiguration _sslConfiguration READ sslConfiguration WRITE setSslConfiguration)
+#endif // QT_NO_SSL
+
+public:
+    Client(const QHostAddress& host = QHostAddress::LocalHost,
+           const quint16 port = 1883,
+           QObject* parent = nullptr);
+
+#ifndef QT_NO_SSL
+    Client(const QString& hostName,
+           const quint16 port,
+           const QSslConfiguration& config,
+           const bool ignoreSelfSigned=false,
+           QObject* parent = nullptr);
+#endif // QT_NO_SSL
+
+    // This function is provided for backward compatibility with older versions of QMQTT.
+    // If the ssl parameter is true, this function will load a private key ('cert.key') and a local
+    // certificate ('cert.crt') from the current working directory. It will also set PeerVerifyMode
+    // to None. This may not be the safest way to set up an SSL connection.
+    Client(const QString& hostName,
+           const quint16 port,
+           const bool ssl,
+           const bool ignoreSelfSigned,
+           QObject* parent = nullptr);
+
+#ifdef QT_WEBSOCKETS_LIB
+    // Create a connection over websockets
+    Client(const QString& url,
+           const QString& origin,
+           QWebSocketProtocol::Version version,
+           bool ignoreSelfSigned = false,
+           QObject* parent = nullptr);
+
+#ifndef QT_NO_SSL
+    Client(const QString& url,
+           const QString& origin,
+           QWebSocketProtocol::Version version,
+           const QSslConfiguration& config,
+           const bool ignoreSelfSigned = false,
+           QObject* parent = nullptr);
+#endif // QT_NO_SSL
+#endif // QT_WEBSOCKETS_LIB
+
+    // for testing purposes only
+    Client(NetworkInterface* network,
+           const QHostAddress& host = QHostAddress::LocalHost,
+           const quint16 port = 1883,
+           QObject* parent = nullptr);
+
+    virtual ~Client();
+
+    QHostAddress host() const;
+    QString hostName() const;
+    quint16 port() const;
+    QString clientId() const;
+    QString username() const;
+    QByteArray password() const;
+    QMQTT::MQTTVersion version() const;
+    quint16 keepAlive() const;
+    bool cleanSession() const;
+    bool autoReconnect() const;
+    int autoReconnectInterval() const;
+    ConnectionState connectionState() const;
+    QString willTopic() const;
+    quint8 willQos() const;
+    bool willRetain() const;
+    QByteArray willMessage() const;
+
+    bool isConnectedToHost() const;
+#ifndef QT_NO_SSL
+    QSslConfiguration sslConfiguration() const;
+    void setSslConfiguration(const QSslConfiguration& config);
+#endif // QT_NO_SSL
+
+public Q_SLOTS:
+    void setHost(const QHostAddress& host);
+    void setHostName(const QString& hostName);
+    void setPort(const quint16 port);
+    void setClientId(const QString& clientId);
+    void setUsername(const QString& username);
+    void setPassword(const QByteArray& password);
+    void setVersion(const MQTTVersion version);
+    void setKeepAlive(const quint16 keepAlive);
+    void setCleanSession(const bool cleanSession);
+    void setAutoReconnect(const bool value);
+    void setAutoReconnectInterval(const int autoReconnectInterval);
+    void setWillTopic(const QString& willTopic);
+    void setWillQos(const quint8 willQos);
+    void setWillRetain(const bool willRetain);
+    void setWillMessage(const QByteArray& willMessage);
+
+    void connectToHost();
+    void disconnectFromHost();
+
+    void subscribe(const QString& topic, const quint8 qos = 0);
+    void unsubscribe(const QString& topic);
+
+    quint16 publish(const QMQTT::Message& message);
+
+#ifndef QT_NO_SSL
+    void ignoreSslErrors();
+    void ignoreSslErrors(const QList<QSslError>& errors);
+#endif // QT_NO_SSL
+
+Q_SIGNALS:
+    void connected();
+    void disconnected();
+    void error(const QMQTT::ClientError error);
+
+    void subscribed(const QString& topic, const quint8 qos = 0);
+    void unsubscribed(const QString& topic);
+    void published(const QMQTT::Message& message, quint16 msgid = 0);
+    void received(const QMQTT::Message& message);
+    void pingresp();
+#ifndef QT_NO_SSL
+    void sslErrors(const QList<QSslError>& errors);
+#endif // QT_NO_SSL
+
+protected Q_SLOTS:
+    void onNetworkConnected();
+    void onNetworkDisconnected();
+    void onNetworkReceived(const QMQTT::Frame& frame);
+    void onTimerPingReq();
+    void onPingTimeout();
+    void onNetworkError(QAbstractSocket::SocketError error);
+#ifndef QT_NO_SSL
+    void onSslErrors(const QList<QSslError>& errors);
+#endif // QT_NO_SSL
+
+protected:
+    QScopedPointer<ClientPrivate> d_ptr;
+
+private:
+    Q_DISABLE_COPY(Client)
+    Q_DECLARE_PRIVATE(Client)
+};
+
+} // namespace QMQTT
+
+Q_DECLARE_METATYPE(QMQTT::ClientError)
+
+#endif // QMQTT_CLIENT_H

MQTTAgent/qmqtt_message.h

@@ -0,0 +1,96 @@
+/*
+ * qmqtt_message.h - qmqtt message header
+ *
+ * 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_MESSAGE_H
+#define QMQTT_MESSAGE_H
+
+#include <qmqtt_global.h>
+
+#include <QMetaType>
+#include <QString>
+#include <QByteArray>
+#include <QSharedDataPointer>
+
+namespace QMQTT {
+
+class MessagePrivate;
+
+class Q_MQTT_EXPORT Message
+{
+public:
+    Message();
+    explicit Message(const quint16 id, const QString &topic, const QByteArray &payload,
+                     const quint8 qos = 0, const bool retain = false, const bool dup = false);
+    Message(const Message &other);
+    ~Message();
+
+    Message &operator=(const Message &other);
+#ifdef Q_COMPILER_RVALUE_REFS
+    inline Message &operator=(Message &&other) Q_DECL_NOTHROW
+    { swap(other); return *this; }
+#endif
+
+    bool operator==(const Message &other) const;
+    inline bool operator!=(const Message &other) const
+    { return !operator==(other); }
+
+    inline void swap(Message &other) Q_DECL_NOTHROW
+    { qSwap(d, other.d); }
+
+    quint16 id() const;
+    void setId(const quint16 id);
+
+    quint8 qos() const;
+    void setQos(const quint8 qos);
+
+    bool retain() const;
+    void setRetain(const bool retain);
+
+    bool dup() const;
+    void setDup(const bool dup);
+
+    QString topic() const;
+    void setTopic(const QString &topic);
+
+    QByteArray payload() const;
+    void setPayload(const QByteArray &payload);
+
+private:
+    QSharedDataPointer<MessagePrivate> d;
+};
+
+} // namespace QMQTT
+
+Q_DECLARE_SHARED(QMQTT::Message)
+
+Q_DECLARE_METATYPE(QMQTT::Message)
+
+#endif // QMQTT_MESSAGE_H