Implement VeQItemMqtt and VeQItemMqttProducer

examples/mqtt_producer/main.cpp

@@ -0,0 +1,73 @@
+#include <QGuiApplication>
+#include <QScopedPointer>
+#include <QQmlEngine>
+#include <QQmlComponent>
+#include <QQuickWindow>
+#include <QtDebug>
+
+#ifdef Q_OS_WASM
+#include <emscripten/val.h>
+#include <emscripten.h>
+#include <QWebSocketProtocol>
+#include <QUrl>
+#include <QUrlQuery>
+#else
+#include <QHostAddress>
+#endif
+
+#include <veutil/qt/ve_qitems_mqtt.hpp>
+#include <veutil/qt/ve_quick_item.hpp>
+
+int main(int argc, char *argv[])
+{
+	qmlRegisterType<VeQuickItem>("Victron.VeUtil", 1, 0, "VeQuickItem");
+
+	QGuiApplication app(argc, argv);
+	QQmlEngine engine;
+	QQmlComponent component(&engine, QUrl(QStringLiteral("qrc:/main.qml")));
+
+	bool running = false;
+	VeQItemMqttProducer producer(VeQItems::getRoot(), QStringLiteral("mqtt"), QStringLiteral("example"));
+	QObject::connect(&producer, &VeQItemMqttProducer::aboutToConnect,
+		[&producer] {
+			producer.setCredentials(QString(), QString());
+			producer.continueConnect();
+		});
+	QObject::connect(&producer, &VeQItemMqttProducer::connectionStateChanged,
+		[&running, &producer] {
+			if (producer.connectionState() == VeQItemMqttProducer::Ready) {
+				running = true;
+			}
+		});
+
+	QTimer::singleShot(10000,
+		[&app, &running] {
+			if (!running) {
+				qWarning() << "MQTT producer never became ready, unable to connect?";
+				app.quit();
+			}
+		});
+
+	QQuickWindow *window = qobject_cast<QQuickWindow*>(component.create());
+	if (!window) {
+		qWarning() << "Unable to create main.qml component! " << component.errors();
+		return 1;
+	}
+
+	window->showFullScreen();
+#ifdef Q_OS_WASM
+	emscripten::val webLocation = emscripten::val::global("location");
+	const QUrl webLocationUrl = QUrl(QString::fromStdString(webLocation["href"].as<std::string>()));
+	const QUrlQuery query(webLocationUrl);
+	const QString ip(query.queryItemValue("mqtt"));
+	if (ip.isEmpty()) {
+		qWarning() << "Please specify the broker's IP address via mqtt query parameter, e.g. ?mqtt=192.168.5.96";
+	} else {
+		const QString mqttUrl(QStringLiteral("ws://%1:9001/").arg(ip)); // e.g.: "ws://192.168.5.96:9001/"
+		producer.open(QUrl(mqttUrl), QMqttClient::MQTT_3_1);
+	}
+#else
+	producer.open(QHostAddress::LocalHost, 1883);
+#endif
+	return app.exec();
+}

examples/mqtt_producer/main.qml

@@ -0,0 +1,26 @@
+import QtQuick
+import QtQuick.Window
+import Victron.VeUtil
+
+Window {
+    id: root
+
+    title: "Testing MQTT VeQItem Support"
+    color: "lightsteelblue"
+
+    width: 600
+    height: 600
+
+    Text {
+        anchors.centerIn: parent
+        text: "Voltage = " + batteryVoltage.value
+        font.family: "Helvetica"
+        font.pointSize: 24
+        color: "black"
+    }
+
+    VeQuickItem {
+        id: batteryVoltage
+        uid: "mqtt/system/0/Dc/Battery/Voltage"
+    }
+}

examples/mqtt_producer/mqtt_producer.pro

@@ -0,0 +1,16 @@
+TARGET = mqtt_producer
+TEMPLATE = app
+QT += core network gui widgets qml quick mqtt
+
+SOURCES += main.cpp
+RESOURCES += qml.qrc
+
+# uncomment the following for websockets support
+DEFINES += MQTT_WEBSOCKETS_ENABLED
+QT += websockets
+
+CONFIG += ve-qitems-mqtt
+VEUTIL = ../..
+INCLUDEPATH += $${VEUTIL}/inc
+include($${VEUTIL}/src/qt/veqitem.pri)
+include($${VEUTIL}/common.pri)

examples/mqtt_producer/qml.qrc

@@ -0,0 +1,6 @@
+<RCC>
+    <qresource prefix="/">
+        <file>main.qml</file>
+    </qresource>
+</RCC>
+

inc/veutil/qt/ve_qitem.hpp

@@ -98,7 +98,7 @@ class VE_QITEM_EXPORT VeQItem : public QObject
 	 * will free the whole tree. Its only usefull for items which are not part of
 	 * a tree, and that is never the case. So lets deprecate that constructor.
 	 */
-	explicit VeQItem(VeQItemProducer *producer, QObject *parent) __attribute((deprecated));
+	Q_DECL_DEPRECATED explicit VeQItem(VeQItemProducer *producer, QObject *parent);
 	explicit VeQItem(VeQItemProducer *producer);
 	~VeQItem();
 
