Buffering MQTT fonctionnel.

Otarcik_CAN/Sources/CloudLogging/MQTTClientWrapper.cpp

@@ -8,9 +8,10 @@ CMQTTClientWrapper::CMQTTClientWrapper()
 {
     connect(&mMQTTClient,&QMqttClient::stateChanged,this,&CMQTTClientWrapper::StateChanged);
     connect(&mMQTTClient,&QMqttClient::errorChanged,this,&CMQTTClientWrapper::MQTTClientError);
-    //connect(&mMQTTClient,&QMqttClient::messageSent,this,&CMQTTClientWrapper::MQTTMessageSent);
-    connect(&mMQTTClient,SIGNAL(messageSent(qint32)),this,SLOT(MQTTMessageSent(qint32)));
-    connect(&mMQTTClient,&QMqttClient::messageStatusChanged,this,&CMQTTClientWrapper::MQTTMessageStatusChanged);
+    connect(&mMQTTClient,&QMqttClient::messageSent,this,&CMQTTClientWrapper::MQTTMessageSent);
+//    connect(&mMQTTClient,SIGNAL(messageSent(qint32)),this,SLOT(MQTTMessageSent(qint32)));
+//    connect(&mMQTTClient,&QMqttClient::messageStatusChanged,this,&CMQTTClientWrapper::MQTTMessageStatusChanged);
+//    connect(&mMQTTClient,SIGNAL(messageStatusChanged(qint32,QMqtt::MessageStatus,QMqttMessageStatusProperties)),this,SLOT(MQTTMessageStatusChanged(qint32,QMqtt::MessageStatus,QMqttMessageStatusProperties)));
     mMQTTRefreshTimer = new QTimer;
     mMQTTRefreshTimer->setSingleShot(true);
     connect(mMQTTRefreshTimer,&QTimer::timeout,this,&CMQTTClientWrapper::MQTTSendTimerExpired);
@@ -275,7 +276,7 @@ void CMQTTClientWrapper::MQTTSendTimerExpired()
                 {
                     for(int i = 0; i < MessagesList->size(); i++)
                     {
-                        qint32 res = mMQTTClient.publish(MessagesList->at(i).mMessageTopic,MessagesList->at(i).mMessagePayload.toLocal8Bit(),0,true);
+                        qint32 res = mMQTTClient.publish(MessagesList->at(i).mMessageTopic,MessagesList->at(i).mMessagePayload.toLocal8Bit(),1,true);
                         QString LogMsg = QString("Envoi d'un message MQTT.  Topic: %1  Payload: %2 Result: %3").arg(MessagesList->at(i).mMessageTopic).arg(MessagesList->at(i).mMessagePayload).arg(res);
                         CGeneralMessagesLogDispatcher::instance()->AddLogMessage(LogMsg,"CMQTTClientWrapper",true,3);
 
@@ -363,7 +364,7 @@ int CMQTTClientWrapper::SendNextQueueMessage()
         return RET_GENERAL_ERROR;
     }
     CMQTTMessage *Msg = mMQTTMessagesQueue.takeFirst();
-    mMQTTClient.publish(Msg->mMessageTopic,Msg->mMessagePayload.toLocal8Bit(),0,true);
+    mMQTTClient.publish(Msg->mMessageTopic,Msg->mMessagePayload.toLocal8Bit(),1,true);
     QString LogMsg = QString("Envoi d'un message MQTT provenant du buffer.  Topic: %1  Payload: %2 FIFO Size: %3").arg(Msg->mMessageTopic).arg(Msg->mMessagePayload).arg(mMQTTMessagesQueue.size());
     CGeneralMessagesLogDispatcher::instance()->AddLogMessage(LogMsg,"CMQTTClientWrapper",true,3);
 
@@ -378,12 +379,23 @@ int CMQTTClientWrapper::SendNextQueueMessage()
         CGeneralMessagesLogDispatcher::instance()->AddLogMessage("Tous les messages MQTT de la FIFO ont été envoyés au serveur, passage en mode de transmission normal","CMQTTClientWrapper",true,2);
     }
 
+    UpdateGUIBufferingStatus();
+
     return RET_OK;
 }
 
 int CMQTTClientWrapper::UpdateGUIBufferingStatus()
 {
-    //mProgramPtr->UpdateMQTTBufferingStatusRequest(mBufferingModeText,mCircularBufferStatusText,mRemainingBufferingTimeText);
+    if(mMessagesQueueMode == MQTT_QUEUE_MSG_MODE)
+    {
+        mBufferingModeText = "Buffering";
+    }
+    else
+    {
+        mBufferingModeText = "Transmitting";
+    }
+    mCircularBufferStatusText = QString("%1").arg(mMQTTMessagesQueue.size());
+    mProgramPtr->UpdateMQTTBufferingStatusRequest(mBufferingModeText,mCircularBufferStatusText,0);
     return RET_OK;
 }
 
