No longer service MQTT message queue asynchronously

Qt For WebAssembly doesn't process the events in a timely manner
(unless the event queue is kicked by a timer event), so abandon
the asynchronous queue.  This may be fixed in a future version of
Qt, at which point in time it may be beneficial to revert this
commit, if stutters/UI stalls are noticed when receiving large
batches of MQTT notifications.

inc/veutil/qt/ve_qitems_mqtt.hpp

@@ -7,7 +7,6 @@
 #include <QHostAddress>
 #include <QTimer>
 #include <QSet>
-#include <QQueue>
 #include <QString>
 #include <QByteArray>
 #include <QUrl>
@@ -101,7 +100,6 @@ 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;
@@ -113,7 +111,6 @@ private Q_SLOTS:
 	QUrl mUrl;
 	QString mHostName;
 	int mPort;
-	QQueue<QPair<QString, QByteArray> > mMessageQueue;
 	ConnectionState mConnectionState;
 	const int mReconnectAttemptIntervals[6] = { 250, 1000, 2000, 5000, 10000, 30000 };
 	quint16 mAutoReconnectAttemptCounter;

src/qt/ve_qitems_mqtt.cpp

@@ -204,7 +204,6 @@ void VeQItemMqttProducer::onDisconnected()
 {
 	setConnectionState(Disconnected);
 	mKeepAliveTimer.stop();
-	mMessageQueue.clear();
 	mReceivedMessage = false;
 
 	if (mAutoReconnectAttemptCounter < mAutoReconnectMaxAttempts) {
@@ -268,15 +267,8 @@ void VeQItemMqttProducer::onMessageReceived(const QByteArray &message, const QMq
 				// ignore keepalive topic.
 			} else {
 				// we have a topic message which we need to expose via VeQItem.
-				// service the queue via a QueuedConnection to ensure
-				// that we don't block the UI.
 				const QString path = topicName.mid(notificationPrefix.size() + 1);
-				mMessageQueue.enqueue(QPair<QString, QByteArray>(path, message));
-				if (mMessageQueue.size() == 1) {
-					QMetaObject::invokeMethod(this, [this] {
-						serviceQueue();
-					}, Qt::QueuedConnection);
-				}
+				parseMessage(path, message);
 			}
 		}
 
@@ -288,17 +280,6 @@ void VeQItemMqttProducer::onMessageReceived(const QByteArray &message, const QMq
 	}
 }
 
-void VeQItemMqttProducer::serviceQueue()
-{
-	if (mMessageQueue.size()) {
-		const QPair<QString, QByteArray> message = mMessageQueue.dequeue();
-		parseMessage(message.first, message.second);
-		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));