Index: CloudConnect/CloudConnectController.cpp =================================================================== diff -u -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 -raaebfee335c74b0250864a6dce0555f866adadea --- CloudConnect/CloudConnectController.cpp (.../CloudConnectController.cpp) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) +++ CloudConnect/CloudConnectController.cpp (.../CloudConnectController.cpp) (revision aaebfee335c74b0250864a6dce0555f866adadea) @@ -10,48 +10,103 @@ * \date (original) 24-May-2026 * */ +#include +#include +#include + #include #include #include +#include +#include +#include #include #include "CloudConnectController.h" +#include "LeahiMsgDefs.pb.h" #include "LeahiMsgProtoUtils.h" -#include // TODO: temporary for capture protobuf -#include // TODO: temporary for capture protobuf -#include // TODO: temporary for capture protobuf +Q_LOGGING_CATEGORY(logConfig, "config") +Q_LOGGING_CATEGORY(logProto, "proto") +Q_LOGGING_CATEGORY(logCanRouting, "can.routing") +Q_LOGGING_CATEGORY(logMqtt, "mqtt") +Q_LOGGING_CATEGORY(logStats, "mqtt.stats") /*! * \brief CloudConnectController::CloudConnectController * \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 canRoutingPath - path to the CAN message routing 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(QObject *parent) : QObject(parent), - _settings(configPath, QSettings::IniFormat), _canInterface(this), _dispatcher(this), _mqttClient(this) { - loadMsgHandling(msgHandlingPath); - - connect(&_canInterface, &Can::CanInterface::didFrameReceive, this, &CloudConnectController::onFrameReceive); - connect(&_dispatcher, &Can::MessageDispatcher::didActionReceive, this, &CloudConnectController::onMessageReceive); + connect(&_canInterface, &Can::CanInterface::didFrameReceive, this, &CloudConnectController::onCanFrameReceive); + connect(&_dispatcher, &Can::MessageDispatcher::didActionReceive, this, &CloudConnectController::onCanMessageReceive); + connect(&_mqttClient, &MqttClient::didStateChanged, this, &CloudConnectController::onCloudStateChanged); + connect(&_mqttClient, &MqttClient::didMessageStatusChanged, this, &CloudConnectController::onCloudMessageStatusChanged); + connect(&_mqttClient, &MqttClient::didMessageReceived, this, &CloudConnectController::onCloudMessageReceived); } /*! * \brief CloudConnectController::~CloudConnectController */ CloudConnectController::~CloudConnectController() = default; +bool CloudConnectController::loadConfig(const QString &configPath) +{ + static auto resolvePath = [](const QString &path) -> QString { + const QFileInfo info(path); + return info.isAbsolute() + ? path + : QString("%1/%2").arg(QCoreApplication::applicationDirPath(), path); + }; + + if (!QFileInfo::exists(configPath)) { + qCCritical(logConfig).noquote() << QString("config file %1 does not exist").arg(configPath); + return false; + } + + QSettings config = QSettings(configPath, QSettings::IniFormat); + if (config.status() != QSettings::NoError) { + qCCritical(logConfig).noquote() << QString("could not read config file %1").arg(configPath); + return false; + } + + // NOTE: do not try to prefix keys in the General (i.e. "General/") or beginGroup("General") or + // QSettings will look for a group in the INI file called "%General", "General" is a special group. + if (!loadCanRouting(resolvePath(config.value(QStringLiteral("CanRoutingConfig")).toString()))) { + return false; + } + + _appServerSocketPath = config.value("App/SocketName", "/tmp/cloudconnect.sock").toString(); + + config.beginGroup(QStringLiteral("Cloud")); + MqttClient::Config mqttConfig; + mqttConfig.endpoint = config.value(QStringLiteral("ServerAddress"), QStringLiteral("localhost")).toString(); + mqttConfig.port = config.value(QStringLiteral("Port"), 8883).toInt(); + mqttConfig.clientId = config.value(QStringLiteral("ClientId")).toString(); + mqttConfig.caPath = resolvePath(config.value(QStringLiteral("CaFile")).toString()); + mqttConfig.certPath = resolvePath(config.value(QStringLiteral("CertFile")).toString()); + mqttConfig.keyPath = resolvePath(config.value(QStringLiteral("KeyFile")).toString()); + _deviceId = config.value(QStringLiteral("SerialNumber")).toString(); + _topicPrefix = config.value(QStringLiteral("TopicPrefix")).toString(); + config.endGroup(); + if (!_mqttClient.init(mqttConfig)) { + return false; + } + + return true; +} + /*! * \brief CloudConnectController::initThread * \details Moves this object and its children onto a worker thread and starts it. @@ -61,7 +116,7 @@ void CloudConnectController::initThread(QThread &thread) { Q_ASSERT_X(QThread::currentThread() == qApp->thread(), __func__, - "CloudConnectController initialization must be done in Main Thread"); + "CloudConnectController initialization must be done in Main Thread"); thread.setObjectName(QString("%1_Thread").arg(metaObject()->className())); moveToThread(&thread); @@ -81,7 +136,7 @@ bool CloudConnectController::startCan() { Q_ASSERT_X(QThread::currentThread() == thread(), __func__, - "startCan() must run on the controller thread"); + "startCan() must run on the controller thread"); return _canInterface.init(); } @@ -97,14 +152,13 @@ bool CloudConnectController::listenForApp() { Q_ASSERT_X(QThread::currentThread() == thread(), __func__, - "listenForApp() must run on the controller thread"); + "listenForApp() must run on the controller thread"); // use Q_ASSERT only during object creation or thread move 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(_appServerSocketPath); } /*! @@ -117,44 +171,23 @@ bool CloudConnectController::connectToCloud() { Q_ASSERT_X(QThread::currentThread() == thread(), __func__, - "connectToCloud() must run on the controller thread"); + "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; - } - - _topicPrefix = _settings.value("Mqtt/TopicPrefix", QStringLiteral("diality/v1/devices")).toString(); - _deviceId = _settings.value("Mqtt/DeviceId", QStringLiteral("test_device")).toString(); - - MqttClient::Config config; - config.endpoint = _settings.value("Mqtt/ServerAddress", QStringLiteral("127.0.0.1")).toString(); - config.port = static_cast(_settings.value("Mqtt/Port", 1883).toUInt()); - config.clientId = _deviceId; - config.caPath = _settings.value("Mqtt/CaFile").toString(); - config.certPath = _settings.value("Mqtt/CertFile").toString(); - config.keyPath = _settings.value("Mqtt/KeyFile").toString(); - - if (!_mqttClient.init(config)) { - qCritical().noquote() << "CloudConnect: could not initialise the MQTT interface"; - return false; - } - return _mqttClient.open(); } /*! - * \brief CloudConnectController::loadMsgHandling - * \details Parses message handling INI and populates _msgHandling. - * \param msgHandlingPath - path to the message handling INI file + * \brief CloudConnectController::loadCanRouting + * \details Parses CAN message routing INI and populates _canRouting. + * \param canRoutingPath - path to the CAN message routing INI file + * \return true if CanRouting config successfully loads, otherwise false */ -void CloudConnectController::loadMsgHandling(const QString &msgHandlingPath) +bool CloudConnectController::loadCanRouting(const QString &canRoutingPath) { - static const QHash actionMap = { - { QStringLiteral("SendAlways"), MsgAction::SendAlways }, - { QStringLiteral("SendDelta"), MsgAction::SendDelta }, - { QStringLiteral("Drop"), MsgAction::Drop }, + static const QHash actionMap = { + { QStringLiteral("SendAlways"), CanAction::SendAlways }, + { QStringLiteral("SendDelta"), CanAction::SendDelta }, + { QStringLiteral("Drop"), CanAction::Drop }, }; static const QHash topicMap = { { QStringLiteral("HighPriority"), CloudConnectFrame::Topic::HighPriority }, @@ -164,121 +197,133 @@ { QStringLiteral("CloudSyncLogFile"), CloudConnectFrame::Topic::CloudSyncLogFile }, }; - if (! QFile::exists(msgHandlingPath)) { - qWarning().noquote() << "CloudConnect: handling INI" << msgHandlingPath << "does not exist"; - return; + if (!QFile::exists(canRoutingPath)) { + qCWarning(logCanRouting).noquote() << QString("CAN mesg routing config file %1 does not exist").arg(canRoutingPath); + return false; } - QSettings msgHandlingIni(msgHandlingPath, QSettings::IniFormat); - if (msgHandlingIni.status() != QSettings::NoError) { - qWarning().noquote() << "CloudConnect: could not read message handling INI" << msgHandlingPath << "— all messages will be dropped"; - return; + QSettings canRoutingConfig(canRoutingPath, QSettings::IniFormat); + if (canRoutingConfig.status() != QSettings::NoError) { + qCWarning(logCanRouting).noquote() << QString("could not read CAN message routing config file %1").arg(canRoutingPath); + return false; } int msgCount = 0; - for (const QString &group : msgHandlingIni.childGroups()) { + for (const QString &group : canRoutingConfig.childGroups()) { bool ok = false; const Can::MsgId msgId = static_cast(group.toUInt(&ok, 16)); if (!ok) { - qWarning().noquote() << QString("CloudConnect: could not convert group \"%1\" to MsgId").arg(group); + qCWarning(logCanRouting).noquote() << QString("could not convert group \"%1\" to MsgId in CAN mesg routing config file %2") + .arg(group, canRoutingPath); continue; } - msgHandlingIni.beginGroup(group); - const QString msgIdStr = msgHandlingIni.value(QStringLiteral("msg_id")).toString(); - 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, QString("%1").arg(quint16(msgId), 4, 16, QChar('0')).toUpper(), actionStr, topicStr); - msgHandlingIni.endGroup(); + canRoutingConfig.beginGroup(group); + const QString msgIdStr = canRoutingConfig.value(QStringLiteral("msg_id")).toString(); + const QString actionStr = canRoutingConfig.value(QStringLiteral("action")).toString().trimmed(); + const QString topicStr = canRoutingConfig.value(QStringLiteral("topic")).toString().trimmed(); + // qCInfo(logCanRouting).noquote() << QString("%1 (0x%2): action=%3, topic=%4") + // .arg(msgIdStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0')).toUpper(), actionStr, topicStr); + canRoutingConfig.endGroup(); - MsgHandling msgHandling; - msgHandling.action = actionMap.value(actionStr, MsgAction::Drop); - msgHandling.topic = topicMap.value(topicStr, CloudConnectFrame::Topic::NormalPriority); + CanRouting canRouting; + // canRouting.action = actionMap.value(actionStr, CanAction::Drop); + canRouting.action = actionMap.value(actionStr, CanAction::SendAlways); + canRouting.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") - .arg(actionStr).arg(QString("%1").arg(quint16(msgId), 4, 16, QChar('0').toUpper())); + qCWarning(logCanRouting).noquote() << QString("unknown message action \"%1\" for msgId=0x%2 — defaulting to Drop") + .arg(actionStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0').toUpper())); } if (topicStr.length() > 0 && !topicMap.contains(topicStr)) { - qWarning().noquote() << QString("CloudConnect: unknown message topic \"%1\" for msgId=0x%2 — defaulting to NormalPriority") - .arg(topicStr).arg(QString("%1").arg(quint16(msgId), 4, 16, QChar('0').toUpper())); + qCWarning(logCanRouting).noquote() << QString("unknown message topic \"%1\" for msgId=0x%2 — defaulting to NormalPriority") + .arg(topicStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0').toUpper())); } - _msgHandling.insert(msgId, msgHandling); + _canRouting.insert(msgId, canRouting); msgCount++; } - qInfo().noquote() << QString("CloudConnect: loaded message handling %1 (%2 entries)").arg(msgHandlingPath).arg(msgCount); + qCInfo(logCanRouting).noquote() << QString("loaded CAN message routing %1 (%2 entries)").arg(canRoutingPath).arg(msgCount); + return true; } /*! - * \brief CloudConnectController::onFrameReceive + * \brief CloudConnectController::onCanFrameReceive * \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) +void CloudConnectController::onCanFrameReceive(const QCanBusFrame &frame) { _dispatcher.onFrameReceive(Can::CanId(frame.frameId()), frame.payload()); } /*! - * \brief CloudConnectController::onMessageReceive - * \details Applies the message handling policy from MsgHandling.ini. + * \brief CloudConnectController::onCanMessageReceive + * \details Applies the CAN mesg routing policy from CanRouting.ini. * \param msg - the reassembled message */ -void CloudConnectController::onMessageReceive(const Can::Message &msg) +void CloudConnectController::onCanMessageReceive(const Can::Message &msg) { - const auto it = _msgHandling.constFind(msg.msgId); - if (it == _msgHandling.constEnd()) { - qInfo().noquote() << QString("CloudConnect: no action defined for %1 (0x%2), dropping") + const auto it = _canRouting.constFind(msg.msgId); + if (it == _canRouting.constEnd()) { + qCInfo(logCanRouting).noquote() << QString("no action defined for %1 (0x%2), dropping") .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); return; } - if (it->action == MsgAction::Drop) { - qInfo().noquote() << QString("CloudConnect: %1 (0x%2) dropped") + if (it->action == CanAction::Drop) { + qCInfo(logCanRouting).noquote() << QString("%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: %1 (0x%2) did not changed from previous, dropping") + auto &[received, cachedMsg] = _canCache[msg.msgId]; + if (it->action == CanAction::SendDelta && received && + cachedMsg.data.chopped(1) == msg.data.chopped(1)) { // move chops to payload sending + qCInfo(logCanRouting).noquote() << QString("%1 (0x%2) did not changed from previous, dropping") .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); return; } - const QByteArray payload = leahi::canMessageToProtobufByteArray( - QDateTime::currentDateTime(), - QStringLiteral("test_device"), - msg); + QByteArray payload; + if (leahi::canMessageToProtobufByteArray(QDateTime::currentDateTime(), msg, payload)) { + // Read the token out before it is advanced: the value handed to publish() is + // what the delivery ledger reports, so logging the post-increment value made + // every trace-back off by one. + qint32 msgId; + if (_mqttClient.publish(mqttTopic(it->topic), payload, msgId)) { + // Logs the token so an unacknowledged publish reported by the MQTT delivery + // ledger can be traced back to the message it carried. + qCWarning(logMqtt).noquote() << QString("%1 (0x%2) published (msgId=%3)") + .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')).arg(msgId); + } + else { + // TODO: spool the message if publish fails + qCWarning(logMqtt).noquote() << QString("could not publish %1 (0x%2), message lost") + .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); + } - // TODO: temporary for capture protobuf - // captureProtobuf(payload); - - const bool published = _mqttEnabled && _mqttClient.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')); + // NOTE: this records that the client ACCEPTED the message, not that AWS + // acknowledged it. Only the PUBACK counted in the MqttClient ledger + // means delivered. + // received = published; } else { - qWarning().noquote() << QString("CloudConnect: %1 (0x%2) published") + qCWarning(logProto).noquote() << QString("could not serialize %1 (0x%2) message to protobuf, message dropped") .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 + * \param topic - message class from the CAN message routing INI * \return Topic of the form {prefix}/{deviceId}/{suffix}. */ QString CloudConnectController::mqttTopic(CloudConnectFrame::Topic topic) const @@ -291,98 +336,135 @@ { CloudConnectFrame::Topic::CloudSyncLogFile, QStringLiteral("cs_log") }, }; - return QString("%1/%2/%3").arg(_topicPrefix, _deviceId, - suffixes.value(topic, QStringLiteral("normal"))); + // return suffixes.value(topic, QStringLiteral("normal")); + Q_UNUSED(topic) + return QString("%1/%2/%3").arg(_topicPrefix, _deviceId, "clinical"); } /*! - * \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 + * \brief CloudConnectController::onCloudStateChanged + * \details Re-establishes the echo subscriptions on every new session, and + * retires anything still awaiting an echo when one ends. + * \param connected - true once the broker has accepted the session + * \note Subscribing here rather than in connectToCloud() is required, not + * stylistic: open() only starts the attempt, and a SUBSCRIBE sent + * before the CONNACK is rejected by the client. */ -void CloudConnectController::captureProtobuf(const QByteArray &payload) +void CloudConnectController::onCloudStateChanged(QMqttClient::ClientState state) { - 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; - }; + switch (state) { + case QMqttClient::Disconnected: + // clear any pending messages + // expireInFlight(true); + break; + case QMqttClient::Connecting: + break; + case QMqttClient::Connected: + (void)_mqttClient.subscribe(mqttTopic(CloudConnectFrame::Topic::NormalPriority)); + break; + default: + break; + } +} - 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(); - } - } +/*! + * \brief CloudConnectController::onCloudMessageStatusChanged + * \details Correlates one message delivered by the broker back to the publish + * that produced it and credits that msgId. + * \param msgId - message identifier + * \param status - new message status + * \param properties - additional properties specified by the server/broker + */ +void CloudConnectController::onCloudMessageStatusChanged(qint32 id, QMqtt::MessageStatus status, + const QMqttMessageStatusProperties &properties) +{ + Q_UNUSED(id) // SQ + Q_UNUSED(properties) // SQ + switch (status) { + case QMqtt::MessageStatus::Unknown: + qCInfo(logMqtt).noquote() << "message status changed to Unknown"; // SQ + break; + case QMqtt::MessageStatus::Published: + qCInfo(logMqtt).noquote() << "message status changed to Published"; // SQ + break; + case QMqtt::MessageStatus::Acknowledged: + qCInfo(logMqtt).noquote() << "message status changed to Acknowledged"; // SQ + break; + case QMqtt::MessageStatus::Received: + qCInfo(logMqtt).noquote() << "message status changed to Received"; // SQ + break; + case QMqtt::MessageStatus::Released: + qCInfo(logMqtt).noquote() << "message status changed to Released"; // SQ + break; + case QMqtt::MessageStatus::Completed: + qCInfo(logMqtt).noquote() << "message status changed to Completed"; // SQ + break; + default: + break; } +} - 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(); +/*! + * \brief CloudConnectController::onCloudMessageReceived + * \details Correlates one message delivered by the broker back to the publish + * that produced it and credits that msgId. + * \param topic - topic the message was delivered on + * \param payload - serialised protobuf message + * \note The msgId is read from the message itself: every Leahi message carries + * Header at field 1, and Envelope exists to parse exactly that much + * without knowing the concrete message type. + */ +void CloudConnectController::onCloudMessageReceived(const QString &topic, const QByteArray &payload) +{ + qCInfo(logMqtt).noquote() << QString("received message with topic=%1").arg(topic); // SQ + + // TODO: filter by topic + + leahi::messages::Envelope envelope; + if (!envelope.ParseFromArray(payload.constData(), payload.size()) || !envelope.has_header()) { + // Unknown fields parse cleanly in proto3, so a foreign payload usually + // lands here as a header-less Envelope rather than as a parse failure. + qCWarning(logMqtt).noquote() << QString("received mesg (size=%1) for topic %2 do not contain a valid protobuf payload, ignoring") + .arg(payload.size()).arg(topic); + return; } - 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"); - } + const leahi::messages::Header &header = envelope.header(); + const Can::MsgId msgId = Can::MsgId(header.msgid()); - _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(); - } - } + // No serial here: Header carries only timestamp/sequence/msgId, and the device + // identity travels in the topic (logged by handlePublishPacket) rather than the + // payload. Printing the timestamp instead also proves updateHeader populated it. + const qint64 headerMsecs = header.timestamp().seconds() * 1000 + header.timestamp().nanos() / 1000000; + qCInfo(logProto).noquote() << QString("header: msgId=0x%1 (%2), seq=%3, timestamp=%4") + .arg(QString("%1").arg(msgId, 4, 16, QChar('0')).toUpper(), leahi::msgIdString(static_cast(msgId))) + .arg(header.sequence()) + .arg(QDateTime::fromMSecsSinceEpoch(headerMsecs).toString(Qt::ISODateWithMs)); + + const std::string &typeName = leahi::msgIdToProtoName(msgId); + if (typeName.empty()) { + qCWarning(logProto).noquote() << QString("no protobuf mapping for message with msgId=0x%1, ignoring").arg(msgId, 4, 16, QChar('0')); + return; } - 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(); - } - } - } + const google::protobuf::Descriptor *descriptor = + google::protobuf::DescriptorPool::generated_pool()->FindMessageTypeByName(typeName); + if (descriptor == nullptr) { + qCWarning(logProto).noquote() << QString("no descriptor for protobuf message %1, ignoring").arg(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()))) { + qCWarning(logProto).noquote() << QString("could not parse protobuf message %1, ignoring").arg(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); + qCDebug(logProto).noquote() << QString::fromStdString(typeName) << ":" << Qt::endl << QString::fromStdString(json); }