@@ -417,7 +417,7 @@ class VE_QITEM_EXPORT VeQItemLocal : public VeQItem
 	Q_OBJECT
 
 public:
-	explicit VeQItemLocal(VeQItemProducer *producer, QObject *parent) __attribute((deprecated));
+	Q_DECL_DEPRECATED explicit VeQItemLocal(VeQItemProducer *producer, QObject *parent);
 	explicit VeQItemLocal(VeQItemProducer *producer);
 
 	int setValue(QVariant const &value) override

inc/veutil/qt/ve_qitems_mqtt.hpp

@@ -0,0 +1,193 @@
+#pragma once
+
+#include <veutil/qt/ve_qitem.hpp>
+
+#include <QMqttClient>
+#include <QMqttSubscription>
+
+#include <QPointer>
+#include <QHostAddress>
+#include <QTimer>
+#include <QSet>
+#include <QString>
+#include <QByteArray>
+#include <QUrl>
+
+#ifdef MQTT_WEBSOCKETS_ENABLED
+#include <QWebSocket>
+#include <QWebSocketProtocol>
+#include <QIODevice>
+class WebSocketDevice;
+#endif // MQTT_WEBSOCKETS_ENABLED
+
+class VeQItemMqttProducer;
+
+// Implementation of a VeQItem over MQTT
+class VeQItemMqtt : public VeQItem
+{
+	Q_OBJECT
+
+public:
+	VeQItemMqtt(VeQItemMqttProducer *producer);
+
+	int setValue(QVariant const &value) override;
+
+    QVariant getValue() override;
+    QVariant getValue(bool force) override;
+
+protected:
+	void setParent(QObject *parent) override;
+
+private:
+	VeQItemMqttProducer *mqttProducer() const;
+};
+
+class VeQItemMqttProducer : public VeQItemProducer
+{
+	Q_OBJECT
+	Q_PROPERTY(HeartbeatState heartbeatState READ heartbeatState NOTIFY heartbeatStateChanged)
+	Q_PROPERTY(ConnectionState connectionState READ connectionState NOTIFY connectionStateChanged)
+	Q_PROPERTY(QMqttClient::ClientError error READ error NOTIFY errorChanged)
+	Q_PROPERTY(QString portalId READ portalId WRITE setPortalId NOTIFY portalIdChanged)
+
+public:
+
+	enum ConnectionState {
+		Idle,
+		Connecting,
+		Connected,
+		Initializing,
+		Ready,
+		Disconnected,
+		Reconnecting,
+		Failed
+	};
+	Q_ENUM(ConnectionState)
+
+	// The heartbeat state tracks whether the CerboGX is actively alive
+	// and writing heartbeat timestamp updates to the MQTT broker.
+	// We might be actively connected to the VRM broker, but the actual
+	// device itself might have gone offline and so the data is "not live".
+	enum HeartbeatState {
+		HeartbeatActive,
+		HeartbeatMissing,
+		HeartbeatInactive,
+	};
+	Q_ENUM(HeartbeatState)
+
+	VeQItemMqttProducer(VeQItem *root, const QString &id, const QString &clientIdPrefix, QObject *parent = nullptr);
+
+	VeQItem *createItem() override;
+
+	void open(const QHostAddress &host, int port);
+
+#ifdef MQTT_WEBSOCKETS_ENABLED
+	void open(
+		const QUrl &url,
+		QMqttClient::ProtocolVersion protocolVersion);
+#endif // MQTT_WEBSOCKETS_ENABLED
+
+	QMqttClient *mqttConnection() const;
+	void setCredentials(const QString &username, const QString &password);
+	bool publishValue(const QString &uid, const QVariant &value);
+	bool requestValue(const QString &uid);
+	HeartbeatState heartbeatState() const;
+	ConnectionState connectionState() const;
+	QMqttClient::ClientError error() const;
+
+	QString portalId() const;
+	void setPortalId(const QString &portalId);
+
+Q_SIGNALS:
+	void heartbeatStateChanged();
+	void connectionStateChanged();
+	void errorChanged();
+	void portalIdChanged();
+	void messageReceived(const QString &path, const QVariant &value);
+	void nullMessageReceived(const QString &path);
+	void aboutToConnect(); // client should handle this by calling setCredentials() and calling continueConnect
+
+public Q_SLOTS:
+	void continueConnect();
+
+private Q_SLOTS:
+	void onConnected();
+	void onDisconnected();
+	void onErrorChanged(QMqttClient::ClientError error);
+	void onStateChanged(QMqttClient::ClientState state);
+	void onMessageReceived(const QByteArray &message, const QMqttTopicName &topic);
+	void onSubscriptionMessageReceived(const QMqttMessage &message);
+	void doKeepAlive(bool suppressRepublish = false);
+
+private:
+	void setHeartbeatState(HeartbeatState heartbeatState);
+	void setConnectionState(ConnectionState connectionState);
+	void setError(QMqttClient::ClientError error);
+	void parseMessage(const QString &path, const QByteArray &message);
+
+	QTimer *mKeepAliveTimer;
+	QTimer *mHeartBeatTimer;
+	QTimer *mReadyStateFallbackTimer;
+	QMqttClient *mMqttConnection;
+	QPointer<QMqttSubscription> mMqttSubscription;
+#ifdef MQTT_WEBSOCKETS_ENABLED
+	WebSocketDevice *mWebSocket = nullptr;
+#endif // MQTT_WEBSOCKETS_ENABLED
+	QString mClientId;
+	QString mPortalId;
+	QUrl mUrl;
+	QString mHostName;
+	int mPort;
+	HeartbeatState mHeartbeatState;
+	ConnectionState mConnectionState;
+	const int mReconnectAttemptIntervals[6] = { 250, 1000, 2000, 5000, 10000, 30000 };
+	quint16 mAutoReconnectAttemptCounter;
+	const quint16 mAutoReconnectMaxAttempts;
+	QMqttClient::ClientError mError;
+	QMqttClient::ProtocolVersion mProtocolVersion;
+	int mMissedHeartbeats;
+	bool mReceivedMessage;
+	bool mIsVrmBroker;
+};
+
+#ifdef MQTT_WEBSOCKETS_ENABLED
+class WebSocketDevice : public QIODevice
+{
+	Q_OBJECT
+	Q_PROPERTY(QUrl url READ url WRITE setUrl NOTIFY urlChanged)
+	Q_PROPERTY(QByteArray protocol READ protocol WRITE setProtocol NOTIFY protocolChanged)
+
+public:
+	WebSocketDevice(QObject *parent = nullptr);
+
+	void setUrl(const QUrl &url);
+	QUrl url() const;
+
+	void setProtocol(const QByteArray &protocol);
+	QByteArray protocol() const;
+
+	bool isValid() const;
+	bool open(QIODeviceBase::OpenMode mode) override;
+	void close() override;
+	qint64 readData(char *data, qint64 maxSize) override;
+	qint64 writeData(const char *data, qint64 maxSize) override;
+	qint64 bytesAvailable() const override;
+	bool isSequential() const override;
+
+Q_SIGNALS:
+	void urlChanged();
+	void protocolChanged();
+	void connected();
+	void disconnected();
+
+public Q_SLOTS:
+	void onBinaryMessageReceived(const QByteArray &message);
+
+private:
+	QUrl mUrl;
+	QByteArray mProtocol;
+	QByteArray mData;
+	QWebSocket mWebSocket;
+};
+#endif // MQTT_WEBSOCKETS_ENABLED
+

