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(); + } + } + } + } +}