@@ -396,7 +408,7 @@ void CMQTTClientWrapper::MQTTClientError(QMqttClient::ClientError error)
 void CMQTTClientWrapper::MQTTMessageSent(qint32 MsgID)
 {
     Q_UNUSED(MsgID)
-    qDebug("MQTT msg ID %d sent",MsgID);
+ //   qDebug("MQTT msg ID %d sent",MsgID);
 #ifndef MQTT_USE_TIMER_TO_FLUSH_QUEUE
     if(mMessagesQueueMode != MQTT_QUEUE_MSG_MODE)
     {
@@ -408,7 +420,7 @@ void CMQTTClientWrapper::MQTTMessageSent(qint32 MsgID)
         mMessagesQueueMode = MQTT_TRANSMIT_MSG_MODE;
         return;
     }
-    CGeneralMessagesLogDispatcher::instance()->AddLogMessage("Message MQTT de la FIFO transmis, transmission du prochain...","CMQTTClientWrapper",true,1);
+    CGeneralMessagesLogDispatcher::instance()->AddLogMessage("Message MQTT de la FIFO transmis, transmission du prochain...","CMQTTClientWrapper",true,3);
     SendNextQueueMessage();
 
 #endif
@@ -425,7 +437,7 @@ CMQTTMessage CMQTTClientWrapper::GetSystemStatusMessage()
     {
         CComputerBoardState tmp = mCPUInterface->GetComputerBoardState();
         QString MsgPayload = "{";
-        QString MsgTopic = "GeneralPCStatus";
+        QString MsgTopic = mMQTTParams.mMQTTTopicPrefix + "GeneralPCStatus";
 
         //Millisecs to EPOCH timestamp format
         qint64 TimeStamp = QDateTime::currentDateTime().toMSecsSinceEpoch();

Otarcik_CAN/Sources/LANDevicePresenceConfig.h

@@ -1,6 +1,7 @@
 #ifndef LANDEVICEPRESENCECONFIG_H
 #define LANDEVICEPRESENCECONFIG_H
 #include <QDataStream>
+#include <QString>
 
 class CLANDevicePresenceConfig
 {

Otarcik_CAN/Sources/defines.h

@@ -8,7 +8,7 @@
 
 #define MQTT_CLIENT_RECONNECT_TIMEOUT 5000  //Time in ms between reconnect attemps to MQTT broker when connection is lost
 #define MQTT_CLIENT_MSG_QUEUE_FLUSH_TIMEOUT 500 //The delay in ms between the TX messages when flushing the msg queue after a server reconnection
-#define MQTT_CLIENT_MSG_QUEUE_SIZE 500 //The maximum number of messages to be queued.  Older messages are lost when adding a new one in a full queue
+#define MQTT_CLIENT_MSG_QUEUE_SIZE 5000 //The maximum number of messages to be queued.  Older messages are lost when adding a new one in a full queue
 #define MQTT_TIMESTAMP_IN_PAYLOAD   //Defining this adds the local timestamp to each MQTT payloads.
 //#define MQTT_USE_TIMER_TO_FLUSH_QUEUE //Use a timer based transmission of MQTT msgs while emtpying the queue instead of the "MessageSent" event.
 

Otarcik_CAN/Sources/main.cpp

@@ -1,6 +1,7 @@
 #include <QApplication>
 #include "OtarcikCan.h"
 #include "windows.h"
+#include <QTextCodec>
 
 
 int main(int argc, char *argv[])