inc/veutil/qt/ve_quick_item.hpp

@@ -38,6 +38,7 @@ class VE_QITEM_EXPORT VeQuickItem : public QObject {
 	Q_PROPERTY(double defaultSourceMin READ getDefaultSourceMin NOTIFY defaultMinChanged)
 	Q_PROPERTY(double defaultSourceMax READ getDefaultSourceMax NOTIFY defaultMaxChanged)
 	Q_PROPERTY(bool isSetting READ getIsSetting NOTIFY isSettingChanged WRITE setIsSetting)
+	Q_PROPERTY(bool isValid READ getIsValid NOTIFY isValidChanged)
 	Q_PROPERTY(bool invalidate READ getInvalidate NOTIFY invalidateChanged WRITE setInvalidate)
 
 public:
@@ -47,6 +48,7 @@ class VE_QITEM_EXPORT VeQuickItem : public QObject {
 		mMaxInDisplayUnits(0),
 		mMinInDisplayUnits(0),
 		mIsSetting(isSetting),
+		mIsValid(0),
 		mInvalidate(1)
 	{
 		setUid("");
@@ -147,6 +149,8 @@ class VE_QITEM_EXPORT VeQuickItem : public QObject {
 	bool getIsSetting() { return mIsSetting; }
 	void setIsSetting(bool value);
 
+	bool getIsValid() const { return mIsValid; }
+
 signals:
 	void defaultChanged();
 	void minChanged();
@@ -159,6 +163,7 @@ class VE_QITEM_EXPORT VeQuickItem : public QObject {
 	void uidChanged();
 	void valueChanged();
 	void stateChanged();
+	void isValidChanged();
 
 	void unitChanged();
 	void decimalsChanged();
@@ -187,7 +192,11 @@ protected slots:
 	}
 
 	void onValueChanged() {
+		const bool prevIsValid = mIsValid;
+		mIsValid = getValue().isValid();
 		emit valueChanged();
+		if (prevIsValid != mIsValid)
+			emit isValidChanged();
 		if (mTextMode == TextMode::Format)
 			emit textChanged();
 	}
@@ -206,6 +215,7 @@ protected slots:
 	uint32_t mMaxInDisplayUnits:1;
 	uint32_t mMinInDisplayUnits:1;
 	uint32_t mIsSetting:1;
+	uint32_t mIsValid:1;
 	uint32_t mIsAllocated:1;
 	uint32_t mInvalidate:1;