Service MQTT message queue asynchronously

victronenergy · Feb 23, 2023 · d8250e9 · d8250e9
1 parent 887e9d9
commit d8250e9
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 5 deletions.
diff --git a/inc/veutil/qt/ve_qitems_mqtt.hpp b/inc/veutil/qt/ve_qitems_mqtt.hpp
@@ -7,6 +7,9 @@
 #include <QHostAddress>
 #include <QTimer>
 #include <QSet>
+#include <QQueue>
+#include <QString>
+#include <QByteArray>
 
 #ifdef MQTT_WEBSOCKETS_ENABLED
 #include <QUrl>
@@ -93,6 +96,7 @@ private Q_SLOTS:
 	void setConnectionState(ConnectionState connectionState);
 	void setError(QMqttClient::ClientError error);
 	void parseMessage(const QString &path, const QByteArray &message);
+	void serviceQueue();
 
 	QTimer mKeepAliveTimer;
 	QMqttClient *mMqttConnection;
@@ -101,6 +105,7 @@ private Q_SLOTS:
 #endif // MQTT_WEBSOCKETS_ENABLED
 	QString mClientId;
 	QString mPortalId;
+	QQueue<QPair<QString, QByteArray> > mMessageQueue;
 	ConnectionState mConnectionState;
 	const int mReconnectAttemptIntervals[6] = { 250, 1000, 2000, 5000, 10000, 30000 };
 	quint16 mAutoReconnectAttemptCounter;

diff --git a/src/qt/ve_qitems_mqtt.cpp b/src/qt/ve_qitems_mqtt.cpp
@@ -223,17 +223,31 @@ void VeQItemMqttProducer::onMessageReceived(const QByteArray &message, const QMq
 				// ignore keepalive topic.
 			} else {
 				// we have a topic message which we need to expose via VeQItem.
-				// invoke the parse method via a QueuedConnection to ensure
+				// service the queue via a QueuedConnection to ensure
 				// that we don't block the UI.
 				const QString path = topicName.mid(notificationPrefix.size() + 1);
-				QMetaObject::invokeMethod(this, [this, path, message] {
-					parseMessage(path, message);
-				}, Qt::QueuedConnection);
+				mMessageQueue.enqueue(QPair<QString, QByteArray>(path, message));
+				if (mMessageQueue.size() == 1) {
+					QMetaObject::invokeMethod(this, [this] {
+						serviceQueue();
+					}, Qt::QueuedConnection);
+				}
 			}
 		}
 	}
 }
 
+void VeQItemMqttProducer::serviceQueue()
+{
+	const QPair<QString, QByteArray> message = mMessageQueue.dequeue();
+	parseMessage(message.first, message.second);
+	if (mMessageQueue.size()) {
+		QMetaObject::invokeMethod(this, [this] {
+			serviceQueue();
+		}, Qt::QueuedConnection);
+	}
+}
+
 void VeQItemMqttProducer::parseMessage(const QString &path, const QByteArray &message)
 {
 	VeQItemMqtt *item = qobject_cast<VeQItemMqtt*>(mProducerRoot->itemGetOrCreate(path, true, true));
@@ -258,7 +272,6 @@ void VeQItemMqttProducer::doKeepAlive()
 	if (mMqttConnection
 			&& mMqttConnection->state() == QMqttClient::Connected
 			&& !mPortalId.isEmpty()) {
-		// TODO: just specific topics?  Or all topics as done here?
 		setConnectionState(Ready);
 		mMqttConnection->publish(QMqttTopicName(QStringLiteral("R/%1/keepalive").arg(mPortalId)), QByteArray());
 	}