Index: lib/Comms/src/MqttClient.cpp =================================================================== diff -u -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 -raaebfee335c74b0250864a6dce0555f866adadea --- lib/Comms/src/MqttClient.cpp (.../MqttClient.cpp) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) +++ lib/Comms/src/MqttClient.cpp (.../MqttClient.cpp) (revision aaebfee335c74b0250864a6dce0555f866adadea) @@ -10,12 +10,16 @@ * \date (original) 30-Jul-2026 * */ +#include + #include #include #include #include #include #include +#include +#include #include #include "MqttClient.h" @@ -44,13 +48,14 @@ /*! * \brief MqttClient::init * \details Validates the configuration and builds the TLS configuration from - * the certificate paths it names. No client is created and no - * connection is attempted here; that is open()'s job. - * \param config - parameters for the session, retained for the next open() - * \return true if the configuration is usable, otherwise false + * the certificate and key paths. Connection to server is not attempted + * until open() is called. + * \note This function does nothing if a client connection already exists. * \note keyPath is required. caPath and certPath are optional: without * caPath the system CA set verifies the broker, and without certPath * the session is server-authenticated only rather than mTLS. + * \param config - parameters for the session, retained for the next open() + * \return true if the configuration is usable, otherwise false */ bool MqttClient::init(const Config &config) { @@ -60,39 +65,54 @@ } QSslConfiguration ssl = QSslConfiguration::defaultConfiguration(); - if (!config.caPath.isEmpty()) { + if (QFile::exists(config.caPath)) { const QList cas = QSslCertificate::fromPath(config.caPath); if (cas.isEmpty()) { qCCritical(logMqtt).noquote() << QString("%1 is not a CA certificate").arg(config.caPath); return false; } ssl.setCaCertificates(cas); } + else { + qCCritical(logMqtt).noquote() << QString("CA certificate file %1 does not exists").arg(config.caPath); + return false; + } - if (!config.certPath.isEmpty()) { + if (QFile::exists(config.certPath)) { const QList certs = QSslCertificate::fromPath(config.certPath); if (certs.isEmpty()) { qCCritical(logMqtt).noquote() << QString("%1 is not a certificate").arg(config.certPath); return false; } ssl.setLocalCertificate(certs.first()); } - - QFile keyFile(config.keyPath); - if (!keyFile.open(QIODevice::ReadOnly)) { - qCCritical(logMqtt).noquote() << QString("could not read private key %1").arg(config.keyPath); + else { + qCCritical(logMqtt).noquote() << QString("certificate file %1 does not exists").arg(config.certPath); return false; } - const QByteArray pem = keyFile.readAll(); - QSslKey key(pem, QSsl::Rsa, QSsl::Pem); - if (key.isNull()) { - key = QSslKey(pem, QSsl::Ec, QSsl::Pem); + + if (QFile::exists(config.keyPath)) { + QFile keyFile(config.keyPath); + if (!keyFile.open(QIODevice::ReadOnly)) { + qCCritical(logMqtt).noquote() << QString("could not read private key %1").arg(config.keyPath); + return false; + } + const QByteArray pem = keyFile.readAll(); + keyFile.close(); + QSslKey key(pem, QSsl::Rsa, QSsl::Pem); + if (key.isNull()) { + key = QSslKey(pem, QSsl::Ec, QSsl::Pem); + } + if (key.isNull()) { + qCCritical(logMqtt).noquote() << QString("private key %1 is not a PEM, RSA, or EC key").arg(config.keyPath); + return false; + } + ssl.setPrivateKey(key); } - if (key.isNull()) { - qCCritical(logMqtt).noquote() << QString("private key %1 is not a PEM, RSA, or EC key").arg(config.keyPath); + else { + qCCritical(logMqtt).noquote() << QString("key file %1 does not exist").arg(config.keyPath); return false; } - ssl.setPrivateKey(key); _sslConfig = ssl; _config = config; @@ -107,8 +127,6 @@ * \return true if the attempt was started * \note Must be called on this object's thread; the client's socket belongs * to the thread that constructs it. - * \note The attempt is asynchronous, so a true return only means the client - * accepted the request. The outcome arrives via onStateChanged(). */ bool MqttClient::open() { @@ -126,7 +144,13 @@ _client->setCleanSession(_config.cleanSession); connect(_client, &QMqttClient::stateChanged, this, &MqttClient::onStateChanged); - connect(_client, &QMqttClient::messageSent, this, &MqttClient::onMessageSent); + connect(_client, &QMqttClient::messageStatusChanged, this, &MqttClient::didMessageStatusChanged); + // Fires for every delivery on this connection, whichever subscription matched. + connect(_client, &QMqttClient::messageReceived, this, + [this](const QByteArray &message, const QMqttTopicName &topic) { + Q_EMIT didMessageReceived(topic.name(), message); + } + ); connect(_client, &QMqttClient::errorChanged, this, [](QMqttClient::ClientError error) { if (error != QMqttClient::NoError) { @@ -136,8 +160,9 @@ ); qCInfo(logMqtt).noquote() << QString("attempting to connect to %1:%2 as %3") - .arg(_config.endpoint).arg(_config.port).arg(_config.clientId); + .arg(_config.endpoint).arg(_config.port).arg(_config.clientId); _client->connectToHostEncrypted(_sslConfig); + return true; } @@ -157,7 +182,7 @@ QMqttClient *client = _client; _client = nullptr; - failPending(); + // stop any signals coming from client before modifying it client->disconnect(this); // flush to force the socket to send the Disconnect so bytesToWrite hopefully @@ -187,32 +212,76 @@ /*! * \brief MqttClient::publish * \details Hands one message to the client for QoS 1 delivery. - * \param topic - fully resolved MQTT topic - * \param payload - serialised message bytes - * \param messageId - caller token echoed back on acknowledgement, or -1 for none + * \param[in] topic - fully resolved MQTT topic + * \param[in] payload - serialised message bytes + * \param[out] msgId - identifier assigned by QMqttClient::publish to track message + * \param[in] qos - QoS level (default=1) * \return true if the client accepted the message. * \note Return status does NOT indicate delivery, only that the client * accepted the message for sending. The PUBACK that confirms delivery * arrives later in onMessageSent(). */ -bool MqttClient::publish(const QString &topic, const QByteArray &payload, qint64 messageId) +bool MqttClient::publish(const QString &topic, const QByteArray &payload, qint32 &msgId, quint8 qos) { if (_client == nullptr || _client->state() != QMqttClient::Connected) { + msgId = -1; return false; } - const qint32 id = _client->publish(QMqttTopicName(topic), payload, 1, false); - if (id == -1) { - qCWarning(logMqtt).noquote() << QString("client rejected publishing to topic %1").arg(topic); + msgId = _client->publish(QMqttTopicName(topic), payload, qos, false); + if (msgId != -1) { + // qCDebug(logMqtt).noquote() << QString("publish accepted (msgId=%1): topic=%2, qos=%3") + // .arg(msgId).arg(topic).arg(qos); + return true; + } + else { + qCWarning(logMqtt).noquote() << QString("client rejected publish: topic=%1, qos=%2") + .arg(topic).arg(qos); return false; } +} - _pending.insert(id, messageId); - // TODO: determine if messages waiting on a PUBACK need to be re-sent - // set a timeout interval to indicate when a message needs to be - // resent and have a timer periodically check _pending for messages - // that have met that threshold +/*! + * \brief MqttClient::subscribe + * \details Subscribes this session to a topic filter. Deliveries arrive on + * didMessageReceived(), which is not per-subscription: it carries the + * topic so the receiver can demultiplex. + * \param topicFilter - MQTT topic filter, wildcards allowed + * \param qos - requested maximum QoS for deliveries on this filter + * \return true if the client accepted the request. + * \note Return status does NOT mean the subscription is active. The SUBACK + * arrives later; until it does, the broker may drop matching messages. + * The granted QoS can also be lower than requested, which is not an error. + * \note Subscriptions live on the QMqttClient and die with it, so they must be + * re-established after every reconnect (see didConnectionChange()). + */ +bool MqttClient::subscribe(const QString &topicFilter, quint8 qos) +{ + if (_client == nullptr || _client->state() != QMqttClient::Connected) { + qCWarning(logMqtt).noquote() << QString("cannot subscribe to %1 while disconnected").arg(topicFilter); + return false; + } + QMqttSubscription *subscription = _client->subscribe(QMqttTopicFilter(topicFilter), qos); + if (subscription == nullptr) { + qCWarning(logMqtt).noquote() << QString("client rejected subscribing to topic %1").arg(topicFilter); + return false; + } + + // A subscription AWS IoT refuses (a filter outside the policy's iot:Receive or + // iot:Subscribe resources) fails here rather than at publish time, and without + // this log the only symptom is that no message ever arrives. + connect(subscription, &QMqttSubscription::stateChanged, this, + [topicFilter](QMqttSubscription::SubscriptionState state) { + if (state == QMqttSubscription::Subscribed) { + qCInfo(logMqtt).noquote() << QString("subscribed to %1").arg(topicFilter); + } + else if (state == QMqttSubscription::Error) { + qCWarning(logMqtt).noquote() << QString("subscription to topic %1 failed").arg(topicFilter); + } + } + ); + return true; } @@ -226,52 +295,24 @@ if (_client != nullptr) { switch (_client->state()) { case QMqttClient::Disconnected: - // NOTE: Receiving state change to Disconnected here means that connection - // was not done by user of this class calling close() and is most - // likely caused by externally factors, like the server side closing - // the connection. + // NOTE: close() disconnects all signals from _client so further signals, + // like stateChanged, will no longer be received qCInfo(logMqtt).noquote() << "client disconnected"; - failPending(); _client->deleteLater(); _client = nullptr; // TODO: reconnect on disconnect - // set a flag reconnect flag indicating if connection should try to - // be re-established + Q_EMIT didStateChanged(QMqttClient::Disconnected); break; case QMqttClient::Connecting: qCInfo(logMqtt).noquote() << "client connecting"; + Q_EMIT didStateChanged(_client->state()); break; case QMqttClient::Connected: qCInfo(logMqtt).noquote() << "client connected"; + // Subscriptions did not survive the previous client, so this is where + // a receiver re-establishes them. + Q_EMIT didStateChanged(_client->state()); break; } } } - -/*! - * \brief MqttClient::onMessageSent - * \details Clears one QoS 1 publish from _pending on PUBACK. - * \param id - QMqttClient's identifier for the acknowledged publish - */ -void MqttClient::onMessageSent(qint32 id) -{ - if (!_pending.contains(id)) { - qCWarning(logMqtt).noquote() << QString("received PUBACK for an unknown published message with id %1").arg(id); - return; - } - _pending.take(id); -} - -/*! - * \brief MqttClient::failPending - * \details Abandons every publish still awaiting a PUBACK. - */ -void MqttClient::failPending() -{ - if (_pending.isEmpty()) { - return; - } - - qCWarning(logMqtt).noquote() << QString("Failing %1 unacknowledged publishes").arg(_pending.size()); - _pending.clear(); -}