Index: CloudConnect/CloudConnectController.cpp =================================================================== diff -u -r6d2ef8c97f4bb34204e95811839b3995000c47c1 -r59b4c22f45a1d098064a886452769204e90cfb4b --- CloudConnect/CloudConnectController.cpp (.../CloudConnectController.cpp) (revision 6d2ef8c97f4bb34204e95811839b3995000c47c1) +++ CloudConnect/CloudConnectController.cpp (.../CloudConnectController.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -10,62 +10,145 @@ * \date (original) 24-May-2026 * */ +#include #include #include #include #include "CloudConnectController.h" #include "LeahiMsgProtoUtils.h" +#include // TODO: temporary for capture protobuf +#include // TODO: temporary for capture protobuf +#include // TODO: temporary for capture protobuf + /*! * \brief CloudConnectController::CloudConnectController - * \details Constructor. Starts the CAN interface and wires the frame→dispatcher→ - * completed-message pipeline. + * \details Constructor. Wires the frame→dispatcher→completed-message pipeline. + * The CAN device itself is started by startCan() once this object is + * on its worker thread. * \param configPath - path to the settings file * \param msgHandlingPath - path to the message handling INI * \param parent - optional QObject parent + * \note Must be constructed on the main thread, but later moved to a worker thread. + * _canInterface is parented to this object so it migrates with it. */ -CloudConnectController::CloudConnectController(const QString &configPath, const QString &msgHandlingPath, QObject *parent) : +CloudConnectController::CloudConnectController(const QString &configPath, const QString &msgHandlingPath, + QObject *parent) : QObject(parent), _settings(configPath, QSettings::IniFormat), - _canInterface(), - _canThread(this), + _canInterface(this), _dispatcher(this), - _appServer(this) + _publisher(this) { loadMsgHandling(msgHandlingPath); - _canInterface.init(_canThread); connect(&_canInterface, &Can::CanInterface::didFrameReceive, this, &CloudConnectController::onFrameReceive); connect(&_dispatcher, &Can::MessageDispatcher::didActionReceive, this, &CloudConnectController::onMessageReceive); } /*! * \brief CloudConnectController::~CloudConnectController - * \details Destructor. Stops and joins the CAN worker thread. */ -CloudConnectController::~CloudConnectController() +CloudConnectController::~CloudConnectController() = default; + +/*! + * \brief CloudConnectController::initThread + * \details Moves this object and its children onto a worker thread and starts it. + * \param thread Thread to move to, owned by the caller. + * \note Must be called from the main thread, before the event loop runs. + */ +void CloudConnectController::initThread(QThread &thread) { - _canThread.quit(); - _canThread.wait(); + Q_ASSERT_X(QThread::currentThread() == qApp->thread(), __func__, + "CloudConnectController initialization must be done in Main Thread"); + + thread.setObjectName(QString("%1_Thread").arg(metaObject()->className())); + moveToThread(&thread); + thread.start(); } /*! + * \brief CloudConnectController::startCan + * \details Creates and connects the CAN device. + * \return true if the device was created and connected. + * \note Must run on the controller thread, after initThread(). connectDevice() + * installs the CAN socket notifier on the calling thread, so starting + * here keeps the notifier, the frame queue and the drain loop on one + * thread. QCanBusDevice::framesAvailable() is unguarded in Qt 5.15, so + * a cross-thread split of those is a data race, not just a style issue. + */ +bool CloudConnectController::startCan() +{ + Q_ASSERT_X(QThread::currentThread() == thread(), __func__, + "startCan() must run on the controller thread"); + + return _canInterface.init(); +} + +/*! * \brief CloudConnectController::listenForApp - * \details Starts the local-socket server that the Luis application connects to, - * using the app socket path from the settings file. + * \details Creates the app socket server, then starts listening. * \return true if the server bound successfully, false otherwise. + * \note Must run on the controller thread, after initThread(). QLocalServer ties + * its socket engine to the thread that calls listen(), and accepted client + * sockets are parented to that engine. */ bool CloudConnectController::listenForApp() { + Q_ASSERT_X(QThread::currentThread() == thread(), __func__, + "listenForApp() must run on the controller thread"); + + if (_appServer == nullptr) { + _appServer = QSharedPointer::create(this); + } + const QString appSocketPath = _settings.value("Socket/AppSocketName", "/tmp/cloudconnect.sock").toString(); - return _appServer.listen(appSocketPath); + return _appServer->listen(appSocketPath); } /*! + * \brief CloudConnectController::connectToCloud + * \details Builds the MQTT transport from the [Mqtt] settings and starts connecting. + * \return true if the attempt was started, or if MQTT is disabled. + * \note Must run on the controller thread, after initThread(). MqttTcpTransport + * owns a QTcpSocket, which belongs to the thread that constructs it. + */ +bool CloudConnectController::connectToCloud() +{ + Q_ASSERT_X(QThread::currentThread() == thread(), __func__, + "connectToCloud() must run on the controller thread"); + + _mqttEnabled = _settings.value("Mqtt/Enabled", false).toBool(); + if (!_mqttEnabled) { + qInfo().noquote() << "CloudConnect: MQTT disabled, forwarding to the app socket only"; + return true; + } + + MqttTransport::Config config; + config.endpoint = _settings.value("Mqtt/ServerAddress", QStringLiteral("127.0.0.1")).toString(); + config.port = static_cast(_settings.value("Mqtt/Port", 1883).toUInt()); + + _topicPrefix = _settings.value("Mqtt/TopicPrefix", QStringLiteral("diality/v1/devices")).toString(); + _deviceId = _settings.value("Mqtt/DeviceId", QStringLiteral("test_device")).toString(); + + _mqttTransport = QSharedPointer::create(); + if (!_publisher.init(_mqttTransport, config)) { + qCritical().noquote() << "CloudConnect: could not initialise the MQTT publisher"; + return false; + } + + connect(&_publisher, &MqttPublisher::didConnectionChange, + this, &CloudConnectController::onCloudConnectionChange); + connect(&_publisher, &MqttPublisher::didPublishAck, + this, &CloudConnectController::onPublishAck); + + return _publisher.open(); +} + +/*! * \brief CloudConnectController::loadMsgHandling * \details Parses message handling INI and populates _msgHandling. - * \note Unknown msg action values default to Drop; unknown topic strings default to ClinicalData. * \param msgHandlingPath - path to the message handling INI file */ void CloudConnectController::loadMsgHandling(const QString &msgHandlingPath) @@ -75,12 +158,12 @@ { QStringLiteral("SendDelta"), MsgAction::SendDelta }, { QStringLiteral("Drop"), MsgAction::Drop }, }; - static const QHash topicMap = { - { QStringLiteral("HighPriority"), CloudConnectFrame::Type::HighPriority }, - { QStringLiteral("NormalPriority"), CloudConnectFrame::Type::NormalPriority }, - { QStringLiteral("DeviceLogFile"), CloudConnectFrame::Type::DeviceLogFile }, - { QStringLiteral("TreatmentLogFile"), CloudConnectFrame::Type::TreatmentLogFile }, - { QStringLiteral("CloudSyncLogFile"), CloudConnectFrame::Type::CloudSyncLogFile }, + static const QHash topicMap = { + { QStringLiteral("HighPriority"), CloudConnectFrame::Topic::HighPriority }, + { QStringLiteral("NormalPriority"), CloudConnectFrame::Topic::NormalPriority }, + { QStringLiteral("DeviceLogFile"), CloudConnectFrame::Topic::DeviceLogFile }, + { QStringLiteral("TreatmentLogFile"), CloudConnectFrame::Topic::TreatmentLogFile }, + { QStringLiteral("CloudSyncLogFile"), CloudConnectFrame::Topic::CloudSyncLogFile }, }; if (! QFile::exists(msgHandlingPath)) { @@ -108,12 +191,12 @@ const QString actionStr = msgHandlingIni.value(QStringLiteral("action")).toString().trimmed(); const QString topicStr = msgHandlingIni.value(QStringLiteral("topic")).toString().trimmed(); qInfo().noquote() << QString("%1 (0x%2): action=%3, topic=%4") - .arg(msgIdStr).arg(QString("%1").arg(quint16(msgId), 4, 16, QChar('0')).toUpper()).arg(actionStr).arg(topicStr); + .arg(msgIdStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0')).toUpper(), actionStr, topicStr); msgHandlingIni.endGroup(); MsgHandling msgHandling; msgHandling.action = actionMap.value(actionStr, MsgAction::Drop); - msgHandling.topic = topicMap.value(topicStr, CloudConnectFrame::Type::NormalPriority); + msgHandling.topic = topicMap.value(topicStr, CloudConnectFrame::Topic::NormalPriority); if (actionStr.length() > 0 && !actionMap.contains(actionStr)) { qWarning().noquote() << QString("CloudConnect: unknown message action \"%1\" for msgId=0x%2 — defaulting to Drop") @@ -133,8 +216,7 @@ /*! * \brief CloudConnectController::onFrameReceive - * \details Unpacks a CAN frame and feeds it to the dispatcher, which reassembles - * multi-frame messages per CAN id. + * \details Unpacks a CAN frame and sends it to the dispatcher to reassemble the entire CAN message. * \param frame - the received CAN frame */ void CloudConnectController::onFrameReceive(const QCanBusFrame frame) @@ -144,9 +226,7 @@ /*! * \brief CloudConnectController::onMessageReceive - * \details Applies the message handling policy from LeahiMsgHandling.ini: drops, - * forwards unconditionally, or forwards only on payload change. Uses the - * section's topic to set the CloudConnectFrame frame msg_id. + * \details Applies the message handling policy from MsgHandling.ini. * \param msg - the reassembled message */ void CloudConnectController::onMessageReceive(const Can::Message &msg) @@ -159,15 +239,15 @@ } if (it->action == MsgAction::Drop) { - qInfo().noquote() << QString("CloudConnect: dropping message %1 (0x%2)") + qInfo().noquote() << QString("CloudConnect: %1 (0x%2) dropped") .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); return; } auto &[received, cachedMsg] = _msgCache[msg.msgId]; if (it->action == MsgAction::SendDelta && received && cachedMsg.data.chopped(1) == msg.data.chopped(1)) { - qInfo().noquote() << QString("CloudConnect: received message %1 (0x%2) did not changed from previous, dropping") + qInfo().noquote() << QString("CloudConnect: %1 (0x%2) did not changed from previous, dropping") .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); return; } @@ -176,9 +256,163 @@ QDateTime::currentDateTime(), QStringLiteral("test_device"), msg); - const quint16 sequence = _txSequence++; - _appServer.send(it->topic, sequence, payload); - received = true; - cachedMsg = msg; + // TODO: temporary for capture protobuf + // captureProtobuf(payload); + + const bool published = _mqttEnabled && _publisher.publish(mqttTopic(it->topic), payload); + if (_mqttEnabled && !published) { + qWarning().noquote() << QString("CloudConnect: could not publish %1 (0x%2), message lost") + .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); + } + else { + qWarning().noquote() << QString("CloudConnect: %1 (0x%2) published") + .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); + } + + // Cache the message only after it has been successfully sent. + // Note: saving before successfully send may prevent a message of with that msgId from being + // sent until a delta message is received if SendDelta is specified for the msgId. + received = published; + cachedMsg = msg; } + +/*! + * \brief CloudConnectController::mqttTopic + * \details Builds the MQTT topic for a message class. + * \param topic - message class from the message handling INI + * \return Topic of the form {prefix}/{deviceId}/{suffix}. + */ +QString CloudConnectController::mqttTopic(CloudConnectFrame::Topic topic) const +{ + static const QMap suffixes = { + { CloudConnectFrame::Topic::HighPriority, QStringLiteral("high") }, + { CloudConnectFrame::Topic::NormalPriority, QStringLiteral("normal") }, + { CloudConnectFrame::Topic::DeviceLogFile, QStringLiteral("log") }, + { CloudConnectFrame::Topic::TreatmentLogFile, QStringLiteral("tx_log") }, + { CloudConnectFrame::Topic::CloudSyncLogFile, QStringLiteral("cs_log") }, + }; + + return QString("%1/%2/%3").arg(_topicPrefix, _deviceId, + suffixes.value(topic, QStringLiteral("normal"))); +} + +/*! + * \brief CloudConnectController::onCloudConnectionChange + * \details Logs MQTT session transitions. + * \param connected - true when the session came up + */ +void CloudConnectController::onCloudConnectionChange(bool connected) +{ + if (connected) { + qInfo().noquote() << "CloudConnect: MQTT session established"; + } + else { + qWarning().noquote() << "CloudConnect: MQTT session lost, messages are not reaching the cloud"; + } +} + +/*! + * \brief CloudConnectController::onPublishAck + * \details Reports the outcome of attempting to publish a message. + * \param messageId - correlation token passed to publish(); -1 when unused + * \param success - true when the broker acknowledged the message + */ +void CloudConnectController::onPublishAck(qint64 messageId, bool success) +{ + if (!success) { + qWarning().noquote() << "CloudConnect: publish was not acknowledged, id=" << messageId; + } +} + +/*! + * \brief CloudConnectController::captureProtobuf + * \details Appends one varint-delimited protobuf record to the .ser capture file and + * the equivalent json to .json capture file. + * \note TODO: temporary for capture protobuf + */ +void CloudConnectController::captureProtobuf(const QByteArray &payload) +{ + static auto encodeVarint = [](quint32 value) -> QByteArray { + QByteArray out; + do { + quint8 byte = value & 0x7F; + value >>= 7; + if (value != 0) { + byte |= 0x80; // more bytes follow + } + out.append(static_cast(byte)); + } while (value != 0); + return out; + }; + + if (!_captureSerFile) { + QString serPath = _settings.value("Capture/SerFile").toString(); + if (!serPath.isEmpty()) { + // Serialized protobuf streams carry the .ser extension. + if (!serPath.endsWith(QStringLiteral(".ser"), Qt::CaseInsensitive)) { + serPath += QStringLiteral(".ser"); + } + + _captureSerFile = std::make_unique(serPath); + if (_captureSerFile->open(QIODevice::WriteOnly | QIODevice::Append)) { + qWarning().noquote() << "CloudConnect: protobuf capture ser writing to" << serPath; + } + else { + qCritical().noquote() << "CloudConnect: cannot open capture ser file" << serPath; + _captureSerFile.reset(); + } + } + } + + if (_captureSerFile) { + qDebug().noquote() << QString("wrote to capture.ser file (sizes: length=%1, payload=%2)") + .arg(_captureSerFile->write(encodeVarint(static_cast(payload.size())))) + .arg(_captureSerFile->write(payload)); + _captureSerFile->flush(); + } + + if (!_captureJsonFile) { + QString jsonPath = _settings.value("Capture/JsonFile").toString(); + if (!jsonPath.isEmpty()) { + // Serialized protobuf streams carry the .ser extension. + if (!jsonPath.endsWith(QStringLiteral(".json"), Qt::CaseInsensitive)) { + jsonPath += QStringLiteral(".json"); + } + + _captureJsonFile = std::make_unique(jsonPath); + if (_captureJsonFile->open(QIODevice::WriteOnly | QIODevice::Append | QIODevice::Text)) { + qWarning().noquote() << "CloudConnect: protobuf capture json writing to" << jsonPath; + } + else { + qCritical().noquote() << "CloudConnect: cannot open capture json file" << jsonPath; + _captureJsonFile.reset(); + } + } + } + + if (_captureJsonFile) { + leahi::messages::Envelope envelope; + if (envelope.ParseFromArray(payload.constData(), payload.size())) { + const leahi::messages::Header &header = envelope.header(); + const google::protobuf::Descriptor *desc = + google::protobuf::DescriptorPool::generated_pool()->FindMessageTypeByName( + leahi::msgIdToProtoName(static_cast(header.msgid()))); + if (desc) { + google::protobuf::DynamicMessageFactory factory; + std::unique_ptr body(factory.GetPrototype(desc)->New()); + if (body->ParseFromArray(payload.constData(), payload.size())) { + std::string json; + google::protobuf::util::JsonPrintOptions opts; + opts.add_whitespace = true; + opts.always_print_primitive_fields = true; + google::protobuf::util::MessageToJsonString(*body, &json, opts); + QString jsonStr = QString::fromStdString(json); + qDebug().noquote() << QString("wrote to capture.json file (size: json=%2)") + .arg(_captureJsonFile->write(jsonStr.toLatin1())); + _captureJsonFile->flush(); + } + } + } + } +} Index: CloudConnect/CloudConnectController.h =================================================================== diff -u -r6d2ef8c97f4bb34204e95811839b3995000c47c1 -r59b4c22f45a1d098064a886452769204e90cfb4b --- CloudConnect/CloudConnectController.h (.../CloudConnectController.h) (revision 6d2ef8c97f4bb34204e95811839b3995000c47c1) +++ CloudConnect/CloudConnectController.h (.../CloudConnectController.h) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -14,19 +14,24 @@ #include #include +#include // TODO: temporary for protobuf capture #include #include #include #include +#include #include #include +#include #include #include "CloudConnectFrame.h" #include "CanInterface.h" #include "CanMessage.h" #include "CloudConnectServer.h" #include "MessageDispatcher.h" +#include "MqttPublisher.h" +#include "MqttTcpTransport.h" using namespace Can; @@ -38,20 +43,15 @@ Q_OBJECT public: - explicit CloudConnectController(const QString &configPath, const QString &msgHandlingPath, QObject *parent = nullptr); + explicit CloudConnectController(const QString &configPath, const QString &msgHandlingPath, + QObject *parent = nullptr); ~CloudConnectController(); + bool startCan(); bool listenForApp(); + bool connectToCloud(); + void initThread(QThread &thread); -Q_SIGNALS: - /*! - * \brief didCanMessageReceive - * \details Emitted when a complete CAN message has been reassembled. - * \param timestamp - time the message was completed - * \param msg - the completed message - */ - void didCanMessageReceive(const QDateTime timestamp, const Can::Message msg); - private: /*! * \brief Send message handling policy for a CAN message. @@ -67,21 +67,32 @@ */ struct MsgHandling { MsgAction action = MsgAction::Drop; - CloudConnectFrame::Type topic = CloudConnectFrame::Type::NormalPriority; + CloudConnectFrame::Topic topic = CloudConnectFrame::Topic::NormalPriority; }; void loadMsgHandling(const QString &msgHandlingPath); + QString mqttTopic(CloudConnectFrame::Topic topic) const; + void captureProtobuf(const QByteArray &payload); // TODO: temporary for protobuf capture QSettings _settings; Can::CanInterface _canInterface; - QThread _canThread; Can::MessageDispatcher _dispatcher; QMap> _msgCache; QHash _msgHandling; - CloudConnectServer _appServer; - quint16 _txSequence = 0; + QSharedPointer _appServer; + QSharedPointer _mqttTransport; + MqttPublisher _publisher; + QString _topicPrefix; // TODO: define in INI? + QString _deviceId; // TODO: this needs to be sent from Leahi app or retrieved from somewhere + bool _mqttEnabled = false; + // TODO: temporary for protobuf capture + std::unique_ptr _captureSerFile; + std::unique_ptr _captureJsonFile; + private Q_SLOTS: void onFrameReceive(const QCanBusFrame frame); void onMessageReceive(const Can::Message &msg); + void onCloudConnectionChange(bool connected); + void onPublishAck(qint64 messageId, bool success); }; Index: CloudConnect/config/CloudConnect.ini =================================================================== diff -u -r6d2ef8c97f4bb34204e95811839b3995000c47c1 -r59b4c22f45a1d098064a886452769204e90cfb4b --- CloudConnect/config/CloudConnect.ini (.../CloudConnect.ini) (revision 6d2ef8c97f4bb34204e95811839b3995000c47c1) +++ CloudConnect/config/CloudConnect.ini (.../CloudConnect.ini) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -1,2 +1,14 @@ [Socket] AppSocketName=/tmp/cloudconnect.sock + +[Mqtt] +ServerAddress=127.0.0.1 +Port=1883 +TopicPrefix=diality/v1/devices +; TODO: DeviceId should be sent by Leahi app or retrieved from somewhere else +DeviceId=test_device + +; TODO: temporary for protobuf capture +[Capture] +SerFile=/home/leahi/Public/leahi-realtime-cdt/cloudconnect_protobuf.ser +JsonFile=/home/leahi/Public/leahi-realtime-cdt/cloudconnect_protobuf.json Index: CloudConnect/config/LeahiMsgHandling.ini =================================================================== diff -u -rc6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f -r59b4c22f45a1d098064a886452769204e90cfb4b --- CloudConnect/config/LeahiMsgHandling.ini (.../LeahiMsgHandling.ini) (revision c6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f) +++ CloudConnect/config/LeahiMsgHandling.ini (.../LeahiMsgHandling.ini) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -31,6 +31,7 @@ [0x0200] msg_id = MSG_ID_ALARM_TRIGGERED +topic = HighPriority [0x0280] msg_id = MSG_ID_TD_SEND_TEST_CONFIGURATION @@ -43,6 +44,7 @@ [0x0300] msg_id = MSG_ID_ALARM_CLEARED +topic = HighPriority [0x0380] msg_id = MSG_ID_TD_BUBBLE_OVERRIDE_REQUEST @@ -381,6 +383,7 @@ [0x1F00] msg_id = MSG_ID_DD_CONDUCTIVITY_DATA +action = SendDelta [0x1F80] msg_id = MSG_ID_TD_ALARM_CLEAR_ALL_ALARMS_REQUEST @@ -465,6 +468,7 @@ [0x2600] msg_id = MSG_ID_DD_TEMPERATURE_DATA +action = SendDelta [0x2680] msg_id = MSG_ID_TD_BLOOD_PUMP_SET_FLOW_RATE_REQUEST @@ -477,6 +481,7 @@ [0x2700] msg_id = MSG_ID_DIALYSATE_PUMPS_DATA +action = SendDelta [0x2780] msg_id = MSG_ID_TD_BLOOD_PUMP_SET_SPEED_REQUEST @@ -558,6 +563,7 @@ [0x2E00] msg_id = MSG_ID_DD_BAL_CHAMBER_DATA +action = SendDelta [0x2E80] msg_id = MSG_ID_TD_RSP_CURRENT_TREATMENT_PARAMETERS @@ -765,6 +771,7 @@ [0x3FA0] msg_id = MSG_ID_DD_SEND_BLOOD_LEAK_EMB_MODE_RESPONSE +action = SendDelta [0x3FB0] msg_id = MSG_ID_FP_FLUSH_CONCENTRATE_TIMER_OVERRIDE_REQUEST @@ -963,6 +970,7 @@ [0x5300] msg_id = MSG_ID_TD_TEMPERATURE_DATA +action = SendDelta [0x5380] msg_id = MSG_ID_TD_SYRINGE_PUMP_POSITION_OVERRIDE_REQUEST @@ -1428,4 +1436,4 @@ [0xFFFF] msg_id = MSG_ID_ACK_MESSAGE_THAT_REQUIRES_ACK - +action = Drop Index: CloudConnect/main.cpp =================================================================== diff -u -rb47dcc8b37ff07a7efec4dffd0ab1c6f63e5e614 -r59b4c22f45a1d098064a886452769204e90cfb4b --- CloudConnect/main.cpp (.../main.cpp) (revision b47dcc8b37ff07a7efec4dffd0ab1c6f63e5e614) +++ CloudConnect/main.cpp (.../main.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -15,7 +15,7 @@ int main(int argc, char *argv[]) { - // Block SIGINT and SIGTERM from normal delivery; redirect them to a fd. + // Block SIGINT and SIGTERM from normal delivery sigset_t mask; sigemptyset(&mask); sigaddset(&mask, SIGINT); @@ -32,12 +32,11 @@ app.setApplicationName("CloudConnect"); app.setApplicationVersion("1.0"); - // Notifier runs on the main thread inside the event loop — safe to call quit(). + // notifier runs on the main thread inside the event loop QSocketNotifier notifier(sfd, QSocketNotifier::Read); QObject::connect(¬ifier, &QSocketNotifier::activated, [&](int fd) { struct signalfd_siginfo info{}; - // signalfd delivers whole signalfd_siginfo records; a short read leaves - // info zeroed, so quitting on one would report no signal at all. + // A short read leaves info zeroed; don't quit on a phantom signal. const ssize_t bytes = read(fd, &info, sizeof(info)); if (bytes != sizeof(info)) { qWarning("Short read from signalfd (%zd bytes); ignoring", bytes); @@ -59,21 +58,49 @@ QCommandLineOption configOption( {"c", "config"}, "Path to the configuration INI file.", "config", - QDir(app.applicationDirPath()).filePath("config/CloudConnect.ini") + QDir(app.applicationDirPath()).filePath("/home/leahi/Public/leahi-realtime-cdt/CloudConnect/config/CloudConnect.ini") ); QCommandLineOption msgHandlingOption( {"m", "msg_handling"}, "Path to the message handling INI file.", "msg_handling", - QDir(app.applicationDirPath()).filePath("config/LeahiMsgHandling.ini") + QDir(app.applicationDirPath()).filePath("/home/leahi/Public/leahi-realtime-cdt/CloudConnect/config/LeahiMsgHandling.ini") ); parser.addOption(configOption); parser.addOption(msgHandlingOption); parser.process(app); - CloudConnectController ccController(parser.value(configOption), - parser.value(msgHandlingOption)); - if (!ccController.listenForApp()) { - return 1; - } + QThread controllerThread; + CloudConnectController ccController(parser.value(configOption), parser.value(msgHandlingOption)); - return app.exec(); + // Bind on the controller thread: QLocalServer's socket engine belongs to the + // thread that calls listen(), and accepted sockets are parented under it. + QObject::connect(&controllerThread, &QThread::started, &ccController, + [&ccController, &app]() { + if (!ccController.listenForApp()) { + qCritical() << "Failed to bind the app socket; shutting down"; + QMetaObject::invokeMethod(&app, [&app]() { app.exit(1); }, Qt::QueuedConnection); + return; + } + // QTcpSocket belongs to the thread that constructs it. + // Broker connect is async, so false = config failure, not unreachable. + if (!ccController.connectToCloud()) { + qCritical() << "Failed to start the MQTT connection; shutting down"; + QMetaObject::invokeMethod(&app, [&app]() { app.exit(1); }, Qt::QueuedConnection); + return; + } + // CAN last; frames arriving before MQTT CONNACK are still dropped. + if (!ccController.startCan()) { + qWarning() << "Failed to start the CAN interface; no frames will be forwarded"; + } + } + ); + + ccController.initThread(controllerThread); + + const int rc = app.exec(); + + // Stop the worker thread before the controller is destroyed at scope exit. + controllerThread.quit(); + controllerThread.wait(); + + return rc; } Index: leahi-realtime-cdt.pro =================================================================== diff -u -r6d2ef8c97f4bb34204e95811839b3995000c47c1 -r59b4c22f45a1d098064a886452769204e90cfb4b --- leahi-realtime-cdt.pro (.../leahi-realtime-cdt.pro) (revision 6d2ef8c97f4bb34204e95811839b3995000c47c1) +++ leahi-realtime-cdt.pro (.../leahi-realtime-cdt.pro) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -5,12 +5,15 @@ Comms \ MsgUtils \ CANDumpPlayer \ + DCSsim \ CloudConnect Comms.subdir = lib/Comms MsgUtils.subdir = lib/MsgUtils CANDumpPlayer.subdir = tools/CANDumpPlayer +DCSsim.subdir = tools/DCSsim CloudConnect.subdir = CloudConnect MsgUtils.depends = Comms +DCSsim.depends = Comms MsgUtils CloudConnect.depends = Comms MsgUtils Index: lib/Comms/CMakeLists.txt =================================================================== diff -u -r4dccc470aedf6f68a0ad73c1101f8a96188cbdb1 -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/CMakeLists.txt (.../CMakeLists.txt) (revision 4dccc470aedf6f68a0ad73c1101f8a96188cbdb1) +++ lib/Comms/CMakeLists.txt (.../CMakeLists.txt) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -7,7 +7,6 @@ find_package(QT NAMES Qt6 Qt5 REQUIRED COMPONENTS Core) find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core Network SerialBus) -find_package(SQLite3 REQUIRED) set(INCLUDES include/CanInterface.h @@ -21,7 +20,10 @@ include/main.h include/MessageBuilder.h include/MessageDispatcher.h - include/MessageSpool.h + include/MqttMessage.h + include/MqttPublisher.h + include/MqttTcpTransport.h + include/MqttTransport.h include/types.h ) @@ -35,7 +37,9 @@ src/FrameInterface.cpp src/MessageBuilder.cpp src/MessageDispatcher.cpp - src/MessageSpool.cpp + src/MqttMessage.cpp + src/MqttPublisher.cpp + src/MqttTcpTransport.cpp src/types.cpp ) @@ -57,7 +61,6 @@ Qt${QT_VERSION_MAJOR}::Core Qt${QT_VERSION_MAJOR}::Network Qt${QT_VERSION_MAJOR}::SerialBus - SQLite::SQLite3 ) target_include_directories(${PROJECT_NAME} PUBLIC $) Index: lib/Comms/Comms.pro =================================================================== diff -u -r4dccc470aedf6f68a0ad73c1101f8a96188cbdb1 -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/Comms.pro (.../Comms.pro) (revision 4dccc470aedf6f68a0ad73c1101f8a96188cbdb1) +++ lib/Comms/Comms.pro (.../Comms.pro) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -21,7 +21,10 @@ include/main.h \ include/MessageBuilder.h \ include/MessageDispatcher.h \ - include/MessageSpool.h \ + include/MqttPacket.h \ + include/MqttPublisher.h \ + include/MqttTcpTransport.h \ + include/MqttTransport.h \ include/types.h SOURCES = \ @@ -34,13 +37,13 @@ src/FrameInterface.cpp \ src/MessageBuilder.cpp \ src/MessageDispatcher.cpp \ - src/MessageSpool.cpp \ + src/MqttPacket.cpp \ + src/MqttPublisher.cpp \ + src/MqttTcpTransport.cpp \ src/types.cpp INCLUDEPATH += $$PWD/include -LIBS += -lsqlite3 - # qmake only removes the shared objects on distclean. # The following ensures that they are removed on clean, too. # It also ensures that all versioned symlinks (*.so.1, *.so.1.0, etc) are removed. Index: lib/Comms/include/CanInterface.h =================================================================== diff -u -rcfc0df719cb5033078d0cac45ce0f6243810f2e7 -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/include/CanInterface.h (.../CanInterface.h) (revision cfc0df719cb5033078d0cac45ce0f6243810f2e7) +++ lib/Comms/include/CanInterface.h (.../CanInterface.h) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -48,8 +48,6 @@ public Q_SLOTS: bool init(); - bool init(QThread &vThread); - void quit(); Q_SIGNALS: /*! @@ -89,8 +87,6 @@ void consoleOut (const QCanBusFrame &vFrame, const QString &vFrameCount); void initConnections(); bool initDevice(); - void initThread(QThread &vThread); - void quitThread(); void status (const QString &vDescription, QString vError = ""); bool testDevice(); bool transmit (const QCanBusFrame &vFrame); @@ -110,7 +106,6 @@ QString _canStatus = ""; bool _enableConsoleOut = false; - // QThread *_thread = nullptr; bool _init = false; FrameCount _rxFrameCount = 0; Index: lib/Comms/include/CloudConnectClient.h =================================================================== diff -u -rc6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/include/CloudConnectClient.h (.../CloudConnectClient.h) (revision c6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f) +++ lib/Comms/include/CloudConnectClient.h (.../CloudConnectClient.h) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -22,11 +22,10 @@ #include "CloudConnectFrame.h" /*! - * \brief UDS interface to the Connectivity Agent - * \details Manages the QLocalSocket connection to the Connectivity Agent + * \brief Local socket interface to CloudConnect + * \details Manages the QLocalSocket connection to CloudConnect * including automatic reconnection on disconnect or error. - * Outbound: call send() to write a CloudConnectFrame to the socket. - * Inbound: emits didMessageReceive() for each complete parsed frame. + * Uses CloudConnectFrame to send data between the client and CloudConnect. */ class CloudConnectClient : public QObject { @@ -37,7 +36,7 @@ bool init(const QString &socketPath, int reconnectIntervalMs); bool init(const QString &socketPath, int reconnectIntervalMs, QThread &thread); - bool send(CloudConnectFrame::Type type, quint16 sequence, const QByteArray &payload = {}); + bool send(CloudConnectFrame::Topic topic, quint16 sequence, const QByteArray &payload = {}); public Q_SLOTS: void quit(); @@ -46,11 +45,11 @@ /*! * \brief didMessageReceive * \details Emitted when a complete inbound CloudConnectFrame has been parsed. - * \param type - message identifier from the frame header + * \param topic - MQTT topic from the frame header * \param sequence - sequence number from the frame header * \param payload - decoded payload bytes, empty for zero-length frames */ - void didMessageReceive(CloudConnectFrame::Type type, quint16 sequence, QByteArray payload); + void didMessageReceive(CloudConnectFrame::Topic topic, quint16 sequence, QByteArray payload); /*! * \brief didConnect Index: lib/Comms/include/CloudConnectFrame.h =================================================================== diff -u -rc6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/include/CloudConnectFrame.h (.../CloudConnectFrame.h) (revision c6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f) +++ lib/Comms/include/CloudConnectFrame.h (.../CloudConnectFrame.h) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -15,23 +15,20 @@ #include /*! - * \brief CloudConnect to Connectivity Agent message framing - * \details Transport-agnostic binary framing. build() makes a wire-ready frame; - * read() parses inbound bytes (see ReadState). On Complete, read - * type()/sequence()/payload(), then reset(). + * \brief CloudConnect frame for communicating between client and CloudConnect. * * Frame layout — header only (payload_length == 0): * * Byte: 0 1 2 3 4 5 6-9 10 11 * ┌─────────┬────────┬────────┬───────────┬────────┐ - * │ AA 55 │ type │ seq_num│ pay_length│hdr_crc │ - * │ sync │uint16BE│uint16BE│ uint32 BE │uint16BE│ + * │ AA 55 │ topic │ seq_num│ pay_length│hdr_crc │ + * │ sync │ uint16 │ uint16 │ uint32 │ uint16 │ * └─────────┴────────┴────────┴───────────┴────────┘ * * Frame layout — with payload (payload_length > 0): * * ┌── 12-byte header ──┬── N bytes payload ──┬── pay_crc (4 B) ──┐ - * │ (see above) │ uint8[] │ CRC-32/ISO-HDLC │ + * │ header │ uint8[] │ CRC-32/ISO-HDLC │ * └────────────────────┴─────────────────────┴───────────────────┘ * * Header CRC: CRC-16/CCITT (poly 0x1021, init 0xFFFF, no reflection). @@ -42,38 +39,29 @@ public: /*! * \brief MQTT topic identifier carried in every frame header - * \details The Connectivity Agent uses this value to determine the MQTT topic. */ - enum class Type : quint16 { + enum class Topic : quint16 { HighPriority = 0x0001, NormalPriority = 0x0002, DeviceLogFile = 0x0003, TreatmentLogFile = 0x0004, CloudSyncLogFile = 0x0005, - // ClinicalData = 0x0001, - // Diagnostic = 0x0002, - // Ack = 0x0003, - // Alarms = 0x0004, - // Audit = 0x0005, - // DeviceLogFile = 0x0006, - // TreatmentLogFile = 0x0007, - // CloudSyncLogFile = 0x0008, }; /*! - * \brief Result returned by feed() after processing each byte chunk + * \brief Result returned by read() after processing each byte chunk */ enum class ReadState { - Incomplete, ///< More bytes needed — continue feeding. - Complete, ///< Full valid frame assembled — read accessors, then call reset(). - HeaderError, ///< Header CRC mismatch — frame dropped, state reset automatically. - PayloadError, ///< Payload CRC mismatch or oversized payload — frame dropped, state reset automatically. + Incomplete, + Complete, + HeaderError, + PayloadError, }; - static QByteArray build(Type type, quint16 sequence, const QByteArray &payload = {}); + static QByteArray build(Topic topic, quint16 sequence, const QByteArray &payload = {}); ReadState read(QByteArray &bytes); - Type type() const; + Topic topic() const; quint16 sequence() const; QByteArray payload() const; @@ -83,7 +71,7 @@ static constexpr int SYNC_SIZE = 2; static constexpr quint8 SYNC[SYNC_SIZE] = {0xAA, 0x55}; static constexpr int HEADER_SIZE = 12; - static constexpr int TYPE_SIZE = 2; + static constexpr int TOPIC_SIZE = 2; static constexpr int SEQUENCE_SIZE = 2; static constexpr int HEADER_CRC_SIZE = 2; static constexpr int PAYLOAD_CRC_SIZE = 4; @@ -93,7 +81,7 @@ static quint32 crc32isohdlc(const quint8 *data, int len); QByteArray _headerBuf; - Type _rxType = Type::NormalPriority; + Topic _rxTopic = Topic::NormalPriority; quint16 _rxSequence = 0; quint32 _rxPayloadLen = 0; QByteArray _rxPayload; Index: lib/Comms/include/CloudConnectServer.h =================================================================== diff -u -rc6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/include/CloudConnectServer.h (.../CloudConnectServer.h) (revision c6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f) +++ lib/Comms/include/CloudConnectServer.h (.../CloudConnectServer.h) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -21,7 +21,7 @@ #include "CloudConnectFrame.h" /*! - * \brief Server to handle incoming local socket connections to the CloudConnect agent. + * \brief Server to handle incoming local socket connections to CloudConnect. */ class CloudConnectServer : public QObject { @@ -31,18 +31,18 @@ explicit CloudConnectServer(QObject *parent = nullptr); bool listen(const QString &socketPath); - bool send(CloudConnectFrame::Type type, quint16 sequence, const QByteArray &payload = {}); + bool send(CloudConnectFrame::Topic topic, quint16 sequence, const QByteArray &payload = {}); bool isConnected() const; Q_SIGNALS: /*! * \brief didMessageReceive * \details Emitted when a complete inbound CloudConnectFrame has been parsed. - * \param type - message identifier from the frame header + * \param topic - MQTT topic from the frame header * \param sequence - sequence number from the frame header * \param payload - decoded payload bytes, empty for zero-length frames */ - void didMessageReceive(CloudConnectFrame::Type type, quint16 sequence, QByteArray payload); + void didMessageReceive(CloudConnectFrame::Topic topic, quint16 sequence, QByteArray payload); /*! * \brief didConnect @@ -65,5 +65,5 @@ QLocalServer _server; QLocalSocket *_client = nullptr; QByteArray _rxBuf; - CloudConnectFrame _rxMsg; + CloudConnectFrame _rxFrame; }; Fisheye: Tag 59b4c22f45a1d098064a886452769204e90cfb4b refers to a dead (removed) revision in file `lib/Comms/include/MessageSpool.h'. Fisheye: No comparison available. Pass `N' to diff? Index: lib/Comms/include/MqttPacket.h =================================================================== diff -u --- lib/Comms/include/MqttPacket.h (revision 0) +++ lib/Comms/include/MqttPacket.h (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,86 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttPacket.h + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#pragma once + +#include +#include + +/*! + * \brief MQTT packet encoding and decoding. + */ +namespace Mqtt { + +enum class PacketType : quint8 +{ + Connect = 1, + ConnAck = 2, + Publish = 3, + PubAck = 4, + PingReq = 12, + PingResp = 13, + Disconnect = 14, +}; + +enum class ConnAckCode : quint8 +{ + Accepted = 0, + UnacceptableProtocol = 1, + IdentifierRejected = 2, + ServerUnavailable = 3, + BadCredentials = 4, + NotAuthorized = 5, +}; + +/// Outcome of one readPacket() call. +enum class ReadState +{ + Incomplete, ///< Need more bytes; nothing was consumed. + Complete, ///< One packet was decoded and consumed. + Error, ///< Malformed; the caller should drop the connection. +}; + +/*! + * \brief One decoded MQTT packet. + */ +struct Packet +{ + PacketType type = PacketType::Disconnect; + + QString topic; + quint16 packetId = 0; + quint8 qos = 0; + QByteArray payload; + bool dup = false; + bool retain = false; + + QString clientId; + quint16 keepAliveSecs = 0; + bool cleanSession = true; + + ConnAckCode returnCode = ConnAckCode::Accepted; + bool sessionPresent = false; +}; + +QByteArray buildConnect(const QString &clientId, quint16 keepAliveSecs, bool cleanSession); +QByteArray buildConnAck(ConnAckCode code, bool sessionPresent = false); +QByteArray buildPublish(const QString &topic, const QByteArray &payload, quint8 qos, quint16 packetId); +QByteArray buildPubAck(quint16 packetId); +QByteArray buildPingReq(); +QByteArray buildPingResp(); +QByteArray buildDisconnect(); + +ReadState readPacket(QByteArray &bytes, Packet &packet); + +QString typeName(PacketType type); + +} // namespace Mqtt Index: lib/Comms/include/MqttPublisher.h =================================================================== diff -u --- lib/Comms/include/MqttPublisher.h (revision 0) +++ lib/Comms/include/MqttPublisher.h (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,74 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttPublisher.h + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "MqttTransport.h" + +/*! + * \brief Publishes messages to the cloud over MQTT and reports their delivery. + */ +class MqttPublisher : public QObject +{ + Q_OBJECT + +public: + explicit MqttPublisher(QObject *parent = nullptr); + ~MqttPublisher() override; + + bool init(QSharedPointer transport, const MqttTransport::Config &config); + bool init(QSharedPointer transport, const MqttTransport::Config &config, QThread &thread); + + bool isConnected() const; + +public Q_SLOTS: + bool open(); + void close(); + bool publish(const QString &topic, const QByteArray &payload, qint64 messageId = -1); + bool reconnect(const QString &certPath, const QString &keyPath, const QString &clientId); + void quit(); + +Q_SIGNALS: + /*! + * \brief didConnectionChange + * \details Emitted when the MQTT session is established or lost. + * \param connected - true when the session came up, false when it went down + */ + void didConnectionChange(bool connected); + + /*! + * \brief didPublishAck + * \details Emitted when a publish is confirmed or fails. + * \param messageId - the value passed to publish(); -1 when the caller supplied none + * \param success - true when the broker acknowledged the message + */ + void didPublishAck(qint64 messageId, bool success); + +private: + void initThread(QThread &thread); + void quitThread(); + void onTransportState(bool connected); + void onTransportAck(qint64 messageId, bool success); + + QSharedPointer _transport; + MqttTransport::Config _config; + QSharedPointer> _alive; + QAtomicInteger _connected{0}; + bool _init = false; +}; Index: lib/Comms/include/MqttTcpTransport.h =================================================================== diff -u --- lib/Comms/include/MqttTcpTransport.h (revision 0) +++ lib/Comms/include/MqttTcpTransport.h (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,71 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttTcpTransport.h + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "MqttPacket.h" +#include "MqttTransport.h" + +/*! + * \brief MqttTransport over a plain TCP socket. + * \note No TLS. This is the test transport; the production path is the AWS + * SDK transport, which mTLS is part of. Pointing this at AWS IoT Core + * will be refused at the TCP layer. + * \note Unlike the AWS transport, the callbacks fire on the Qt thread that owns + * the socket rather than a foreign event loop. That makes MqttPublisher's + * marshalling a same-thread queued call, so it exercises the ordering but + * not the cross-thread hazard. + */ +class MqttTcpTransport : public QObject, public MqttTransport +{ + Q_OBJECT + +public: + explicit MqttTcpTransport(QObject *parent = nullptr); + ~MqttTcpTransport() override; + + bool open(const Config &config) override; + void close() override; + bool isConnected() const override; + bool publish(const QString &topic, const QByteArray &payload, qint64 messageId) override; + void setCallbacks(StateCallback onState, AckCallback onAck) override; + void clearCallbacks() override; + +private Q_SLOTS: + void onSocketConnected(); + void onSocketDisconnected(); + void onReadyRead(); + void onKeepAliveTimer(); + +private: + void handlePacket(const Mqtt::Packet &packet); + void setSessionState(bool established); + void failPending(); + quint16 nextPacketId(); + + QTcpSocket _socket; + QTimer _keepAliveTimer; + Config _config; + QByteArray _rxBuf; + QHash _pending; + StateCallback _onState; + AckCallback _onAck; + quint16 _lastPacketId = 0; + bool _sessionUp = false; +}; Index: lib/Comms/include/MqttTransport.h =================================================================== diff -u --- lib/Comms/include/MqttTransport.h (revision 0) +++ lib/Comms/include/MqttTransport.h (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,66 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttTransport.h + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#pragma once + +#include +#include + +#include + +/*! + * \brief Transport between MqttPublisher and a concrete MQTT client. + */ +class MqttTransport +{ +public: + /*! + * \brief Parameters for one MQTT-over-mTLS session. + */ + struct Config + { + QString endpoint; ///< ATS endpoint FQDN; no scheme, port or trailing slash. + quint16 port = 8883; ///< Port 443 additionally requires ALPN "x-amzn-mqtt-ca". + QString certPath; ///< Client certificate, PEM. + QString keyPath; ///< Unencrypted private key matching certPath, PEM. + QString caPath; ///< Root CA bundle used to verify the broker. + QString clientId; ///< Must equal the IoT Thing name; the device policy scopes iot:Connect to it. + quint16 keepAliveSecs = 30; ///< MQTT keep-alive interval. + bool cleanSession = true; ///< No broker-side queue; durability is the device's job. + }; + + /// Reports a connection state transition, on the client library's thread. + using StateCallback = std::function; + + /// Reports one publish outcome; messageId echoes the value publish() was given. + using AckCallback = std::function; + + virtual ~MqttTransport() = default; + + /// Begins connecting. False means the attempt could not be started; success arrives via StateCallback. + virtual bool open(const Config &config) = 0; + + /// Closes the connection. The transition is reported through StateCallback. + virtual void close() = 0; + + /// True while the session is established. + virtual bool isConnected() const = 0; + + /// Sends one QoS 1 publish. True means handed to the client, not delivered; delivery is the AckCallback. + virtual bool publish(const QString &topic, const QByteArray &payload, qint64 messageId) = 0; + + /// Installs the callbacks. Must be called before open(). + virtual void setCallbacks(StateCallback onState, AckCallback onAck) = 0; + + /// Removes the callbacks and waits for any still running to finish. + virtual void clearCallbacks() = 0; +}; Index: lib/Comms/src/CanInterface.cpp =================================================================== diff -u -rcfc0df719cb5033078d0cac45ce0f6243810f2e7 -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/src/CanInterface.cpp (.../CanInterface.cpp) (revision cfc0df719cb5033078d0cac45ce0f6243810f2e7) +++ lib/Comms/src/CanInterface.cpp (.../CanInterface.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -104,41 +104,16 @@ initConnections(); + // Close the CAN device on application exit + connect(qApp, &QCoreApplication::aboutToQuit, this, &CanInterface::quitDevice); + status(QString("Connected")); qDebug().noquote() << QString("UI,%1,%2").arg(QString("%1 Initialized").arg(metaObject()->className())).arg(status()); return true; } /*! - * \brief CanInterface::init - * \details Initialized the Class by calling the init() method first - * And initializes the thread vThread by calling initThread - * on success init(). - * \param vThread - the thread - * \return returns the return value of the init() method - */ -bool CanInterface::init(QThread &vThread) -{ - qDebug().noquote() << "*** CanInterface::init"; // SQ - if (! init()) { - return false; - } - initThread(vThread); - return true; -} - -/*! - * \brief CanInterface quit - * \details quits the class - * Calls quitThread - */ -void CanInterface::quit() -{ - quitThread(); // verified -} - -/*! * \brief frameFlags * \details CANBus message frame type as flags * \param vFrame - CANBus message frame @@ -226,37 +201,6 @@ } /*! - * \brief CanInterface::initThread - * \details Moves this object into the thread vThread. - * And checks that this method is called from main thread. - * Also connects quitThread to application aboutToQuit. - * \param vThread - the thread - */ -void CanInterface::initThread(QThread &vThread) -{ - qDebug().noquote() << "*** CanInterface::initThread"; // SQ - // runs in main thread - Q_ASSERT_X(QThread::currentThread() == qApp->thread() , __func__, "The Class initialization must be done in Main Thread" ); - vThread.setObjectName(QString("%1_Thread").arg(metaObject()->className())); - connect(qApp, &QCoreApplication::aboutToQuit, this, &CanInterface::quit); - moveToThread(&vThread); - vThread.start(); -} - -/*! - * \brief CanInterface::quitThread - * \details Moves this object to main thread to be handled by QApplication - * And to be destroyed there. - */ -void CanInterface::quitThread() -{ - // runs in thread - if (QThread::currentThread() == qApp->thread()) { - moveToThread(qApp->thread()); // verified - } -} - -/*! * \brief CanInterface status * \details Sets the Can interface status description * \param vDescription - Description about the CANBus Interface errors Index: lib/Comms/src/CloudConnectClient.cpp =================================================================== diff -u -rc6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/src/CloudConnectClient.cpp (.../CloudConnectClient.cpp) (revision c6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f) +++ lib/Comms/src/CloudConnectClient.cpp (.../CloudConnectClient.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -16,7 +16,10 @@ #include #include -static void qRegister() { qRegisterMetaType("CloudConnectFrame::Type"); } +static void qRegister() +{ + qRegisterMetaType("CloudConnectFrame::Topic"); +} Q_COREAPP_STARTUP_FUNCTION(qRegister) /*! @@ -41,7 +44,7 @@ * first connection attempt. Guards against double-initialisation. * \param socketPath - path to the Unix domain socket * \param reconnectIntervalMs - milliseconds between reconnect attempts - * \return true on success, false if already initialised + * \return true on success, false if already initialized */ bool CloudConnectClient::init(const QString &socketPath, int reconnectIntervalMs) { @@ -58,12 +61,12 @@ /*! * \brief CloudConnectClient::init - * \details Calls init() then moves this object onto vThread. - * Must be called from the main thread. + * \details Calls init() then moves this object onto thread. + * \note Must be called from the main thread. * \param socketPath - path to the Unix domain socket * \param reconnectIntervalMs - milliseconds between reconnect attempts * \param thread - the thread to move this object onto - * \return true on success, false if already initialised + * \return true on success, false if already initialized */ bool CloudConnectClient::init(const QString &socketPath, int reconnectIntervalMs, QThread &thread) { @@ -77,17 +80,17 @@ /*! * \brief CloudConnectClient::send * \details Builds a CloudConnectFrame and writes it to the socket. - * \param type - message identifier for Agent MQTT topic + * \param topic - MQTT topic carried in the frame header * \param sequence - caller-managed sequence number * \param payload - message payload; pass empty for zero-length frames * \return true if written to the socket, false if not connected */ -bool CloudConnectClient::send(CloudConnectFrame::Type type, quint16 sequence, const QByteArray &payload) +bool CloudConnectClient::send(CloudConnectFrame::Topic topic, quint16 sequence, const QByteArray &payload) { if (_socket.state() != QLocalSocket::ConnectedState) { return false; } - const QByteArray frame = CloudConnectFrame::build(type, sequence, payload); + const QByteArray frame = CloudConnectFrame::build(topic, sequence, payload); _socket.write(frame); _socket.flush(); return true; @@ -97,18 +100,24 @@ * \brief CloudConnectClient::quit * \details Moves this object back to the main thread for safe destruction. */ -void CloudConnectClient::quit() { quitThread(); } +void CloudConnectClient::quit() +{ + quitThread(); +} /*! * \brief CloudConnectClient::connectToServer * \details Initiates a connection to the stored socket path. */ -void CloudConnectClient::connectToServer() { _socket.connectToServer(_socketPath, QLocalSocket::ReadWrite); } +void CloudConnectClient::connectToServer() +{ + _socket.connectToServer(_socketPath, QLocalSocket::ReadWrite); +} /*! * \brief CloudConnectClient::initThread * \details Moves this object onto vThread and starts it. - * Must be called from the main thread. + * \note Must be called from the main thread. * \param thread - the thread to move this object onto */ void CloudConnectClient::initThread(QThread &thread) @@ -138,7 +147,7 @@ */ void CloudConnectClient::onConnected() { - qDebug().noquote() << "Agent socket connected to" << _socketPath; + qDebug().noquote() << "CloudConnect client socket connected to" << _socketPath; _reconnectTimer.stop(); emit didConnect(); } @@ -149,7 +158,7 @@ */ void CloudConnectClient::onDisconnected() { - qDebug().noquote() << "Agent socket disconnected — retrying in" << _reconnectTimer.interval() / 1000 << "s"; + qDebug().noquote() << "CloudConnect client socket disconnected, retrying in" << _reconnectTimer.interval() / 1000 << "s"; _rxBuf.clear(); _rxMsg.reset(); _reconnectTimer.start(); @@ -163,7 +172,7 @@ */ void CloudConnectClient::onError(QLocalSocket::LocalSocketError error) { - qDebug().noquote() << "Agent socket error:" << _socket.errorString() << QString("(%1)").arg(error); + qDebug().noquote() << "CloudConnect client socket error:" << _socket.errorString() << QString("(%1)").arg(error); if (!_reconnectTimer.isActive()) { _reconnectTimer.start(); } @@ -182,7 +191,7 @@ do { state = _rxMsg.read(_rxBuf); if (state == CloudConnectFrame::ReadState::Complete) { - emit didMessageReceive(_rxMsg.type(), _rxMsg.sequence(), _rxMsg.payload()); + emit didMessageReceive(_rxMsg.topic(), _rxMsg.sequence(), _rxMsg.payload()); _rxMsg.reset(); } } while (state == CloudConnectFrame::ReadState::Complete && !_rxBuf.isEmpty()); @@ -197,6 +206,6 @@ if (_socket.state() != QLocalSocket::UnconnectedState) { return; } - qDebug().noquote() << "Agent socket reconnecting to" << _socketPath; + qDebug().noquote() << "CloudConnect client socket reconnecting to" << _socketPath; connectToServer(); } Index: lib/Comms/src/CloudConnectFrame.cpp =================================================================== diff -u -rc6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/src/CloudConnectFrame.cpp (.../CloudConnectFrame.cpp) (revision c6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f) +++ lib/Comms/src/CloudConnectFrame.cpp (.../CloudConnectFrame.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -10,35 +10,31 @@ * \date (original) 24-May-2026 * */ -#include "CloudConnectFrame.h" - #include -// --------------------------------------------------------------------------- -// Outbound -// --------------------------------------------------------------------------- +#include "CloudConnectFrame.h" /*! * \brief CloudConnectFrame::build - * \details Builds a complete wire-ready frame with header and optional payload CRCs. - * \param type - message identifier for Agent MQTT topic + * \details Builds a complete frame with header and optional payload CRCs. + * \param topic - message identifier for MQTT topic * \param sequence - caller-managed sequence number - * \param payload - optional payload; pass empty for zero-length frames (e.g. Ack) - * \return complete frame ready to write to the transport + * \param payload - optional payload; pass empty for no payload frames + * \return complete frame with calculated CRCs */ -QByteArray CloudConnectFrame::build(Type type, quint16 sequence, const QByteArray &payload) +QByteArray CloudConnectFrame::build(Topic topic, quint16 sequence, const QByteArray &payload) { const quint32 payloadLen = static_cast(payload.size()); - // Header: sync(2) + msg_id(2) + sequence(2) + payload_length(4) + header_crc(2) = 12 bytes + // Header: sync(2) + topic(2) + sequence(2) + payload_length(4) + header_crc(2) = 12 bytes QByteArray msg(HEADER_SIZE, Qt::Uninitialized); quint8 *header = reinterpret_cast(msg.data()); header[0] = SYNC[0]; header[1] = SYNC[1]; - qToBigEndian(static_cast(type), header + SYNC_SIZE); - qToBigEndian(sequence, header + SYNC_SIZE + TYPE_SIZE); - qToBigEndian(payloadLen, header + SYNC_SIZE + TYPE_SIZE + SEQUENCE_SIZE); + qToBigEndian(static_cast(topic), header + SYNC_SIZE); + qToBigEndian(sequence, header + SYNC_SIZE + TOPIC_SIZE); + qToBigEndian(payloadLen, header + SYNC_SIZE + TOPIC_SIZE + SEQUENCE_SIZE); const quint16 hCrc = crc16ccitt(header, HEADER_SIZE - HEADER_CRC_SIZE); qToBigEndian(hCrc, header + HEADER_SIZE - HEADER_CRC_SIZE); @@ -54,43 +50,37 @@ return msg; } -// --------------------------------------------------------------------------- -// Inbound -// --------------------------------------------------------------------------- - /*! * \brief CloudConnectFrame::read - * \details Feeds raw bytes into the inbound parser state machine. - * Consumed bytes are removed from the front of the buffer. - * On HeaderError or PayloadError the caller may call read() again + * \details Reads raw bytes into the inbound parser state machine. + * \note Consumed bytes are removed from the front of the buffer. + * \note On HeaderError or PayloadError the caller may call read() again * immediately if the buffer is non-empty. * \param bytes - raw bytes from the transport; modified in-place - * \return ReadState indicating the parser outcome + * \return ReadState containing the frame state after parsing the incoming bytes. */ CloudConnectFrame::ReadState CloudConnectFrame::read(QByteArray &bytes) { int pos = 0; ReadState state = ReadState::Incomplete; - // scan for a valid header — skipped when _headerBuf is already populated from a prior read() call + // scan for a valid header, skip when _headerBuf is already populated from a prior read() call while (_headerBuf.size() == 0 && bytes.size() - pos >= HEADER_SIZE && state != ReadState::HeaderError) { if (static_cast(bytes.at(pos)) == SYNC[0] && static_cast(bytes.at(pos + 1)) == SYNC[1]) { _headerBuf.append(bytes.constData() + pos, HEADER_SIZE); if (crc16ccitt(reinterpret_cast(_headerBuf.constData()), HEADER_SIZE - HEADER_CRC_SIZE) == - qFromBigEndian(reinterpret_cast(_headerBuf.constData() + HEADER_SIZE - - HEADER_CRC_SIZE))) + qFromBigEndian(reinterpret_cast(_headerBuf.constData() + HEADER_SIZE - HEADER_CRC_SIZE))) { const quint8 *header = reinterpret_cast(_headerBuf.constData()); int header_pos = SYNC_SIZE; - _rxType = static_cast(qFromBigEndian(header + header_pos)); - header_pos += TYPE_SIZE; + _rxTopic = static_cast(qFromBigEndian(header + header_pos)); + header_pos += TOPIC_SIZE; _rxSequence = qFromBigEndian(header + header_pos); header_pos += SEQUENCE_SIZE; _rxPayloadLen = qFromBigEndian(header + header_pos); pos += HEADER_SIZE; } else { - // TODO: log the header CRC failure _headerBuf.clear(); pos += SYNC_SIZE; state = ReadState::HeaderError; @@ -107,7 +97,6 @@ state = ReadState::Complete; } else if (_rxPayloadLen > MAX_PAYLOAD_LEN) { - // TODO: log the oversized payload _headerBuf.clear(); state = ReadState::PayloadError; } @@ -121,7 +110,6 @@ state = ReadState::Complete; } else { - // TODO: log the payload CRC failure _headerBuf.clear(); state = ReadState::PayloadError; } @@ -135,21 +123,21 @@ } /*! - * \brief CloudConnectFrame::type - * \details Message identifier of the last complete frame. - * Valid only after read() returns ReadState::Complete. - * \return Type of the last complete frame + * \brief CloudConnectFrame::topic + * \details Frame topic getter + * \note Valid only after read() returns ReadState::Complete. + * \return Topic of the last complete frame */ -CloudConnectFrame::Type CloudConnectFrame::type() const +CloudConnectFrame::Topic CloudConnectFrame::topic() const { - return _rxType; + return _rxTopic; } /*! * \brief CloudConnectFrame::sequence - * \details Sequence number of the last complete frame. - * Valid only after read() returns ReadState::Complete. - * \return sequence number of the last complete frame + * \details Sequence number getter + * \note Valid only after read() returns ReadState::Complete. + * \return Sequence number of the last complete frame */ quint16 CloudConnectFrame::sequence() const { @@ -159,8 +147,8 @@ /*! * \brief CloudConnectFrame::payload * \details Payload bytes of the last complete frame. Empty for zero-length frames. - * Valid only after read() returns ReadState::Complete. - * \return payload of the last complete frame + * \note Valid only after read() returns ReadState::Complete. + * \return Payload of the last complete frame */ QByteArray CloudConnectFrame::payload() const { @@ -169,13 +157,13 @@ /*! * \brief CloudConnectFrame::reset - * \details Resets the inbound parser to its initial sync-scanning state. - * Must be called after consuming a Complete frame. + * \details Resets the frame and frame parser state. + * \note Must be called after consuming a Complete frame. */ void CloudConnectFrame::reset() { _headerBuf.clear(); - _rxType = Type::NormalPriority; + _rxTopic = Topic::NormalPriority; _rxSequence = 0; _rxPayloadLen = 0; _rxPayload.clear(); Index: lib/Comms/src/CloudConnectServer.cpp =================================================================== diff -u -rc6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f -r59b4c22f45a1d098064a886452769204e90cfb4b --- lib/Comms/src/CloudConnectServer.cpp (.../CloudConnectServer.cpp) (revision c6a4b63a37f3beb1e8a51702ec3a56a1c32cfe8f) +++ lib/Comms/src/CloudConnectServer.cpp (.../CloudConnectServer.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -16,7 +16,7 @@ /*! * \brief CloudConnectServer::CloudConnectServer - * \details Constructor. Wires the server newConnection signal. + * \details Constructor * \param parent Optional QObject parent. */ CloudConnectServer::CloudConnectServer(QObject *parent) : QObject(parent) @@ -27,7 +27,7 @@ /*! * \brief CloudConnectServer::listen * \details Removes any stale socket file, then starts the server. - * \param socketPath Path to the Unix domain socket. + * \param socketPath Path to the Unix domain socket (UDS). * \return true on success, false if the server could not bind. */ bool CloudConnectServer::listen(const QString &socketPath) @@ -45,25 +45,25 @@ /*! * \brief CloudConnectServer::send * \details Builds a CloudConnectFrame and writes it to the connected client. - * \param type Message identifier for the frame header. + * \param topic MQTT topic for the frame header. * \param sequence Caller-managed sequence number. * \param payload Message payload; pass empty for zero-length frames. * \return true if written to the socket, false if no client is connected. */ -bool CloudConnectServer::send(CloudConnectFrame::Type type, quint16 sequence, const QByteArray &payload) +bool CloudConnectServer::send(CloudConnectFrame::Topic topic, quint16 sequence, const QByteArray &payload) { if (_client == nullptr || _client->state() != QLocalSocket::ConnectedState) { return false; } - const QByteArray frame = CloudConnectFrame::build(type, sequence, payload); + const QByteArray frame = CloudConnectFrame::build(topic, sequence, payload); _client->write(frame); _client->flush(); return true; } /*! * \brief CloudConnectServer::isConnected - * \details Reports whether a client is currently attached to the server. + * \details Reports whether a client is currently connected to the server. * \return true if a client is currently connected. */ bool CloudConnectServer::isConnected() const @@ -80,7 +80,7 @@ { if (_client) { qWarning().noquote() << metaObject()->className() - << ": second connection attempt rejected — already connected"; + << ": previous client connected, new client connection refused"; _server.nextPendingConnection()->deleteLater(); return; } @@ -96,36 +96,34 @@ /*! * \brief CloudConnectServer::onDisconnected - * \details Releases the client socket and resets inbound parser state. + * \details Releases the client socket and resets incoming frame state. */ void CloudConnectServer::onDisconnected() { qInfo().noquote() << metaObject()->className() << ": client disconnected"; _client->deleteLater(); _client = nullptr; _rxBuf.clear(); - _rxMsg.reset(); + _rxFrame.reset(); emit didDisconnect(); } /*! * \brief CloudConnectServer::onReadyRead - * \details Appends incoming bytes to the receive buffer and drains it through - * the CloudConnectFrame parser, emitting didMessageReceive() for each - * complete frame. + * \details Handler for incoming data from the client. */ void CloudConnectServer::onReadyRead() { _rxBuf.append(_client->readAll()); CloudConnectFrame::ReadState state; do { - state = _rxMsg.read(_rxBuf); + state = _rxFrame.read(_rxBuf); switch (state) { case CloudConnectFrame::ReadState::Complete: - emit didMessageReceive(_rxMsg.type(), _rxMsg.sequence(), _rxMsg.payload()); - _rxMsg.reset(); + emit didMessageReceive(_rxFrame.topic(), _rxFrame.sequence(), _rxFrame.payload()); + _rxFrame.reset(); break; case CloudConnectFrame::ReadState::HeaderError: qWarning().noquote() << metaObject()->className() << ": header CRC error — frame dropped"; Fisheye: Tag 59b4c22f45a1d098064a886452769204e90cfb4b refers to a dead (removed) revision in file `lib/Comms/src/MessageSpool.cpp'. Fisheye: No comparison available. Pass `N' to diff? Index: lib/Comms/src/MqttPacket.cpp =================================================================== diff -u --- lib/Comms/src/MqttPacket.cpp (revision 0) +++ lib/Comms/src/MqttPacket.cpp (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,380 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttPacket.cpp + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#include "MqttPacket.h" + +namespace { + +constexpr int MAX_REMAINING_LENGTH_BYTES = 4; ///< MQTT 3.1.1 section 2.2.3. +constexpr quint8 PROTOCOL_LEVEL_311 = 0x04; +constexpr quint8 CONNECT_FLAG_CLEAN_SESSION = 0x02; + +/*! + * \brief encodeRemainingLength + * \details Encodes a length as the MQTT variable-length field: seven value bits + * per byte, high bit set while more bytes follow. + * \param length Byte count of everything after the fixed header. + * \return One to four encoded bytes. + */ +QByteArray encodeRemainingLength(quint32 length) +{ + QByteArray out; + do { + quint8 byte = static_cast(length % 128); + length /= 128; + if (length > 0) { + byte = static_cast(byte | 0x80); + } + out.append(static_cast(byte)); + } while (length > 0); + return out; +} + +/*! + * \brief decodeRemainingLength + * \details Reads the variable-length field starting at offset. + * \param bytes Buffer to read from. + * \param offset Index of the first length byte. + * \param value Receives the decoded length. + * \return Bytes consumed, 0 when more input is needed, or -1 when malformed. + */ +int decodeRemainingLength(const QByteArray &bytes, int offset, quint32 &value) +{ + quint32 multiplier = 1; + value = 0; + + for (int i = 0; i < MAX_REMAINING_LENGTH_BYTES; ++i) { + if (offset + i >= bytes.size()) { + return 0; + } + const quint8 byte = static_cast(bytes.at(offset + i)); + value += static_cast(byte & 0x7F) * multiplier; + if ((byte & 0x80) == 0) { + return i + 1; + } + multiplier *= 128; + } + return -1; +} + +/*! + * \brief appendString + * \details Appends a UTF-8 string in MQTT form: a two-byte big-endian length + * followed by the encoded bytes. + * \param out Buffer to append to. + * \param value String to encode. + */ +void appendString(QByteArray &out, const QString &value) +{ + const QByteArray utf8 = value.toUtf8(); + out.append(static_cast((utf8.size() >> 8) & 0xFF)); + out.append(static_cast(utf8.size() & 0xFF)); + out.append(utf8); +} + +/*! + * \brief readString + * \details Reads one length-prefixed UTF-8 string and advances pos past it. + * \param bytes Buffer to read from. + * \param pos Read cursor, advanced on success. + * \param end One past the last readable index. + * \param out Receives the decoded string. + * \return true if a complete string was read. + */ +bool readString(const QByteArray &bytes, int &pos, int end, QString &out) +{ + if (pos + 2 > end) { + return false; + } + const int length = (static_cast(bytes.at(pos)) << 8) | static_cast(bytes.at(pos + 1)); + pos += 2; + + if (pos + length > end) { + return false; + } + out = QString::fromUtf8(bytes.constData() + pos, length); + pos += length; + return true; +} + +/*! + * \brief readUint16 + * \details Reads one big-endian 16-bit value and advances pos past it. + * \param bytes Buffer to read from. + * \param pos Read cursor, advanced on success. + * \param end One past the last readable index. + * \param out Receives the decoded value. + * \return true if two bytes were available. + */ +bool readUint16(const QByteArray &bytes, int &pos, int end, quint16 &out) +{ + if (pos + 2 > end) { + return false; + } + out = static_cast((static_cast(bytes.at(pos)) << 8) | static_cast(bytes.at(pos + 1))); + pos += 2; + return true; +} + +/*! + * \brief buildFixedHeader + * \details Prepends the fixed header to an already-built packet body. + * \param type Control packet type. + * \param flags Low nibble of byte one. + * \param body Variable header and payload. + * \return The complete packet. + */ +QByteArray buildFixedHeader(Mqtt::PacketType type, quint8 flags, const QByteArray &body) +{ + QByteArray out; + out.append(static_cast((static_cast(type) << 4) | (flags & 0x0F))); + out.append(encodeRemainingLength(static_cast(body.size()))); + out.append(body); + return out; +} + +} // namespace + +namespace Mqtt { + +/*! + * \brief Mqtt::buildConnect + * \details Builds a CONNECT for a publish-only session: no will, no credentials. + * \param clientId Client identifier; must match the IoT Thing name on AWS. + * \param keepAliveSecs Keep-alive interval advertised to the broker. + * \param cleanSession True to ask the broker to retain no session state. + * \return The encoded packet. + */ +QByteArray buildConnect(const QString &clientId, quint16 keepAliveSecs, bool cleanSession) +{ + QByteArray body; + appendString(body, QStringLiteral("MQTT")); + body.append(static_cast(PROTOCOL_LEVEL_311)); + body.append(static_cast(cleanSession ? CONNECT_FLAG_CLEAN_SESSION : 0x00)); + body.append(static_cast((keepAliveSecs >> 8) & 0xFF)); + body.append(static_cast(keepAliveSecs & 0xFF)); + appendString(body, clientId); + + return buildFixedHeader(PacketType::Connect, 0, body); +} + +/*! + * \brief Mqtt::buildConnAck + * \details Builds the broker's response to a CONNECT. + * \param code Acceptance or the reason for refusal. + * \param sessionPresent True when the broker resumed an existing session. + * \return The encoded packet. + */ +QByteArray buildConnAck(ConnAckCode code, bool sessionPresent) +{ + QByteArray body; + body.append(static_cast(sessionPresent ? 0x01 : 0x00)); + body.append(static_cast(code)); + + return buildFixedHeader(PacketType::ConnAck, 0, body); +} + +/*! + * \brief Mqtt::buildPublish + * \details Builds a PUBLISH. The packet identifier is only present at QoS > 0. + * \param topic Fully resolved topic name. + * \param payload Application bytes, passed through untouched. + * \param qos Delivery quality, 0 or 1. + * \param packetId Identifier to correlate the PUBACK; ignored at QoS 0. + * \return The encoded packet. + */ +QByteArray buildPublish(const QString &topic, const QByteArray &payload, quint8 qos, quint16 packetId) +{ + QByteArray body; + appendString(body, topic); + if (qos > 0) { + body.append(static_cast((packetId >> 8) & 0xFF)); + body.append(static_cast(packetId & 0xFF)); + } + body.append(payload); + + return buildFixedHeader(PacketType::Publish, static_cast((qos & 0x03) << 1), body); +} + +/*! + * \brief Mqtt::buildPubAck + * \details Builds the QoS 1 acknowledgement for a received PUBLISH. + * \param packetId Identifier copied from the PUBLISH being acknowledged. + * \return The encoded packet. + */ +QByteArray buildPubAck(quint16 packetId) +{ + QByteArray body; + body.append(static_cast((packetId >> 8) & 0xFF)); + body.append(static_cast(packetId & 0xFF)); + + return buildFixedHeader(PacketType::PubAck, 0, body); +} + +/*! + * \brief Mqtt::buildPingReq + * \details Builds the keep-alive request. + * \return The encoded packet. + */ +QByteArray buildPingReq() +{ + return buildFixedHeader(PacketType::PingReq, 0, QByteArray()); +} + +/*! + * \brief Mqtt::buildPingResp + * \details Builds the keep-alive response. + * \return The encoded packet. + */ +QByteArray buildPingResp() +{ + return buildFixedHeader(PacketType::PingResp, 0, QByteArray()); +} + +/*! + * \brief Mqtt::buildDisconnect + * \details Builds the graceful shutdown notification. + * \return The encoded packet. + */ +QByteArray buildDisconnect() +{ + return buildFixedHeader(PacketType::Disconnect, 0, QByteArray()); +} + +/*! + * \brief Mqtt::readPacket + * \details Decodes one packet from the front of bytes, consuming it on success. + * \param bytes Accumulated stream data; modified in place. + * \param packet Receives the decoded packet. + * \return Complete when a packet was consumed, Incomplete when more data is + * needed, or Error when the stream is malformed. + */ +ReadState readPacket(QByteArray &bytes, Packet &packet) +{ + if (bytes.size() < 2) { + return ReadState::Incomplete; + } + + const quint8 header = static_cast(bytes.at(0)); + const quint8 rawType = static_cast(header >> 4); + const quint8 flags = static_cast(header & 0x0F); + + quint32 remainingLength = 0; + const int lengthBytes = decodeRemainingLength(bytes, 1, remainingLength); + if (lengthBytes == 0) { + return ReadState::Incomplete; + } + if (lengthBytes < 0) { + return ReadState::Error; + } + + const int headerSize = 1 + lengthBytes; + const int totalSize = headerSize + static_cast(remainingLength); + if (bytes.size() < totalSize) { + return ReadState::Incomplete; + } + + packet = Packet(); + packet.type = static_cast(rawType); + + int pos = headerSize; + const int end = totalSize; + bool ok = true; + + switch (packet.type) { + case PacketType::Connect: { + QString protocolName; + ok = readString(bytes, pos, end, protocolName) && protocolName == QStringLiteral("MQTT"); + if (ok && pos + 2 <= end) { + // Protocol level is validated by the caller, which decides the CONNACK code. + pos++; + packet.cleanSession = (static_cast(bytes.at(pos)) & CONNECT_FLAG_CLEAN_SESSION) != 0; + pos++; + ok = readUint16(bytes, pos, end, packet.keepAliveSecs) && readString(bytes, pos, end, packet.clientId); + } + else { + ok = false; + } + break; + } + + case PacketType::ConnAck: { + if (pos + 2 <= end) { + packet.sessionPresent = (static_cast(bytes.at(pos)) & 0x01) != 0; + packet.returnCode = static_cast(static_cast(bytes.at(pos + 1))); + pos += 2; + } + else { + ok = false; + } + break; + } + + case PacketType::Publish: { + packet.dup = (flags & 0x08) != 0; + packet.qos = static_cast((flags >> 1) & 0x03); + packet.retain = (flags & 0x01) != 0; + + ok = readString(bytes, pos, end, packet.topic); + if (ok && packet.qos > 0) { + ok = readUint16(bytes, pos, end, packet.packetId); + } + if (ok) { + packet.payload = bytes.mid(pos, end - pos); + pos = end; + } + break; + } + + case PacketType::PubAck: + ok = readUint16(bytes, pos, end, packet.packetId); + break; + + case PacketType::PingReq: + case PacketType::PingResp: + case PacketType::Disconnect: + break; + + default: + ok = false; + break; + } + + if (!ok) { + return ReadState::Error; + } + + bytes.remove(0, totalSize); + return ReadState::Complete; +} + +/*! + * \brief Mqtt::typeName + * \details Maps a packet type to its specification name, for logging. + * \param type Packet type to name. + * \return The name, or "UNKNOWN" for a type this codec does not handle. + */ +QString typeName(PacketType type) +{ + switch (type) { + case PacketType::Connect: return QStringLiteral("CONNECT"); + case PacketType::ConnAck: return QStringLiteral("CONNACK"); + case PacketType::Publish: return QStringLiteral("PUBLISH"); + case PacketType::PubAck: return QStringLiteral("PUBACK"); + case PacketType::PingReq: return QStringLiteral("PINGREQ"); + case PacketType::PingResp: return QStringLiteral("PINGRESP"); + case PacketType::Disconnect: return QStringLiteral("DISCONNECT"); + } + return QStringLiteral("UNKNOWN"); +} + +} // namespace Mqtt Index: lib/Comms/src/MqttPublisher.cpp =================================================================== diff -u --- lib/Comms/src/MqttPublisher.cpp (revision 0) +++ lib/Comms/src/MqttPublisher.cpp (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,252 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttPublisher.cpp + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#include +#include + +#include "MqttPublisher.h" + +Q_LOGGING_CATEGORY(lcMqtt, "cloudconnect.mqtt") + +/*! + * \brief MqttPublisher::MqttPublisher + * \details Constructor + * \param parent - optional QObject parent + */ +MqttPublisher::MqttPublisher(QObject *parent) : + QObject(parent) +{ +} + +/*! + * \brief MqttPublisher::~MqttPublisher + * \details Stops callbacks reaching this object, then closes the session. + */ +MqttPublisher::~MqttPublisher() +{ + if (_alive != nullptr) { + _alive->storeRelease(0); + } + if (_transport != nullptr) { + _transport->clearCallbacks(); + _transport->close(); + } +} + +/*! + * \brief MqttPublisher::init + * \details Attaches the transport and installs the callbacks. + * \param transport - the MQTT client to publish through + * \param config - endpoint and mTLS credentials + * \return true if the transport was accepted. + * \note Callbacks are installed before any connect so no event can be missed. + */ +bool MqttPublisher::init(QSharedPointer transport, const MqttTransport::Config &config) +{ + if (transport == nullptr) { + qCCritical(lcMqtt) << "init() requires a transport"; + return false; + } + + _transport = transport; + _config = config; + _alive = QSharedPointer>::create(1); + + const QSharedPointer> alive = _alive; + + _transport->setCallbacks( + [this, alive](bool connected) { + if (alive->loadAcquire() == 0) { + return; + } + _connected.storeRelease(connected ? 1 : 0); + QMetaObject::invokeMethod( + this, [this, connected]() { onTransportState(connected); }, Qt::QueuedConnection); + }, + [this, alive](qint64 messageId, bool success) { + if (alive->loadAcquire() == 0) { + return; + } + QMetaObject::invokeMethod( + this, [this, messageId, success]() { onTransportAck(messageId, success); }, Qt::QueuedConnection); + } + ); + + _init = true; + return true; +} + +/*! + * \brief MqttPublisher::init + * \details Calls init() then moves this object onto thread. + * \param transport - the MQTT client to publish through + * \param config - endpoint and mTLS credentials + * \param thread - the thread to move this object onto + * \return true if the transport was accepted. + * \note Must be called from the main thread. + */ +bool MqttPublisher::init(QSharedPointer transport, const MqttTransport::Config &config, + QThread &thread) +{ + if (!init(transport, config)) { + return false; + } + initThread(thread); + return true; +} + +/*! + * \brief MqttPublisher::isConnected + * \details Reports whether the MQTT session is currently established. + * \return true while connected. + */ +bool MqttPublisher::isConnected() const +{ + return _connected.loadAcquire() != 0; +} + +/*! + * \brief MqttPublisher::open + * \details Starts an MQTT connection attempt. + * \return true if the attempt was started. + */ +bool MqttPublisher::open() +{ + if (!_init || _transport == nullptr) { + qCCritical(lcMqtt) << "open() before init()"; + return false; + } + + qCInfo(lcMqtt).noquote() << "Connecting to" << _config.endpoint << "port" << _config.port + << "as" << _config.clientId; + + if (!_transport->open(_config)) { + qCCritical(lcMqtt).noquote() << "Could not start the connection to" << _config.endpoint; + return false; + } + return true; +} + +/*! + * \brief MqttPublisher::close + * \details Closes the MQTT session. + */ +void MqttPublisher::close() +{ + if (_transport != nullptr) { + _transport->close(); + } +} + +/*! + * \brief MqttPublisher::publish + * \param topic - fully resolved MQTT topic + * \param payload - serialised message bytes + * \param messageId - caller token echoed back on acknowledgement, or -1 for none + * \return true if the transport accepted the message. + */ +bool MqttPublisher::publish(const QString &topic, const QByteArray &payload, qint64 messageId) +{ + if (_transport == nullptr || !isConnected()) { + return false; + } + + if (!_transport->publish(topic, payload, messageId)) { + qCWarning(lcMqtt).noquote() << "Transport rejected a publish to" << topic; + return false; + } + return true; +} + +/*! + * \brief MqttPublisher::reconnect + * \details Swaps the mTLS credentials and identity, then rebuilds the session. + * \param certPath - new client certificate, PEM + * \param keyPath - new private key, PEM + * \param clientId - MQTT client id for the new session + * \return true if the new connection attempt was started. + */ +bool MqttPublisher::reconnect(const QString &certPath, const QString &keyPath, const QString &clientId) +{ + _config.certPath = certPath; + _config.keyPath = keyPath; + _config.clientId = clientId; + + close(); + return open(); +} + +/*! + * \brief MqttPublisher::quit + * \details Moves this object back to the main thread for safe destruction. + */ +void MqttPublisher::quit() +{ + quitThread(); +} + +/*! + * \brief MqttPublisher::initThread + * \details Moves this object onto thread and starts it. + * \param thread - the thread to move this object onto + * \note Must be called from the main thread. + */ +void MqttPublisher::initThread(QThread &thread) +{ + Q_ASSERT_X(QThread::currentThread() == qApp->thread(), __func__, + "MqttPublisher::init must be called from the main thread"); + thread.setObjectName(QString("%1_Thread").arg(metaObject()->className())); + connect(qApp, &QCoreApplication::aboutToQuit, this, &MqttPublisher::quit); + moveToThread(&thread); + thread.start(); +} + +/*! + * \brief MqttPublisher::quitThread + * \details Moves this object back to the main thread. + */ +void MqttPublisher::quitThread() +{ + if (QThread::currentThread() != qApp->thread()) { + moveToThread(qApp->thread()); + } +} + +/*! + * \brief MqttPublisher::onTransportState + * \details Reports a connection transition that has been marshalled onto this thread. + * \param connected - true when the session came up + */ +void MqttPublisher::onTransportState(bool connected) +{ + if (connected) { + qCInfo(lcMqtt).noquote() << "Connected to" << _config.endpoint; + } + else { + qCWarning(lcMqtt).noquote() << "Disconnected from" << _config.endpoint; + } + emit didConnectionChange(connected); +} + +/*! + * \brief MqttPublisher::onTransportAck + * \details Report a publish outcome. + * \param messageId - caller token the message carried, or -1 + * \param success - true when the broker acknowledged the message + */ +void MqttPublisher::onTransportAck(qint64 messageId, bool success) +{ + if (!success) { + qCWarning(lcMqtt).noquote() << "Publish failed for message id" << messageId; + } + emit didPublishAck(messageId, success); +} Index: lib/Comms/src/MqttTcpTransport.cpp =================================================================== diff -u --- lib/Comms/src/MqttTcpTransport.cpp (revision 0) +++ lib/Comms/src/MqttTcpTransport.cpp (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,315 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttTcpTransport.cpp + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#include + +#include "MqttTcpTransport.h" + +Q_LOGGING_CATEGORY(lcMqttTcp, "cloudconnect.mqtt.tcp") + +namespace { +constexpr int KEEPALIVE_DIVISOR = 2; +} + +/*! + * \brief MqttTcpTransport::MqttTcpTransport + * \details Constructor + * \param parent - optional QObject parent + */ +MqttTcpTransport::MqttTcpTransport(QObject *parent) : + QObject(parent) +{ + connect(&_socket, &QTcpSocket::connected, this, &MqttTcpTransport::onSocketConnected); + connect(&_socket, &QTcpSocket::disconnected, this, &MqttTcpTransport::onSocketDisconnected); + connect(&_socket, &QTcpSocket::readyRead, this, &MqttTcpTransport::onReadyRead); + connect(&_socket, &QAbstractSocket::errorOccurred, this, [this](QAbstractSocket::SocketError error) { + qCWarning(lcMqttTcp).noquote() << "Socket error:" << _socket.errorString() + << QString("(%1)").arg(static_cast(error)); + setSessionState(false); + }); + + _keepAliveTimer.setSingleShot(false); + connect(&_keepAliveTimer, &QTimer::timeout, this, &MqttTcpTransport::onKeepAliveTimer); +} + +/*! + * \brief MqttTcpTransport::~MqttTcpTransport + * \details Drops the callbacks before closing so teardown cannot re-enter the owner. + */ +MqttTcpTransport::~MqttTcpTransport() +{ + clearCallbacks(); + close(); +} + +/*! + * \brief MqttTcpTransport::open + * \details Starts the TCP connection. CONNECT is sent once the socket is up, and + * the session is not established until CONNACK arrives. + * \param config - endpoint, port, client id and keep-alive + * \return true if the connection attempt was started. + */ +bool MqttTcpTransport::open(const Config &config) +{ + _config = config; + + if (_socket.state() != QAbstractSocket::UnconnectedState) { + qCWarning(lcMqttTcp) << "open() while the socket is already in use"; + return false; + } + + qCInfo(lcMqttTcp).noquote() << "Connecting to" << _config.endpoint << "port" << _config.port; + _socket.connectToHost(_config.endpoint, _config.port); + return true; +} + +/*! + * \brief MqttTcpTransport::close + * \details Sends DISCONNECT when a session is up, then closes the socket. + */ +void MqttTcpTransport::close() +{ + if (_sessionUp && _socket.state() == QAbstractSocket::ConnectedState) { + _socket.write(Mqtt::buildDisconnect()); + _socket.flush(); + } + + _keepAliveTimer.stop(); + + if (_socket.state() != QAbstractSocket::UnconnectedState) { + _socket.disconnectFromHost(); + } + else { + setSessionState(false); + } +} + +/*! + * \brief MqttTcpTransport::isConnected + * \details Reports whether the MQTT session is established. + * \return true once CONNACK has been accepted and before the link drops. + */ +bool MqttTcpTransport::isConnected() const +{ + return _sessionUp; +} + +/*! + * \brief MqttTcpTransport::publish + * \details Sends one QoS 1 PUBLISH and records the packet identifier so the + * matching PUBACK can be tied back to the caller's message. + * \param topic - fully resolved topic name + * \param payload - message bytes, sent unmodified + * \param messageId - caller token to report on acknowledgement, or -1 + * \return true if the packet was written to the socket. + */ +bool MqttTcpTransport::publish(const QString &topic, const QByteArray &payload, qint64 messageId) +{ + if (!_sessionUp) { + return false; + } + + const quint16 packetId = nextPacketId(); + _pending.insert(packetId, messageId); + + const QByteArray packet = Mqtt::buildPublish(topic, payload, 1, packetId); + if (_socket.write(packet) != packet.size()) { + qCWarning(lcMqttTcp).noquote() << "Short write publishing to" << topic; + _pending.remove(packetId); + return false; + } + return true; +} + +/*! + * \brief MqttTcpTransport::setCallbacks + * \details Installs the state and acknowledgement callbacks. + * \param onState - invoked on session establishment and loss + * \param onAck - invoked once per publish outcome + */ +void MqttTcpTransport::setCallbacks(StateCallback onState, AckCallback onAck) +{ + _onState = std::move(onState); + _onAck = std::move(onAck); +} + +/*! + * \brief MqttTcpTransport::clearCallbacks + * \details Drops both callbacks. + */ +void MqttTcpTransport::clearCallbacks() +{ + _onState = nullptr; + _onAck = nullptr; +} + +/*! + * \brief MqttTcpTransport::onSocketConnected + * \details Sends CONNECT once the TCP connection is up. + */ +void MqttTcpTransport::onSocketConnected() +{ + qCInfo(lcMqttTcp).noquote() << "Socket connected, sending CONNECT as" << _config.clientId; + _socket.write(Mqtt::buildConnect(_config.clientId, _config.keepAliveSecs, _config.cleanSession)); +} + +/*! + * \brief MqttTcpTransport::onSocketDisconnected + * \details Tears down session state when the link drops. + */ +void MqttTcpTransport::onSocketDisconnected() +{ + qCInfo(lcMqttTcp).noquote() << "Socket disconnected from" << _config.endpoint; + setSessionState(false); +} + +/*! + * \brief MqttTcpTransport::onReadyRead + * \details Accumulates inbound bytes and drains whole packets out of the buffer. + */ +void MqttTcpTransport::onReadyRead() +{ + _rxBuf.append(_socket.readAll()); + + Mqtt::ReadState state = Mqtt::ReadState::Incomplete; + do { + Mqtt::Packet packet; + state = Mqtt::readPacket(_rxBuf, packet); + + if (state == Mqtt::ReadState::Complete) { + handlePacket(packet); + } + else if (state == Mqtt::ReadState::Error) { + qCCritical(lcMqttTcp) << "Malformed packet, dropping the connection"; + _rxBuf.clear(); + _socket.abort(); + setSessionState(false); + } + } while (state == Mqtt::ReadState::Complete && !_rxBuf.isEmpty()); +} + +/*! + * \brief MqttTcpTransport::onKeepAliveTimer + * \details Sends PINGREQ so the broker does not time the session out. + */ +void MqttTcpTransport::onKeepAliveTimer() +{ + if (_sessionUp) { + _socket.write(Mqtt::buildPingReq()); + } +} + +/*! + * \brief MqttTcpTransport::handlePacket + * \details Dispatches one decoded inbound packet. + * \param packet - the decoded packet + */ +void MqttTcpTransport::handlePacket(const Mqtt::Packet &packet) +{ + switch (packet.type) { + case Mqtt::PacketType::ConnAck: + if (packet.returnCode == Mqtt::ConnAckCode::Accepted) { + qCInfo(lcMqttTcp).noquote() << "CONNACK accepted by" << _config.endpoint; + if (_config.keepAliveSecs > 0) { + _keepAliveTimer.start(_config.keepAliveSecs * 1000 / KEEPALIVE_DIVISOR); + } + setSessionState(true); + } + else { + qCCritical(lcMqttTcp).noquote() + << "CONNACK refused with code" << static_cast(packet.returnCode); + setSessionState(false); + _socket.disconnectFromHost(); + } + break; + + case Mqtt::PacketType::PubAck: { + // take() returns a default-constructed value for an unknown key, so an + // unsolicited PUBACK is reported against -1 rather than silently dropped. + if (!_pending.contains(packet.packetId)) { + qCWarning(lcMqttTcp) << "PUBACK for an unknown packet id" << packet.packetId; + break; + } + const qint64 messageId = _pending.take(packet.packetId); + if (_onAck) { + _onAck(messageId, true); + } + break; + } + + case Mqtt::PacketType::PingResp: + break; + + default: + qCWarning(lcMqttTcp).noquote() << "Unexpected" << Mqtt::typeName(packet.type) << "from the broker"; + break; + } +} + +/*! + * \brief MqttTcpTransport::setSessionState + * \details Applies a session transition and reports it, suppressing repeats. + * \param established - true when the session came up + */ +void MqttTcpTransport::setSessionState(bool established) +{ + if (_sessionUp == established) { + return; + } + _sessionUp = established; + + if (!established) { + _keepAliveTimer.stop(); + _rxBuf.clear(); + failPending(); + } + + if (_onState) { + _onState(established); + } +} + +/*! + * \brief MqttTcpTransport::failPending + * \details Reports every unacknowledged publish as failed. + */ +void MqttTcpTransport::failPending() +{ + if (_pending.isEmpty()) { + return; + } + + qCWarning(lcMqttTcp) << "Failing" << _pending.size() << "unacknowledged publishes"; + + const QHash pending = _pending; + _pending.clear(); + + if (_onAck) { + for (auto it = pending.constBegin(); it != pending.constEnd(); ++it) { + _onAck(it.value(), false); + } + } +} + +/*! + * \brief MqttTcpTransport::nextPacketId + * \details Produces the next packet identifier. + * \return A value in 1..65535. + */ +quint16 MqttTcpTransport::nextPacketId() +{ + ++_lastPacketId; + if (_lastPacketId == 0) { + _lastPacketId = 1; + } + return _lastPacketId; +} Index: scripts/MsgUtils/msgutils/templates/MsgProtoUtils_cpp.jinja =================================================================== diff -u -rf9c6b488aa4135e8cd47ccd3fdc6c3ae1cd831aa -r59b4c22f45a1d098064a886452769204e90cfb4b --- scripts/MsgUtils/msgutils/templates/MsgProtoUtils_cpp.jinja (.../MsgProtoUtils_cpp.jinja) (revision f9c6b488aa4135e8cd47ccd3fdc6c3ae1cd831aa) +++ scripts/MsgUtils/msgutils/templates/MsgProtoUtils_cpp.jinja (.../MsgProtoUtils_cpp.jinja) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -86,7 +86,6 @@ if (payload.fromQByteArray(msg.data) == false) { qDebug().noquote() << "ERROR: could not convert CAN message with MsgId={{ msg['msg_name'] }} to struct"; } - payload.dump(); return serializeProto(payload, timestamp, deviceSerialNum, msg.msgId, msg.sequence); } {%- endfor %} Index: tools/CMakeLists.txt =================================================================== diff -u -r6d2ef8c97f4bb34204e95811839b3995000c47c1 -r59b4c22f45a1d098064a886452769204e90cfb4b --- tools/CMakeLists.txt (.../CMakeLists.txt) (revision 6d2ef8c97f4bb34204e95811839b3995000c47c1) +++ tools/CMakeLists.txt (.../CMakeLists.txt) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -1,3 +1,2 @@ -if(UNIX AND NOT APPLE) - add_subdirectory(CANDumpPlayer) -endif() +add_subdirectory(CANDumpPlayer) +add_subdirectory(DCSsim) Index: tools/DCSsim/CMakeLists.txt =================================================================== diff -u --- tools/DCSsim/CMakeLists.txt (revision 0) +++ tools/DCSsim/CMakeLists.txt (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,52 @@ +cmake_minimum_required(VERSION 3.16) + +project(DCSsim LANGUAGES CXX) + +set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_INCLUDE_CURRENT_DIR ON) +set(CMAKE_AUTOMOC ON) +set(CMAKE_DISABLE_IN_SOURCE_BUILD ON) + +set(TOOLS_BIN ${CMAKE_CURRENT_SOURCE_DIR}/../bin) +file(MAKE_DIRECTORY ${TOOLS_BIN}) + +find_package(QT NAMES Qt6 Qt5 REQUIRED COMPONENTS Core Network) +find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core Network) +find_package(Comms HINTS ${CMAKE_CURRENT_SOURCE_DIR}/../../lib/Comms REQUIRED) +find_package(MsgUtils HINTS ${CMAKE_CURRENT_SOURCE_DIR}/../../lib/MsgUtils REQUIRED) + +set(INCLUDES + DCSSimController.h +) + +set(SRCS + DCSSimController.cpp + main.cpp +) + +add_executable(${PROJECT_NAME} + ${INCLUDES} ${SRCS} +) + +target_link_libraries(${PROJECT_NAME} PRIVATE + Comms + MsgUtils + Qt${QT_VERSION_MAJOR}::Core + Qt${QT_VERSION_MAJOR}::Network +) + +# rpath to the in-tree libraries rather than needing LD_LIBRARY_PATH +set(DCSSIM_RPATH + "${CMAKE_CURRENT_SOURCE_DIR}/../../lib/Comms/lib;${CMAKE_CURRENT_SOURCE_DIR}/../../lib/MsgUtils/lib" +) + +set_target_properties(${PROJECT_NAME} PROPERTIES + RUNTIME_OUTPUT_DIRECTORY ${TOOLS_BIN} + BUILD_RPATH "${DCSSIM_RPATH}" + INSTALL_RPATH "${DCSSIM_RPATH}" +) + +install(TARGETS ${PROJECT_NAME} + RUNTIME DESTINATION ${TOOLS_BIN} +) Index: tools/DCSsim/DCSSimController.cpp =================================================================== diff -u --- tools/DCSsim/DCSSimController.cpp (revision 0) +++ tools/DCSsim/DCSSimController.cpp (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,224 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file DCSSimController.cpp + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#include +#include +#include +#include +#include + +#include "DCSSimController.h" +#include "LeahiMsgDefs.h" +#include "LeahiMsgProtoUtils.h" + +/*! + * \brief DCSSimController::DCSSimController + * \details Constructor + * \param port TCP port to listen on. + * \param parent QObject parent. + */ +DCSSimController::DCSSimController(quint16 port, QObject *parent) : + QObject(parent), + _port(port) +{ + connect(&_server, &QTcpServer::newConnection, this, &DCSSimController::onNewConnection); +} + +/*! + * \brief DCSSimController::listen + * \details Starts the controller listening on all interfaces. + * \return true on success, false if the port cannot be bound. + */ +bool DCSSimController::listen() +{ + if (!_server.listen(QHostAddress::Any, _port)) { + qCritical().noquote() << "DCSSimController: cannot listen on port" << _port << "—" << _server.errorString(); + return false; + } + qInfo().noquote() << "DCSSimController: listening on port" << _port; + return true; +} + +/*! + * \brief DCSSimController::onNewConnection + * \details Accepts the pending connection, if another client is not already connect. + */ +void DCSSimController::onNewConnection() +{ + if (_client != nullptr) { + qWarning().noquote() << "DCSSimController: connection refused, client already connected"; + _server.nextPendingConnection()->deleteLater(); + return; + } + + _client = _server.nextPendingConnection(); + qInfo().noquote() << "DCSSimController: client connected from" << _client->peerAddress().toString(); + + connect(_client, &QTcpSocket::readyRead, this, &DCSSimController::onReadyRead); + connect(_client, &QTcpSocket::disconnected, this, &DCSSimController::onDisconnected); +} + +/*! + * \brief DCSSimController::onDisconnected + * \details Perform housekeeping on client disconnect. + */ +void DCSSimController::onDisconnected() +{ + qInfo().noquote() << "DCSSimController: client disconnected after" << _publishCount << "messages"; + _client->deleteLater(); + _client = nullptr; + _rxBuf.clear(); +} + +/*! + * \brief DCSSimController::onReadyRead + * \details Handler for incoming data from the socket. + */ +void DCSSimController::onReadyRead() +{ + _rxBuf.append(_client->readAll()); + + Mqtt::ReadState state = Mqtt::ReadState::Incomplete; + do { + Mqtt::Packet packet; + state = Mqtt::readPacket(_rxBuf, packet); + + if (state == Mqtt::ReadState::Complete) { + handlePacket(packet); + } + else if (state == Mqtt::ReadState::Error) { + qWarning().noquote() << "DCSSimController: client read error, dropping the connection"; + _rxBuf.clear(); + _client->abort(); + return; + } + } while (state == Mqtt::ReadState::Complete && !_rxBuf.isEmpty() && _client != nullptr); +} + +/*! + * \brief DCSSimController::handlePacketn + * \details Incoming MQTT message handler. + * \param packet The decoded MQTT message. + */ +void DCSSimController::handlePacket(const Mqtt::Packet &packet) +{ + switch (packet.type) { + case Mqtt::PacketType::Connect: + handleConnectPacket(packet); + break; + case Mqtt::PacketType::Publish: + handlePublishPacket(packet); + break; + case Mqtt::PacketType::PingReq: + _client->write(Mqtt::buildPingResp()); + break; + case Mqtt::PacketType::Disconnect: + qInfo().noquote() << "DCSSimController: DISCONNECT requested from client"; + _client->disconnectFromHost(); + break; + default: + qWarning().noquote() << "DCSSimController: received unexpected" << Mqtt::typeName(packet.type) << "from client"; + break; + } +} + +/*! + * \brief DCSSimController::handleConnectPacket + * \details Handle received CONNECT message. + * \param packet The decoded message. + */ +void DCSSimController::handleConnectPacket(const Mqtt::Packet &packet) +{ + qInfo().noquote() << QString("DCSSimController: CONNECT requested: clientId=%1 keepAlive=%2s cleanSession=%3") + .arg(packet.clientId) + .arg(packet.keepAliveSecs) + .arg(packet.cleanSession ? "true" : "false"); + + _client->write(Mqtt::buildConnAck(Mqtt::ConnAckCode::Accepted)); + qInfo().noquote() << "DCSSimController: CONNACK sent"; +} + +/*! + * \brief DCSSimController::handlePublishPacket + * \details Handle received PUBLISH message. + * \param packet The decoded message. + */ +void DCSSimController::handlePublishPacket(const Mqtt::Packet &packet) +{ + ++_publishCount; + + qInfo().noquote() << QString("DCSSimController: PUBLISH: topic=%1 packetId=%2 qos=%3 bytes=%4") + .arg(packet.topic) + .arg(packet.packetId) + .arg(packet.qos) + .arg(packet.payload.size()); + + if (packet.qos == 1) { + _client->write(Mqtt::buildPubAck(packet.packetId)); + qInfo().noquote() << QString("DCSSimController: PUBACK sent (packetId=%1)").arg(packet.packetId); + } + + dumpPayload(packet.payload); +} + +/*! + * \brief DCSSimController::dumpPayload + * \details Decodes a payload containing a serialized protobuf message and prints it as JSON. + * \param payload Serialized protobuf bytes from the PUBLISH message. + */ +void DCSSimController::dumpPayload(const QByteArray &payload) +{ + // The Envelope header is field 1 of every typed message, so parsing the + // payload as an Envelope reads the header without knowing the real type yet. + leahi::messages::Envelope envelope; + if (!envelope.ParseFromArray(payload.constData(), static_cast(payload.size()))) { + qWarning().noquote() << "DCSSimController: could not parse the Envelope header, payload dropped"; + return; + } + const leahi::messages::Header &header = envelope.header(); + const quint16 msgId = static_cast(header.msgid()); + + qInfo().noquote() << QString("DCSSimController: msgId=0x%1 (%2) serial=%3 seq=%4") + .arg(msgId, 4, 16, QChar('0')) + .arg(leahi::msgIdString(static_cast(msgId))) + .arg(QString::fromStdString(header.deviceserialnum())) + .arg(header.sequence()); + + const std::string &typeName = leahi::msgIdToProtoName(msgId); + if (typeName.empty()) { + qWarning().noquote() << QString("DCSSimController: unknown message with msgId=0x%1") + .arg(msgId, 4, 16, QChar('0')); + return; + } + + const google::protobuf::Descriptor *descriptor = + google::protobuf::DescriptorPool::generated_pool()->FindMessageTypeByName(typeName); + if (descriptor == nullptr) { + qWarning().noquote() << "DCSSimController: no descriptor for" << QString::fromStdString(typeName); + return; + } + + google::protobuf::DynamicMessageFactory factory; + std::unique_ptr body(factory.GetPrototype(descriptor)->New()); + if (!body->ParseFromArray(payload.constData(), static_cast(payload.size()))) { + qWarning().noquote() << "DCSSimController: could not parse" << QString::fromStdString(typeName); + return; + } + + std::string json; + google::protobuf::util::JsonPrintOptions options; + options.add_whitespace = true; + options.always_print_primitive_fields = true; + google::protobuf::util::MessageToJsonString(*body, &json, options); + + qDebug().noquote() << QString::fromStdString(typeName) << ":" << Qt::endl << QString::fromStdString(json); +} Index: tools/DCSsim/DCSSimController.h =================================================================== diff -u --- tools/DCSsim/DCSSimController.h (revision 0) +++ tools/DCSsim/DCSSimController.h (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,51 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file DCSSimController.h + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#pragma once + +#include +#include +#include +#include +#include + +#include "MqttPacket.h" + +/*! + * \brief Diality Cloud System (DCS) simulator + */ +class DCSSimController : public QObject +{ + Q_OBJECT + +public: + explicit DCSSimController(quint16 port, QObject *parent = nullptr); + + bool listen(); + +private Q_SLOTS: + void onNewConnection(); + void onReadyRead(); + void onDisconnected(); + +private: + void handlePacket(const Mqtt::Packet &packet); + void handleConnectPacket(const Mqtt::Packet &packet); + void handlePublishPacket(const Mqtt::Packet &packet); + void dumpPayload(const QByteArray &payload); + + QTcpServer _server; + QPointer _client; + QByteArray _rxBuf; + quint16 _port; + quint64 _publishCount = 0; +}; Index: tools/DCSsim/DCSsim.pro =================================================================== diff -u --- tools/DCSsim/DCSsim.pro (revision 0) +++ tools/DCSsim/DCSsim.pro (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,23 @@ +TEMPLATE = app +TARGET = DCSsim +CONFIG += c++20 +QT += core network +QT -= gui + +CONFIG += moc + +QMAKE_CXXFLAGS += -Wall -Werror -Wextra + +DESTDIR = $$PWD/../bin + +HEADERS = \ + DCSSimController.h + +SOURCES = \ + DCSSimController.cpp \ + main.cpp + +INCLUDEPATH += $$PWD + +include($$PWD/../../lib/Comms/Comms.pri) +include($$PWD/../../lib/MsgUtils/MsgUtils.pri) Index: tools/DCSsim/main.cpp =================================================================== diff -u --- tools/DCSsim/main.cpp (revision 0) +++ tools/DCSsim/main.cpp (revision 59b4c22f45a1d098064a886452769204e90cfb4b) @@ -0,0 +1,48 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file main.cpp + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#include +#include +#include + +#include "DCSSimController.h" + +constexpr quint16 DEFAULT_MQTT_PORT = 1883; + +int main(int argc, char *argv[]) +{ + QCoreApplication app(argc, argv); + QCoreApplication::setApplicationName("DCSsim"); + + QCommandLineParser parser; + parser.setApplicationDescription("Diality Cloud System simulator"); + parser.addHelpOption(); + + QCommandLineOption portOption(QStringList() << "p" << "port", + "TCP listening port", "port", QString::number(DEFAULT_MQTT_PORT)); + parser.addOption(portOption); + parser.process(app); + + bool ok = false; + const uint port = parser.value(portOption).toUInt(&ok); + if (!ok || port == 0 || port > 65535) { + qCritical().noquote() << "DCSsim: invalid port" << parser.value(portOption); + return 1; + } + + DCSSimController simController(static_cast(port)); + if (!simController.listen()) { + return 1; + } + + return app.exec(); +}