Revision control
++ // Copyright (C) 2017 The Qt Company Ltd.
++ // SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
++ // Qt-Security score:critical reason:network-protocol
++
++ #include "qmqttconnection_p.h"
++ #include "qmqttconnectionproperties_p.h"
++ #include "qmqttcontrolpacket_p.h"
++ #include "qmqttmessage_p.h"
++ #include "qmqttpublishproperties_p.h"
++ #include "qmqttsubscription_p.h"
++ #include "qmqttclient_p.h"
++ #include "transportLayers/mqtt_websocket_io_p.h"
++ #include "transportLayers/mqtt_secure_websocket_io_p.h"
++
++ #include <QtCore/QLoggingCategory>
++ #include <QtNetwork/QSslSocket>
++ #include <QtNetwork/QTcpSocket>
++
++ #include <limits>
++ #include <cstdint>
++
++ QT_BEGIN_NAMESPACE
++
++ Q_LOGGING_CATEGORY(lcMqttConnection, "qt.mqtt.connection")
++ Q_STATIC_LOGGING_CATEGORY(lcMqttConnectionVerbose, "qt.mqtt.connection.verbose");
++
++ template <typename T>
++ T QMqttConnection::readBufferTyped(qint64 *dataSize)
++ {
++ Q_STATIC_ASSERT(std::is_integral<T>::value);
++
++ T result = 0;
++ if (Q_UNLIKELY(dataSize != nullptr && *dataSize < qint64(sizeof(result)))) {
++ qCWarning(lcMqttConnection) << "Attempt to read past the data";
++ *dataSize = 0;
++ return result;
++ }
++ const bool isOk = readBuffer(reinterpret_cast<char *>(&result), sizeof(result));
++ if (dataSize != nullptr) {
++ if (isOk)
++ *dataSize -= sizeof(result);
++ else
++ *dataSize = 0;
++ }
++ return qFromBigEndian(result);
++ }
++
++ template<>
++ QByteArray QMqttConnection::readBufferTyped(qint64 *dataSize)
++ {
++ const quint16 size = readBufferTyped<quint16>(dataSize);
++ if (Q_UNLIKELY(dataSize != nullptr && *dataSize < qint64(size))) {
++ qCWarning(lcMqttConnection) << "Attempt to read past the data";
++ *dataSize = 0;
++ return QByteArray();
++ }
++ QByteArray ba(int(size), Qt::Uninitialized);
++ const bool isOk = readBuffer(ba.data(), size);
++ if (!isOk) {
++ if (dataSize != nullptr)
++ *dataSize = 0;
++ return QByteArray();
++ }
++ if (dataSize != nullptr)
++ *dataSize -= size;
++ return ba;
++ }
++
++ template<>
++ QString QMqttConnection::readBufferTyped(qint64 *dataSize)
++ {
++ return QString::fromUtf8(readBufferTyped<QByteArray>(dataSize));
++ }
++
++ bool QMqttConnection::isPendingUnsubscribe(QMqttSubscription *sub) const
++ {
++ for (const auto [key, value] : m_pendingUnsubscriptions.asKeyValueRange()) {
++ if (value == sub)
++ return true;
++ }
++ return false;
++ }
++
++ QMqttSubscription *QMqttConnection::findActiveSubscription(const QMqttTopicFilter &topic) const
++ {
++ const auto range = m_activeSubscriptions.equal_range(topic);
++ for (auto it = range.first; it != range.second; ++it) {
++ if (!isPendingUnsubscribe(it.value()))
++ return it.value();
++ }
++ return nullptr;
++ }
++
++ QMqttConnection::QMqttConnection(QObject *parent) : QObject(parent)
++ {
++ }
++
++ QMqttConnection::~QMqttConnection()
++ {
++ if (m_internalState == BrokerConnected)
++ sendControlDisconnect();
++ m_transport = nullptr;
++ }
++
++ void QMqttConnection::timerEvent(QTimerEvent *event)
++ {
++ if (Q_LIKELY(event->timerId() == m_pingTimer.timerId())) {
++ sendControlPingRequest();
++ return;
++ }
++
++ QObject::timerEvent(event);
++ }
++
++ void QMqttConnection::disconnectAndResetTransport()
++ {
++ if (m_transport) {
++ if (m_connectedTransport) {
++ //
++ // Looks weird. We know that m_transport is not set from
++ // the outside because m_connectedTransport is only set
++ // by us for the websocket cases, or in ensureTransport.
++ //
++ // Thus we can disconnect everything.
++ //
++ disconnect(m_transport.get());
++ } else {
++ //
++ // m_transport is set from outside with setTransport,
++ // there might be other connections that shall not
++ // be removed.
++ //
++ disconnect(m_transport.get(), &QIODevice::aboutToClose, this, &QMqttConnection::transportConnectionClosed);
++ disconnect(m_transport.get(), &QIODevice::readyRead, this, &QMqttConnection::transportReadyRead);
++ }
++ }
++ m_transport = nullptr;
++ m_transportType = QMqttClient::TransportType::IODevice;
++ }
++
++ void QMqttConnection::setTransport(QIODevice *device, QMqttClient::TransportType transport)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << device << " Type:" << transport;
++
++ disconnectAndResetTransport();
++
++ m_transport = std::shared_ptr<QIODevice>(device, [](QIODevice *) { ; });
++ m_transportType = transport;
++ m_connectedTransport = false;
++ m_transportIsSet = true;
++
++ connect(m_transport.get(), &QIODevice::aboutToClose, this, &QMqttConnection::transportConnectionClosed);
++ connect(m_transport.get(), &QIODevice::readyRead, this, &QMqttConnection::transportReadyRead);
++ }
++
++ void QMqttConnection::connectTransport(QMqttClient::TransportType transportType, const std::shared_ptr<QIODevice> &transport)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << " Type:" << transportType;
++
++ disconnectAndResetTransport();
++
++ m_transportType = transportType;
++ m_transport = transport;
++ m_transportIsSet = true;
++ m_connectedTransport = true;
++
++ if (!m_transport) {
++ qWarning(lcMqttConnection) << " No transport created for " << transportType;
++ return;
++ }
++
++ QObject::connect(m_transport.get(), &QIODevice::aboutToClose, this, &QMqttConnection::transportConnectionClosed);
++ QObject::connect(m_transport.get(), &QIODevice::readyRead, this, &QMqttConnection::transportReadyRead);
++
++ switch (m_transportType) {
++ case QMqttClient::TransportType::IODevice:
++ break;
++ case QMqttClient::TransportType::AbstractSocket:
++ QObject::connect(qobject_cast<QAbstractSocket *>(m_transport.get()), &QAbstractSocket::connected, this, &QMqttConnection::transportConnectionEstablished);
++ QObject::connect(qobject_cast<QAbstractSocket *>(m_transport.get()), &QAbstractSocket::disconnected, this, &QMqttConnection::transportConnectionClosed);
++ QObject::connect(qobject_cast<QAbstractSocket *>(m_transport.get()), &QAbstractSocket::errorOccurred, this, &QMqttConnection::transportError);
++ break;
++ case QMqttClient::TransportType::WebSocket:
++ #ifdef QT_MQTT_WITH_WEBSOCKETS
++ QObject::connect(qobject_cast<QMqttWebSocketIO *>(m_transport.get()), &QMqttWebSocketIO::connected, this, &QMqttConnection::transportConnectionEstablished);
++ QObject::connect(qobject_cast<QMqttWebSocketIO *>(m_transport.get()), &QMqttWebSocketIO::disconnected, this, &QMqttConnection::transportConnectionClosed);
++ QObject::connect(qobject_cast<QMqttWebSocketIO *>(m_transport.get()), &QMqttWebSocketIO::errorOccurred, this, &QMqttConnection::transportError);
++ #endif
++ break;
++
++ case QMqttClient::TransportType::SecureSocket:
++ #ifndef QT_NO_SSL
++ QObject::connect(qobject_cast<QSslSocket *>(m_transport.get()), &QSslSocket::encrypted, this, &QMqttConnection::transportConnectionEstablished);
++ QObject::connect(qobject_cast<QSslSocket *>(m_transport.get()), &QAbstractSocket::disconnected, this, &QMqttConnection::transportConnectionClosed);
++ QObject::connect(qobject_cast<QSslSocket *>(m_transport.get()), &QAbstractSocket::errorOccurred, this, &QMqttConnection::transportError);
++ #endif
++ break;
++
++ case QMqttClient::TransportType::SecureWebSocket:
++ #ifdef QT_MQTT_WITH_WEBSOCKETS
++ QObject::connect(qobject_cast<QMqttSecureWebSocketIO *>(m_transport.get()), &QMqttSecureWebSocketIO::encrypted, this, &QMqttConnection::transportConnectionEstablished);
++ QObject::connect(qobject_cast<QMqttSecureWebSocketIO *>(m_transport.get()), &QMqttSecureWebSocketIO::disconnected, this, &QMqttConnection::transportConnectionClosed);
++ QObject::connect(qobject_cast<QMqttSecureWebSocketIO *>(m_transport.get()), &QMqttSecureWebSocketIO::errorOccurred, this, &QMqttConnection::transportError);
++ #endif
++ break;
++
++ default:
++ Q_ASSERT(false);
++ break;
++ } // end switch
++ }
++
++ QIODevice *QMqttConnection::transport() const
++ {
++ return m_transport.get();
++ }
++
++ bool QMqttConnection::ensureTransport(bool createSecureIfNeeded)
++ {
++ Q_UNUSED(createSecureIfNeeded); // QT_NO_SSL
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << m_transport.get();
++
++ if (m_transportIsSet)
++ return true; // transport set by setTransport
++
++ disconnectAndResetTransport();
++
++ // We are asked to create a transport layer
++ if (m_clientPrivate->m_hostname.isEmpty() || m_clientPrivate->m_port == 0) {
++ qCDebug(lcMqttConnection) << "No hostname specified, or port is 0. Not able to create a transport layer.";
++ return false;
++ }
++
++ #ifdef Q_OS_WASM
++ qCWarning(lcMqttConnection) << "For WebAssembly, use connectToHostWebSocket[Encrypted]";
++ return false;
++ #endif
++
++ #ifndef QT_NO_SSL
++ if (createSecureIfNeeded) {
++ auto socket = std::make_shared<QSslSocket>();
++ m_transport = socket;
++ m_connectedTransport = true;
++ m_transportType = QMqttClient::SecureSocket;
++
++ QObject::connect(socket.get(), &QSslSocket::encrypted, this, &QMqttConnection::transportConnectionEstablished);
++ QObject::connect(socket.get(), &QAbstractSocket::disconnected, this, &QMqttConnection::transportConnectionClosed);
++ QObject::connect(socket.get(), &QAbstractSocket::errorOccurred, this, &QMqttConnection::transportError);
++ }
++ #endif
++
++ if (!m_transport) {
++ auto socket = std::make_shared<QTcpSocket>();
++ m_transport = socket;
++ m_connectedTransport = true;
++ m_transportType = QMqttClient::AbstractSocket;
++
++ QObject::connect(socket.get(), &QAbstractSocket::connected, this, &QMqttConnection::transportConnectionEstablished);
++ QObject::connect(socket.get(), &QAbstractSocket::disconnected, this, &QMqttConnection::transportConnectionClosed);
++ QObject::connect(socket.get(), &QAbstractSocket::errorOccurred, this, &QMqttConnection::transportError);
++ }
++ QObject::connect(m_transport.get(), &QIODevice::aboutToClose, this, &QMqttConnection::transportConnectionClosed);
++ QObject::connect(m_transport.get(), &QIODevice::readyRead, this, &QMqttConnection::transportReadyRead);
++
++ return true;
++ }
++
++ bool QMqttConnection::ensureTransportOpen(const QString &sslPeerName)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << m_transportType;
++
++ if (m_transportType == QMqttClient::IODevice) {
++ if (m_transport->isOpen())
++ return sendControlConnect();
++
++ if (!m_transport->open(QIODevice::ReadWrite)) {
++ qCDebug(lcMqttConnection) << "Could not open Transport IO device.";
++ m_internalState = BrokerDisconnected;
++ return false;
++ }
++ return sendControlConnect();
++ }
++
++ if (m_transportType == QMqttClient::AbstractSocket) {
++ auto socket = qobject_cast<QTcpSocket *>(m_transport.get());
++ Q_ASSERT(socket);
++ if (socket->state() == QAbstractSocket::ConnectedState)
++ return sendControlConnect();
++
++ m_internalState = BrokerConnecting;
++ socket->connectToHost(m_clientPrivate->m_hostname, m_clientPrivate->m_port);
++ }
++ else if (m_transportType == QMqttClient::WebSocket) {
++ #ifdef QT_MQTT_WITH_WEBSOCKETS
++ auto socket = qobject_cast<QMqttWebSocketIO *>(m_transport.get());
++ Q_ASSERT(socket);
++ if (socket->state() == QAbstractSocket::ConnectedState)
++ return sendControlConnect();
++
++ m_internalState = BrokerConnecting;
++ socket->connectToHost(m_clientPrivate->m_hostname, m_clientPrivate->m_port, m_clientPrivate->m_protocolVersion);
++ #endif
++ }
++ else if (m_transportType == QMqttClient::SecureWebSocket) {
++ #ifdef QT_MQTT_WITH_WEBSOCKETS
++ auto socket = qobject_cast<QMqttSecureWebSocketIO *>(m_transport.get());
++ Q_ASSERT(socket);
++ if (socket->state() == QAbstractSocket::ConnectedState)
++ return sendControlConnect();
++
++ m_internalState = BrokerConnecting;
++ socket->connectToHostEncrypted(m_clientPrivate->m_hostname, m_clientPrivate->m_port, m_clientPrivate->m_protocolVersion);
++ #endif
++ }
++ #ifndef QT_NO_SSL
++ else if (m_transportType == QMqttClient::SecureSocket) {
++ auto socket = qobject_cast<QSslSocket *>(m_transport.get());
++ Q_ASSERT(socket);
++ if (socket->state() == QAbstractSocket::ConnectedState)
++ return sendControlConnect();
++
++ m_internalState = BrokerConnecting;
++ if (!m_sslConfiguration.isNull())
++ socket->setSslConfiguration(m_sslConfiguration);
++ socket->connectToHostEncrypted(m_clientPrivate->m_hostname, m_clientPrivate->m_port, sslPeerName);
++ }
++ #endif
++ Q_UNUSED(sslPeerName);
++ return true;
++ }
++
++ bool QMqttConnection::sendControlConnect()
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO;
++
++ QMqttControlPacket packet(QMqttControlPacket::CONNECT);
++
++ // Variable header
++ // 3.1.2.1 Protocol Name
++ // 3.1.2.2 Protocol Level
++ switch (m_clientPrivate->m_protocolVersion) {
++ case QMqttClient::MQTT_3_1:
++ packet.append("MQIsdp");
++ packet.append(char(3)); // Version 3.1
++ break;
++ case QMqttClient::MQTT_3_1_1:
++ packet.append("MQTT");
++ packet.append(char(4)); // Version 3.1.1
++ break;
++ case QMqttClient::MQTT_5_0:
++ packet.append("MQTT");
++ packet.append(char(5)); // Version 5.0
++ break;
++ }
++
++ // 3.1.2.3 Connect Flags
++ quint8 flags = 0;
++ // Clean session
++ if (m_clientPrivate->m_cleanSession)
++ flags |= 1 << 1;
++
++ if (!m_clientPrivate->m_willTopic.isEmpty()) {
++ flags |= 1 << 2;
++ if (m_clientPrivate->m_willQoS > 2) {
++ qCDebug(lcMqttConnection) << "Invalid Will QoS specified.";
++ return false;
++ }
++ if (m_clientPrivate->m_willQoS == 1)
++ flags |= 1 << 3;
++ else if (m_clientPrivate->m_willQoS == 2)
++ flags |= 1 << 4;
++ if (m_clientPrivate->m_willRetain)
++ flags |= 1 << 5;
++ }
++ if (m_clientPrivate->m_username.size())
++ flags |= 1 << 7;
++
++ if (m_clientPrivate->m_password.size())
++ flags |= 1 << 6;
++
++ packet.append(char(flags));
++
++ // 3.1.2.10 Keep Alive
++ packet.append(m_clientPrivate->m_keepAlive);
++
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
++ packet.appendRaw(writeConnectProperties());
++
++ // 3.1.3 Payload
++ // 3.1.3.1 Client Identifier
++ const QByteArray clientStringArray = m_clientPrivate->m_clientId.toUtf8();
++ if (clientStringArray.size()) {
++ packet.append(clientStringArray);
++ } else {
++ packet.append(char(0));
++ packet.append(char(0));
++ }
++
++ if (!m_clientPrivate->m_willTopic.isEmpty()) {
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
++ packet.appendRaw(writeLastWillProperties());
++
++ packet.append(m_clientPrivate->m_willTopic.toUtf8());
++ packet.append(m_clientPrivate->m_willMessage);
++ }
++
++ if (m_clientPrivate->m_username.size())
++ packet.append(m_clientPrivate->m_username.toUtf8());
++
++ if (m_clientPrivate->m_password.size())
++ packet.append(m_clientPrivate->m_password.toUtf8());
++
++ m_internalState = BrokerWaitForConnectAck;
++ m_missingData = 0;
++
++ if (!writePacketToTransport(packet)) {
++ qCDebug(lcMqttConnection) << "Could not write CONNECT frame to transport.";
++ return false;
++ }
++ return true;
++ }
++
++ void QMqttConnection::setReadBufferSize(qint64 size)
++ {
++ // Nothing to apply yet. QMqttClient applies the size again once a transport is
++ // set or created.
++ if (!m_transport)
++ return;
++ switch (m_transportType) {
++ case QMqttClient::TransportType::IODevice:
++ if (size != 0)
++ qCWarning(lcMqttClient) << "TransportType::IODevice doesn't support a read buffer limit.";
++ break;
++ case QMqttClient::TransportType::AbstractSocket:
++ case QMqttClient::TransportType::SecureSocket:
++ // QSslSocket is a QAbstractSocket and setReadBufferSize() is declared on the
++ // base class, so this also works in a build without SSL support.
++ setSocketBufferSize<QAbstractSocket>(size);
++ break;
++ case QMqttClient::TransportType::WebSocket:
++ case QMqttClient::TransportType::SecureWebSocket:
++ #ifdef QT_MQTT_WITH_WEBSOCKETS
++ // Both web socket transports share the base class holding the QWebSocket.
++ setSocketBufferSize<QMqttWebSocketIOBase>(size);
++ #endif
++ break;
++ }
++ }
++
++ bool QMqttConnection::sendControlAuthenticate(const QMqttAuthenticationProperties &properties)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO;
++
++ QMqttControlPacket packet(QMqttControlPacket::AUTH);
++
++ switch (m_internalState) {
++ case BrokerDisconnected:
++ case BrokerConnecting:
++ case ClientDestruction:
++ qCDebug(lcMqttConnection) << "Using AUTH while disconnected.";
++ return false;
++ case BrokerWaitForConnectAck:
++ qCDebug(lcMqttConnection) << "AUTH while connecting, set continuation flag.";
++ packet.append(char(QMqtt::ReasonCode::ContinueAuthentication));
++ break;
++ case BrokerConnected:
++ qCDebug(lcMqttConnection) << "AUTH while connected, initiate re-authentication.";
++ packet.append(char(QMqtt::ReasonCode::ReAuthenticate));
++ break;
++ }
++
++ packet.appendRaw(writeAuthenticationProperties(properties));
++
++ if (!writePacketToTransport(packet)) {
++ qCDebug(lcMqttConnection) << "Could not write AUTH frame to transport.";
++ return false;
++ }
++
++ return true;
++ }
++
++ qint32 QMqttConnection::sendControlPublish(const QMqttTopicName &topic,
++ const QByteArray &message,
++ quint8 qos,
++ bool retain,
++ const QMqttPublishProperties &properties)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << topic << " Size:" << message.size() << " bytes."
++ << "QoS:" << qos << " Retain:" << retain;
++
++ if (!topic.isValid())
++ return -1;
++
++ quint8 header = QMqttControlPacket::PUBLISH;
++ if (qos == 1)
++ header |= 0x02;
++ else if (qos == 2)
++ header |= 0x04;
++
++ if (retain)
++ header |= 0x01;
++
++ QSharedPointer<QMqttControlPacket> packet(new QMqttControlPacket(header));
++ // topic alias
++ QMqttPublishProperties publishProperties(properties);
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
++ // 3.3.4 A PUBLISH packet sent from a Client to a Server MUST NOT contain a Subscription Identifier
++ if (publishProperties.availableProperties() & QMqttPublishProperties::SubscriptionIdentifier) {
++ qCWarning(lcMqttConnection) << "SubscriptionIdentifier must not be specified for publish.";
++ return -1;
++ }
++
++ const quint16 topicAlias = publishProperties.topicAlias();
++ if (topicAlias > 0) { // User specified topic Alias
++ // m_publishAliases is sized from the maximum the server declared in CONNACK,
++ // so bounds-check against the list itself rather than against a property value.
++ if (topicAlias > m_publishAliases.size()) {
++ qCDebug(lcMqttConnection) << "TopicAlias publish: overflow.";
++ return -1;
++ }
++ if (m_publishAliases.at(topicAlias - 1) != topic) {
++ qCDebug(lcMqttConnection) << "TopicAlias publish: Assign:" << topicAlias << ":" << topic;
++ m_publishAliases[topicAlias - 1] = topic;
++ packet->append(topic.name().toUtf8());
++ } else {
++ qCDebug(lcMqttConnectionVerbose) << "TopicAlias publish: Reuse:" << topicAlias;
++ packet->append(quint16(0));
++ }
++ } else if (m_publishAliases.size() > 0) { // Automatic module alias assignment
++ int autoAlias = m_publishAliases.indexOf(topic);
++ if (autoAlias != -1) {
++ qCDebug(lcMqttConnectionVerbose) << "TopicAlias publish: Use auto alias:" << autoAlias;
++ packet->append(quint16(0));
++ publishProperties.setTopicAlias(quint16(autoAlias + 1));
++ } else {
++ autoAlias = m_publishAliases.indexOf(QMqttTopicName());
++ if (autoAlias != -1) {
++ qCDebug(lcMqttConnectionVerbose) << "TopicAlias publish: auto alias assignment:" << autoAlias;
++ m_publishAliases[autoAlias] = topic;
++ publishProperties.setTopicAlias(quint16(autoAlias) + 1);
++ } else
++ qCDebug(lcMqttConnectionVerbose) << "TopicAlias publish: alias storage full, using full topic";
++ packet->append(topic.name().toUtf8());
++ }
++ } else {
++ packet->append(topic.name().toUtf8());
++ }
++ } else { // ! MQTT_5_0
++ packet->append(topic.name().toUtf8());
++ }
++ quint16 identifier = 0;
++ if (qos > 0) {
++ identifier = unusedPacketIdentifier();
++ packet->append(identifier);
++ m_pendingMessages.insert(identifier, packet);
++ }
++
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
++ packet->appendRaw(writePublishProperties(publishProperties));
++
++ packet->appendRaw(message);
++
++ const bool written = writePacketToTransport(*packet.data());
++
++ if (!written && qos > 0)
++ m_pendingMessages.remove(identifier);
++ return written ? identifier : -1;
++ }
++
++ bool QMqttConnection::sendControlPublishAcknowledge(quint16 id)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << id;
++ QMqttControlPacket packet(QMqttControlPacket::PUBACK);
++ packet.append(id);
++ return writePacketToTransport(packet);
++ }
++
++ bool QMqttConnection::sendControlPublishRelease(quint16 id)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << id;
++ quint8 header = QMqttControlPacket::PUBREL;
++ header |= 0x02; // MQTT-3.6.1-1
++
++ QMqttControlPacket packet(header);
++ packet.append(id);
++ return writePacketToTransport(packet);
++ }
++
++ bool QMqttConnection::sendControlPublishReceive(quint16 id)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << id;
++ QMqttControlPacket packet(QMqttControlPacket::PUBREC);
++ packet.append(id);
++ return writePacketToTransport(packet);
++ }
++
++ bool QMqttConnection::sendControlPublishComp(quint16 id)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << id;
++ QMqttControlPacket packet(QMqttControlPacket::PUBCOMP);
++ packet.append(id);
++ return writePacketToTransport(packet);
++ }
++
++ QMqttSubscription *QMqttConnection::sendControlSubscribe(const QMqttTopicFilter &topic,
++ quint8 qos,
++ const QMqttSubscriptionProperties &properties)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << " Topic:" << topic << " qos:" << qos;
++
++ // Overflow protection
++ if (Q_UNLIKELY(!topic.isValid())) {
++ qCWarning(lcMqttConnection) << "Invalid subscription topic filter.";
++ return nullptr;
++ }
++
++ if (Q_UNLIKELY(qos > 2)) {
++ qCWarning(lcMqttConnection) << "Invalid subscription QoS.";
++ return nullptr;
++ }
++
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
++ const QString sharedSubscriptionName = topic.sharedSubscriptionName();
++ if (!sharedSubscriptionName.isEmpty()) {
++ const QMqttTopicFilter filter(topic.filter().section(QLatin1Char('/'), 2));
++ auto sub = findActiveSubscription(filter);
++ if (sub && (sub->sharedSubscriptionName() == sharedSubscriptionName))
++ return sub;
++ } else {
++ auto sub = findActiveSubscription(topic);
++ if (sub && !sub->isSharedSubscription())
++ return sub;
++ }
++ } else {
++ auto sub = findActiveSubscription(topic);
++ if (sub)
++ return sub;
++ }
++
++ // has to have 0010 as bits 3-0, maybe update SUBSCRIBE instead?
++ // MQTT-3.8.1-1
++ const quint8 header = QMqttControlPacket::SUBSCRIBE + 0x02;
++ QMqttControlPacket packet(header);
++
++ // Add Packet Identifier
++ const quint16 identifier = unusedPacketIdentifier();
++
++ packet.append(identifier);
++
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
++ packet.appendRaw(writeSubscriptionProperties(properties));
++
++ packet.append(topic.filter().toUtf8());
++ char options = char(qos);
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0 && properties.noLocal())
++ options |= 1 << 2;
++ packet.append(options);
++
++ auto result = new QMqttSubscription(this);
++ result->setTopic(topic);
++ result->setClient(m_clientPrivate->m_client);
++ result->setQos(qos);
++ result->setState(QMqttSubscription::SubscriptionPending);
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0 && !topic.sharedSubscriptionName().isEmpty()) {
++ result->setSharedSubscriptionName(topic.sharedSubscriptionName());
++ result->setSharedSubscription(true);
++ result->setTopic(topic.filter().section(QLatin1Char('/'), 2));
++ }
++
++ if (!writePacketToTransport(packet)) {
++ delete result;
++ return nullptr;
++ }
++
++ // SUBACK must contain identifier MQTT-3.8.4-2
++ m_pendingSubscriptionAck.insert(identifier, result);
++ m_activeSubscriptions.insert(result->topic(), result);
++ return result;
++ }
++
++ bool QMqttConnection::sendControlUnsubscribe(const QMqttTopicFilter &topic, const QMqttUnsubscriptionProperties &properties)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << " Topic:" << topic;
++
++ // MQTT-3.10.3-2
++ if (!topic.isValid())
++ return false;
++
++ auto sub = findActiveSubscription(topic);
++ if (!sub)
++ // Already unsubscribed
++ return false;
++
++ if (m_internalState != QMqttConnection::BrokerConnected) {
++ m_activeSubscriptions.remove(topic, sub);
++ return true;
++ }
++
++ // has to have 0010 as bits 3-0, maybe update UNSUBSCRIBE instead?
++ // MQTT-3.10.1-1
++ const quint8 header = QMqttControlPacket::UNSUBSCRIBE + 0x02;
++ QMqttControlPacket packet(header);
++
++ // Add Packet Identifier
++ const quint16 identifier = unusedPacketIdentifier();
++
++ packet.append(identifier);
++
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
++ packet.appendRaw(writeUnsubscriptionProperties(properties));
++ }
++
++ packet.append(topic.filter().toUtf8());
++ sub->setState(QMqttSubscription::UnsubscriptionPending);
++
++ if (!writePacketToTransport(packet))
++ return false;
++
++ // Do not remove from m_activeSubscriptions as there might be QoS1/2 messages to still
++ // be sent before UNSUBSCRIBE is acknowledged.
++ m_pendingUnsubscriptions.insert(identifier, sub);
++
++ return true;
++ }
++
++ bool QMqttConnection::sendControlPingRequest(bool isAuto)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO;
++
++ if (m_internalState != QMqttConnection::BrokerConnected)
++ return false;
++
++
++ if (!isAuto && m_clientPrivate->m_autoKeepAlive) {
++ qCDebug(lcMqttConnection) << "Requesting a manual ping while autoKeepAlive is enabled "
++ << "is not allowed.";
++ return false;
++ }
++
++ // 3.1.2.10 If a Client does not receive a PINGRESP packet within a reasonable amount of time
++ // after it has sent a PINGREQ, it SHOULD close the Network Connection to the Server
++ if (m_pingTimeout > 1) {
++ closeConnection(QMqttClient::ServerUnavailable);
++ return false;
++ }
++
++ const QMqttControlPacket packet(QMqttControlPacket::PINGREQ);
++ if (!writePacketToTransport(packet)) {
++ qCDebug(lcMqttConnection) << "Failed to write PINGREQ to transport.";
++ return false;
++ }
++ m_pingTimeout++;
++ return true;
++ }
++
++ bool QMqttConnection::sendControlDisconnect(std::optional<QMqtt::ReasonCode> reasonCode)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO;
++
++ m_pingTimer.stop();
++ m_pingTimeout = 0;
++
++ m_activeSubscriptions.clear();
++
++ m_receiveAliases.clear();
++ m_publishAliases.clear();
++
++ QMqttControlPacket packet(QMqttControlPacket::DISCONNECT);
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0 && reasonCode) {
++ // MQTT-3.14.2 The Disconnect Reason Code is a part of the variable header.
++ // Reason codes exist in MQTT 5.0 only, so this is never used for an older version.
++ packet.append(char(quint8(*reasonCode)));
++ packet.append(char(0)); // MQTT-3.14.2.2 property length, we send no properties
++ }
++ if (!writePacketToTransport(packet)) {
++ qCDebug(lcMqttConnection) << "Failed to write DISCONNECT to transport.";
++ return false;
++ }
++ if (m_internalState != ClientDestruction)
++ m_internalState = BrokerDisconnected;
++
++ if (m_transport->waitForBytesWritten(30000)) {
++ // MQTT-3.14.4-1 must disconnect
++ m_transport->close();
++ return true;
++ }
++ return false;
++ }
++
++ void QMqttConnection::setClientPrivate(QMqttClientPrivate *clientPrivate)
++ {
++ m_clientPrivate = clientPrivate;
++ }
++
++ quint16 QMqttConnection::unusedPacketIdentifier() const
++ {
++ // MQTT-2.3.1-1 Control Packets MUST contain a non-zero 16-bit Packet Identifier
++ static quint16 packetIdentifierCounter = 1;
++ const std::uint16_t u16max = std::numeric_limits<std::uint16_t>::max();
++
++ // MQTT-2.3.1-2 ...it MUST assign it a currently unused Packet Identifier
++ const quint16 lastIdentifier = packetIdentifierCounter;
++ do {
++ if (packetIdentifierCounter == u16max)
++ packetIdentifierCounter = 1;
++ else
++ packetIdentifierCounter++;
++
++ if (lastIdentifier == packetIdentifierCounter) {
++ qCDebug(lcMqttConnection) << "Could not generate unique packet identifier.";
++ break;
++ }
++ } while (m_pendingSubscriptionAck.contains(packetIdentifierCounter)
++ || m_pendingUnsubscriptions.contains(packetIdentifierCounter)
++ || m_pendingMessages.contains(packetIdentifierCounter)
++ || m_pendingReleaseMessages.contains(packetIdentifierCounter));
++ return packetIdentifierCounter;
++ }
++
++ void QMqttConnection::cleanSubscriptions()
++ {
++ for (auto item : std::as_const(m_pendingSubscriptionAck))
++ item->setState(QMqttSubscription::Unsubscribed);
++ m_pendingSubscriptionAck.clear();
++
++ for (auto item : std::as_const(m_pendingUnsubscriptions))
++ item->setState(QMqttSubscription::Unsubscribed);
++ m_pendingUnsubscriptions.clear();
++
++ for (auto item : std::as_const(m_activeSubscriptions))
++ item->setState(QMqttSubscription::Unsubscribed);
++ m_activeSubscriptions.clear();
++ }
++
++ void QMqttConnection::transportConnectionEstablished()
++ {
++ if (m_internalState != BrokerConnecting) {
++ qCWarning(lcMqttConnection) << "Connection established at an unexpected time";
++ return;
++ }
++
++ if (!sendControlConnect()) {
++ qCDebug(lcMqttConnection) << "Failed to write CONNECT to transport.";
++ // ### Who disconnects now? Connection or client?
++ m_clientPrivate->setStateAndError(QMqttClient::Disconnected, QMqttClient::TransportInvalid);
++ }
++ }
++
++ void QMqttConnection::transportConnectionClosed()
++ {
++ m_readBuffer.clear();
++ m_readPosition = 0;
++ m_pingTimer.stop();
++ m_pingTimeout = 0;
++ if (m_internalState == ClientDestruction)
++ return;
++ if (m_internalState == BrokerDisconnected) // We manually disconnected
++ m_clientPrivate->setStateAndError(QMqttClient::Disconnected, QMqttClient::NoError);
++ else
++ m_clientPrivate->setStateAndError(QMqttClient::Disconnected, QMqttClient::TransportInvalid);
++ }
++
++ void QMqttConnection::transportReadyRead()
++ {
++ qCDebug(lcMqttConnectionVerbose) << Q_FUNC_INFO;
++ m_readBuffer.append(m_transport->readAll());
++ processData();
++ }
++
++ void QMqttConnection::transportError(QAbstractSocket::SocketError e)
++ {
++ qCDebug(lcMqttConnection) << Q_FUNC_INFO << e;
++ closeConnection(QMqttClient::TransportInvalid);
++ }
++
++ bool QMqttConnection::readBuffer(char *data, quint64 size)
++ {
++ if (Q_UNLIKELY(quint64(m_readBuffer.size() - m_readPosition) < size)) {
++ qCDebug(lcMqttConnection) << "Reaching out of buffer, protocol violation";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ memcpy(data, m_readBuffer.constData() + m_readPosition, size);
++ m_readPosition += size;
++ return true;
++ }
++
++ qint32 QMqttConnection::readVariableByteInteger(qint64 *dataSize)
++ {
++ quint32 multiplier = 1;
++ qint32 msgLength = 0;
++ quint8 b = 0;
++ quint8 iteration = 0;
++ do {
++ iteration++;
++ if (iteration > 4) {
++ qCDebug(lcMqttConnection) << "Overflow trying to read variable integer.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return -1;
++ }
++ b = readBufferTyped<quint8>(dataSize);
++ msgLength += (b & 127) * multiplier;
++ multiplier *= 128;
++ } while ((b & 128) != 0);
++ return msgLength;
++ }
++
++ void QMqttConnection::closeConnection(QMqttClient::ClientError error)
++ {
++ m_readBuffer.clear();
++ m_readPosition = 0;
++ m_pingTimer.stop();
++ m_pingTimeout = 0;
++ m_activeSubscriptions.clear();
++ m_internalState = BrokerDisconnected;
++ m_transport->disconnect();
++ m_transport->close();
++ m_clientPrivate->setStateAndError(QMqttClient::Disconnected, error);
++ }
++
++ QByteArray QMqttConnection::readBuffer(quint64 size)
++ {
++ if (Q_UNLIKELY(quint64(m_readBuffer.size() - m_readPosition) < size)) {
++ qCDebug(lcMqttConnection) << "Reaching out of buffer, protocol violation";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return QByteArray();
++ }
++ QByteArray res(m_readBuffer.constData() + m_readPosition, int(size));
++ m_readPosition += size;
++ return res;
++ }
++
++ void QMqttConnection::readAuthProperties(QMqttAuthenticationProperties &properties)
++ {
++ qint64 propertyLength = readVariableByteInteger(&m_missingData);
++ if (propertyLength < 0) // Connection closed inside readVariableByteInteger
++ return;
++ m_missingData -= propertyLength;
++
++ QMqttUserProperties userProperties;
++ while (propertyLength > 0) {
++ quint8 propertyId = readBufferTyped<quint8>(&propertyLength);
++
++ switch (propertyId) {
++ case 0x15: { //3.15.2.2.2 Authentication Method
++ const QString method = readBufferTyped<QString>(&propertyLength);
++ properties.setAuthenticationMethod(method);
++ break;
++ }
++ case 0x16: { // 3.15.2.2.3 Authentication Data
++ const QByteArray data = readBufferTyped<QByteArray>(&propertyLength);
++ properties.setAuthenticationData(data);
++ break;
++ }
++ case 0x1F: { // 3.15.2.2.4 Reason String
++ const QString reasonString = readBufferTyped<QString>(&propertyLength);
++ properties.setReason(reasonString);
++ break;
++ }
++ case 0x26: { // 3.15.2.2.5 User property
++ const QString propertyName = readBufferTyped<QString>(&propertyLength);
++ const QString propertyValue = readBufferTyped<QString>(&propertyLength);
++
++ userProperties.append(QMqttStringPair(propertyName, propertyValue));
++ break;
++ }
++ default:
++ qCDebug(lcMqttConnection) << "Unknown property id in AUTH:" << propertyId;
++ break;
++ }
++ }
++ if (!userProperties.isEmpty())
++ properties.setUserProperties(userProperties);
++ }
++
++ void QMqttConnection::readConnackProperties(QMqttServerConnectionProperties &properties)
++ {
++ qint64 propertyLength = readVariableByteInteger(&m_missingData);
++ if (propertyLength < 0) // Connection closed inside readVariableByteInteger
++ return;
++ m_missingData -= propertyLength;
++
++ properties.serverData->valid = true;
++
++ while (propertyLength > 0) {
++ quint8 propertyId = readBufferTyped<quint8>(&propertyLength);
++ switch (propertyId) {
++ case 0x11: { // 3.2.2.3.2 Session Expiry Interval
++ const quint32 expiryInterval = readBufferTyped<quint32>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::SessionExpiryInterval;
++ properties.setSessionExpiryInterval(expiryInterval);
++ break;
++ }
++ case 0x21: { // 3.2.2.3.3 Receive Maximum
++ const quint16 receiveMaximum = readBufferTyped<quint16>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::MaximumReceive;
++ properties.setMaximumReceive(receiveMaximum);
++ break;
++ }
++ case 0x24: { // 3.2.2.3.4 Maximum QoS Level
++ const quint8 maxQoS = readBufferTyped<quint8>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::MaximumQoS;
++ properties.serverData->maximumQoS = maxQoS;
++ break;
++ }
++ case 0x25: { // 3.2.2.3.5 Retain available
++ const quint8 retainAvailable = readBufferTyped<quint8>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::RetainAvailable;
++ properties.serverData->retainAvailable = retainAvailable == 1;
++ break;
++ }
++ case 0x27: { // 3.2.2.3.6 Maximum packet size
++ const quint32 maxPacketSize = readBufferTyped<quint32>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::MaximumPacketSize;
++ properties.setMaximumPacketSize(maxPacketSize);
++ break;
++ }
++ case 0x12: { // 3.2.2.3.7 Assigned clientId
++ const QString assignedClientId = readBufferTyped<QString>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::AssignedClientId;
++ m_clientPrivate->setClientId(assignedClientId);
++ break;
++ }
++ case 0x22: { // 3.2.2.3.8 Topic Alias Maximum
++ const quint16 topicAliasMaximum = readBufferTyped<quint16>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::MaximumTopicAlias;
++ properties.setMaximumTopicAlias(topicAliasMaximum);
++ break;
++ }
++ case 0x1F: { // 3.2.2.3.9 Reason String
++ const QString reasonString = readBufferTyped<QString>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::ReasonString;
++ properties.serverData->reasonString = reasonString;
++ break;
++ }
++ case 0x26: { // 3.2.2.3.10 User property
++ const QString propertyName = readBufferTyped<QString>(&propertyLength);
++ const QString propertyValue = readBufferTyped<QString>(&propertyLength);
++
++ properties.serverData->details |= QMqttServerConnectionProperties::UserProperty;
++ properties.data->userProperties.append(QMqttStringPair(propertyName, propertyValue));
++ break;
++ }
++ case 0x28: { // 3.2.2.3.11 Wildcard subscriptions available
++ const quint8 available = readBufferTyped<quint8>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::WildCardSupported;
++ properties.serverData->wildcardSupported = available == 1;
++ break;
++ }
++ case 0x29: { // 3.2.2.3.12 Subscription identifiers available
++ const quint8 available = readBufferTyped<quint8>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::SubscriptionIdentifierSupport;
++ properties.serverData->subscriptionIdentifierSupported = available == 1;
++ break;
++ }
++ case 0x2A: { // 3.2.2.3.13 Shared subscriptions available
++ const quint8 available = readBufferTyped<quint8>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::SharedSubscriptionSupport;
++ properties.serverData->sharedSubscriptionSupported = available == 1;
++ break;
++ }
++ case 0x13: { // 3.2.2.3.14 Server Keep Alive
++ const quint16 serverKeepAlive = readBufferTyped<quint16>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::ServerKeepAlive;
++ m_clientPrivate->m_client->setKeepAlive(serverKeepAlive);
++ break;
++ }
++ case 0x1A: { // 3.2.2.3.15 Response information
++ const QString responseInfo = readBufferTyped<QString>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::ResponseInformation;
++ properties.serverData->responseInformation = responseInfo;
++ break;
++ }
++ case 0x1C: { // 3.2.2.3.16 Server reference
++ const QString serverReference = readBufferTyped<QString>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::ServerReference;
++ properties.serverData->serverReference = serverReference;
++ break;
++ }
++ case 0x15: { // 3.2.2.3.17 Authentication method
++ const QString method = readBufferTyped<QString>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::AuthenticationMethod;
++ properties.data->authenticationMethod = method;
++ break;
++ }
++ case 0x16: { // 3.2.2.3.18 Authentication data
++ const QByteArray data = readBufferTyped<QByteArray>(&propertyLength);
++ properties.serverData->details |= QMqttServerConnectionProperties::AuthenticationData;
++ properties.data->authenticationData = data;
++ break;
++ }
++ default:
++ qCDebug(lcMqttConnection) << "Unknown property id in CONNACK:" << int(propertyId);
++ break;
++ }
++ }
++ }
++
++ void QMqttConnection::readMessageStatusProperties(QMqttMessageStatusProperties &properties)
++ {
++ qint64 propertyLength = readVariableByteInteger(&m_missingData);
++ if (propertyLength < 0) // Connection closed inside readVariableByteInteger
++ return;
++ m_missingData -= propertyLength;
++
++ while (propertyLength > 0) {
++ const quint8 propertyId = readBufferTyped<quint8>(&propertyLength);
++ switch (propertyId) {
++ case 0x1f: { // 3.4.2.2.2 Reason String
++ const QString content = readBufferTyped<QString>(&propertyLength);
++ properties.data->reasonString = content;
++ break;
++ }
++ case 0x26: { // 3.4.2.2.3 User Properites
++ const QString propertyName = readBufferTyped<QString>(&propertyLength);
++ const QString propertyValue = readBufferTyped<QString>(&propertyLength);
++ properties.data->userProperties.append(QMqttStringPair(propertyName, propertyValue));
++ break;
++ }
++ default:
++ qCDebug(lcMqttConnection) << "Unknown subscription property received.";
++ break;
++ }
++ }
++ }
++
++ void QMqttConnection::readPublishProperties(QMqttPublishProperties &properties)
++ {
++ qint64 propertyLength = readVariableByteInteger(&m_missingData);
++ if (propertyLength < 0) // Connection closed inside readVariableByteInteger
++ return;
++ m_missingData -= propertyLength;
++
++ QMqttUserProperties userProperties;
++ QList<quint32> subscriptionIds;
++
++ while (propertyLength > 0) {
++ const quint8 propertyId = readBufferTyped<quint8>(&propertyLength);
++ switch (propertyId) {
++ case 0x01: { // 3.3.2.3.2 Payload Format Indicator
++ const quint8 format = readBufferTyped<quint8>(&propertyLength);
++ if (format == 1)
++ properties.setPayloadFormatIndicator(QMqtt::PayloadFormatIndicator::UTF8Encoded);
++ break;
++ }
++ case 0x02: { // 3.3.2.3.3 Message Expiry Interval
++ const quint32 interval = readBufferTyped<quint32>(&propertyLength);
++ properties.setMessageExpiryInterval(interval);
++ break;
++ }
++ case 0x23: { // 3.3.2.3.4 Topic alias
++ const quint16 alias = readBufferTyped<quint16>(&propertyLength);
++ properties.setTopicAlias(alias);
++ break;
++ }
++ case 0x08: { // 3.3.2.3.5 Response Topic
++ const QString responseTopic = readBufferTyped<QString>(&propertyLength);
++ properties.setResponseTopic(responseTopic);
++ break;
++ }
++ case 0x09: { // 3.3.2.3.6 Correlation Data
++ const QByteArray data = readBufferTyped<QByteArray>(&propertyLength);
++ properties.setCorrelationData(data);
++ break;
++ }
++ case 0x26: { // 3.3.2.3.7 User property
++ const QString propertyName = readBufferTyped<QString>(&propertyLength);
++ const QString propertyValue = readBufferTyped<QString>(&propertyLength);
++ userProperties.append(QMqttStringPair(propertyName, propertyValue));
++ break;
++ }
++ case 0x0b: { // 3.3.2.3.8 Subscription Identifier
++ qint32 id = readVariableByteInteger(&propertyLength);
++ if (id < 0)
++ return; // readVariableByteInteger closes connection
++ subscriptionIds.append(quint32(id));
++ break;
++ }
++ case 0x03: { // 3.3.2.3.9 Content Type
++ const QString content = readBufferTyped<QString>(&propertyLength);
++ properties.setContentType(content);
++ break;
++ }
++ default:
++ qCDebug(lcMqttConnection) << "Unknown publish property received.";
++ break;
++ }
++ }
++ if (!userProperties.isEmpty())
++ properties.setUserProperties(userProperties);
++
++ if (!subscriptionIds.isEmpty())
++ properties.setSubscriptionIdentifiers(subscriptionIds);
++ }
++
++ void QMqttConnection::readSubscriptionProperties(QMqttSubscription *sub)
++ {
++ qint64 propertyLength = readVariableByteInteger(&m_missingData);
++ if (propertyLength < 0) // Connection closed inside readVariableByteInteger
++ return;
++ m_missingData -= propertyLength;
++
++ while (propertyLength > 0) {
++ const quint8 propertyId = readBufferTyped<quint8>(&propertyLength);
++ switch (propertyId) {
++ case 0x1f: { // 3.9.2.1.2 Reason String
++ const QString content = readBufferTyped<QString>(&propertyLength);
++ sub->d_func()->m_reasonString = content;
++ break;
++ }
++ case 0x26: { // 3.9.2.1.3
++ const QString propertyName = readBufferTyped<QString>(&propertyLength);
++ const QString propertyValue = readBufferTyped<QString>(&propertyLength);
++
++ sub->d_func()->m_userProperties.append(QMqttStringPair(propertyName, propertyValue));
++ break;
++ }
++ default:
++ qCDebug(lcMqttConnection) << "Unknown subscription property received.";
++ break;
++ }
++ }
++ }
++
++ QByteArray QMqttConnection::writeConnectProperties()
++ {
++ QMqttControlPacket properties;
++
++ // According to MQTT5 3.1.2.11 default values do not need to be included in the
++ // connect statement.
++
++ // 3.1.2.11.2
++ if (m_clientPrivate->m_connectionProperties.sessionExpiryInterval() != 0) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: specify sessionExpiryInterval";
++ properties.append(char(0x11));
++ properties.append(m_clientPrivate->m_connectionProperties.sessionExpiryInterval());
++ }
++
++ // 3.1.2.11.3
++ if (m_clientPrivate->m_connectionProperties.maximumReceive() != 65535) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: specify maximumReceive";
++ properties.append(char(0x21));
++ properties.append(m_clientPrivate->m_connectionProperties.maximumReceive());
++ }
++
++ // 3.1.2.11.4
++ if (m_clientPrivate->m_connectionProperties.maximumPacketSize() != std::numeric_limits<quint32>::max()) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: specify maximumPacketSize";
++ properties.append(char(0x27));
++ properties.append(m_clientPrivate->m_connectionProperties.maximumPacketSize());
++ }
++
++ // 3.1.2.11.5
++ if (m_clientPrivate->m_connectionProperties.maximumTopicAlias() != 0) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: specify maximumTopicAlias";
++ properties.append(char(0x22));
++ properties.append(m_clientPrivate->m_connectionProperties.maximumTopicAlias());
++ }
++
++ // 3.1.2.11.6
++ if (m_clientPrivate->m_connectionProperties.requestResponseInformation()) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: specify requestResponseInformation";
++ properties.append(char(0x19));
++ properties.append(char(1));
++ }
++
++ // 3.1.2.11.7
++ if (!m_clientPrivate->m_connectionProperties.requestProblemInformation()) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: specify requestProblemInformation";
++ properties.append(char(0x17));
++ properties.append(char(0));
++ }
++
++ // 3.1.2.11.8 Add User properties
++ auto userProperties = m_clientPrivate->m_connectionProperties.userProperties();
++ if (!userProperties.isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: specify user properties";
++ for (const auto &prop : userProperties) {
++ properties.append(char(0x26));
++ properties.append(prop.name().toUtf8());
++ properties.append(prop.value().toUtf8());
++ }
++ }
++
++ // 3.1.2.11.9 Add Authentication
++ const QString authenticationMethod = m_clientPrivate->m_connectionProperties.authenticationMethod();
++ if (!authenticationMethod.isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: specify AuthenticationMethod:";
++ qCDebug(lcMqttConnectionVerbose) << " " << authenticationMethod;
++ properties.append(char(0x15));
++ properties.append(authenticationMethod.toUtf8());
++ // 3.1.2.11.10
++ const QByteArray authenticationData = m_clientPrivate->m_connectionProperties.authenticationData();
++ if (!authenticationData.isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Connection Properties: Authentication Data:";
++ qCDebug(lcMqttConnectionVerbose) << " " << authenticationData;
++ properties.append(char(0x16));
++ properties.append(authenticationData);
++ }
++ }
++
++ return properties.serializePayload();
++ }
++
++ QByteArray QMqttConnection::writeLastWillProperties() const
++ {
++ QMqttControlPacket properties;
++ const QMqttLastWillProperties &lastWillProperties = m_clientPrivate->m_lastWillProperties;
++ // Will Delay interval 3.1.3.2.2
++ if (lastWillProperties.willDelayInterval() > 0) {
++ const quint32 delay = lastWillProperties.willDelayInterval();
++ qCDebug(lcMqttConnectionVerbose) << "Last Will Properties: specify will delay interval:"
++ << delay;
++ properties.append(char(0x18));
++ properties.append(delay);
++ }
++
++ // Payload Format Indicator 3.1.3.2.3
++ if (lastWillProperties.payloadFormatIndicator() != QMqtt::PayloadFormatIndicator::Unspecified) {
++ qCDebug(lcMqttConnectionVerbose) << "Last Will Properties: payload format indicator specified";
++ properties.append(char(0x01));
++ properties.append(char(0x01)); // UTF8
++ }
++
++ // Message Expiry Interval 3.1.3.2.4
++ if (lastWillProperties.messageExpiryInterval() > 0) {
++ const quint32 interval = lastWillProperties.messageExpiryInterval();
++ qCDebug(lcMqttConnectionVerbose) << "Last Will Properties: Message Expiry interval:"
++ << interval;
++ properties.append(char(0x02));
++ properties.append(interval);
++ }
++
++ // Content Type 3.1.3.2.5
++ if (!lastWillProperties.contentType().isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Last Will Properties: Content Type:"
++ << lastWillProperties.contentType();
++ properties.append(char(0x03));
++ properties.append(lastWillProperties.contentType().toUtf8());
++ }
++
++ // Response Topic 3.1.3.2.6
++ if (!lastWillProperties.responseTopic().isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Last Will Properties: Response Topic:"
++ << lastWillProperties.responseTopic();
++ properties.append(char(0x08));
++ properties.append(lastWillProperties.responseTopic().toUtf8());
++ }
++
++ // Correlation Data 3.1.3.2.7
++ if (!lastWillProperties.correlationData().isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Last Will Properties: Correlation Data:"
++ << lastWillProperties.correlationData();
++ properties.append(char(0x09));
++ properties.append(lastWillProperties.correlationData());
++ }
++
++ // User Properties 3.1.3.2.8
++ if (!lastWillProperties.userProperties().isEmpty()) {
++ auto userProperties = lastWillProperties.userProperties();
++ qCDebug(lcMqttConnectionVerbose) << "Last Will Properties: specify user properties";
++ for (const auto &prop : userProperties) {
++ properties.append(char(0x26));
++ properties.append(prop.name().toUtf8());
++ properties.append(prop.value().toUtf8());
++ }
++ }
++
++ return properties.serializePayload();
++ }
++
++ QByteArray QMqttConnection::writePublishProperties(const QMqttPublishProperties &properties)
++ {
++ QMqttControlPacket packet;
++
++ // 3.3.2.3.2 Payload Indicator
++ if (properties.availableProperties() & QMqttPublishProperties::PayloadFormatIndicator &&
++ properties.payloadFormatIndicator() != QMqtt::PayloadFormatIndicator::Unspecified) {
++ qCDebug(lcMqttConnectionVerbose) << "Publish Properties: Payload Indicator:"
++ << (properties.payloadFormatIndicator() == QMqtt::PayloadFormatIndicator::UTF8Encoded ? 1 : 0);
++ packet.append(char(0x01));
++ switch (properties.payloadFormatIndicator()) {
++ case QMqtt::PayloadFormatIndicator::UTF8Encoded:
++ packet.append(char(0x01));
++ break;
++ default:
++ qCDebug(lcMqttConnection) << "Unknown payload indicator.";
++ break;
++ }
++ }
++
++ // 3.3.2.3.3 Message Expiry
++ if (properties.availableProperties() & QMqttPublishProperties::MessageExpiryInterval &&
++ properties.messageExpiryInterval() > 0) {
++ qCDebug(lcMqttConnectionVerbose) << "Publish Properties: Message Expiry :"
++ << properties.messageExpiryInterval();
++ packet.append(char(0x02));
++ packet.append(properties.messageExpiryInterval());
++ }
++
++ // 3.3.2.3.4 Topic alias
++ if (properties.availableProperties() & QMqttPublishProperties::TopicAlias &&
++ properties.topicAlias() > 0) {
++ qCDebug(lcMqttConnectionVerbose) << "Publish Properties: Topic Alias :"
++ << properties.topicAlias();
++ if (m_clientPrivate->m_serverConnectionProperties.availableProperties() & QMqttServerConnectionProperties::MaximumTopicAlias
++ && properties.topicAlias() > m_clientPrivate->m_serverConnectionProperties.maximumTopicAlias()) {
++ qCDebug(lcMqttConnection) << "Invalid topic alias specified: " << properties.topicAlias()
++ << " Maximum by server is:"
++ << m_clientPrivate->m_serverConnectionProperties.maximumTopicAlias();
++
++ } else {
++ packet.append(char(0x23));
++ packet.append(properties.topicAlias());
++ }
++ }
++
++ // 3.3.2.3.5 Response Topic
++ if (properties.availableProperties() & QMqttPublishProperties::ResponseTopic &&
++ !properties.responseTopic().isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Publish Properties: Response Topic :"
++ << properties.responseTopic();
++ packet.append(char(0x08));
++ packet.append(properties.responseTopic().toUtf8());
++ }
++
++ // 3.3.2.3.6 Correlation Data
++ if (properties.availableProperties() & QMqttPublishProperties::CorrelationData &&
++ !properties.correlationData().isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Publish Properties: Correlation Data :"
++ << properties.correlationData();
++ packet.append(char(0x09));
++ packet.append(properties.correlationData());
++ }
++
++ // 3.3.2.3.7 User Property
++ if (properties.availableProperties() & QMqttPublishProperties::UserProperty) {
++ auto userProperties = properties.userProperties();
++ if (!userProperties.isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Publish Properties: specify user properties";
++ for (const auto &prop : userProperties) {
++ packet.append(char(0x26));
++ packet.append(prop.name().toUtf8());
++ packet.append(prop.value().toUtf8());
++ }
++ }
++ }
++
++ // 3.3.2.3.8 Subscription Identifier
++ if (properties.availableProperties() & QMqttPublishProperties::SubscriptionIdentifier) {
++ const auto subscriptionIds = properties.subscriptionIdentifiers();
++ for (auto id : subscriptionIds) {
++ qCDebug(lcMqttConnectionVerbose) << "Publish Properties: Subscription ID:" << id;
++ packet.append(char(0x0b));
++ packet.appendRawVariableInteger(id);
++ }
++ }
++
++ // 3.3.2.3.9 Content Type
++ if (properties.availableProperties() & QMqttPublishProperties::ContentType &&
++ !properties.contentType().isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Publish Properties: Content Type :"
++ << properties.contentType();
++ packet.append(char(0x03));
++ packet.append(properties.contentType().toUtf8());
++ }
++
++ return packet.serializePayload();
++ }
++
++ QByteArray QMqttConnection::writeSubscriptionProperties(const QMqttSubscriptionProperties &properties)
++ {
++ QMqttControlPacket packet;
++
++ // 3.8.2.1.2 Subscription Identifier
++ if (properties.subscriptionIdentifier() > 0) {
++ qCDebug(lcMqttConnectionVerbose) << "Subscription Properties: Subscription Identifier";
++ packet.append(char(0x0b));
++ packet.appendRawVariableInteger(properties.subscriptionIdentifier());
++ }
++
++ // 3.8.2.1.3 User Property
++ auto userProperties = properties.userProperties();
++ if (!userProperties.isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Subscription Properties: specify user properties";
++ for (const auto &prop : userProperties) {
++ packet.append(char(0x26));
++ packet.append(prop.name().toUtf8());
++ packet.append(prop.value().toUtf8());
++ }
++ }
++
++ return packet.serializePayload();
++ }
++
++ QByteArray QMqttConnection::writeUnsubscriptionProperties(const QMqttUnsubscriptionProperties &properties)
++ {
++ QMqttControlPacket packet;
++
++ // 3.10.2.1.2
++ auto userProperties = properties.userProperties();
++ if (!userProperties.isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Unsubscription Properties: specify user properties";
++ for (const auto &prop : userProperties) {
++ packet.append(char(0x26));
++ packet.append(prop.name().toUtf8());
++ packet.append(prop.value().toUtf8());
++ }
++ }
++
++ return packet.serializePayload();
++ }
++
++ QByteArray QMqttConnection::writeAuthenticationProperties(const QMqttAuthenticationProperties &properties)
++ {
++ QMqttControlPacket packet;
++
++ // 3.15.2.2.2
++ if (!properties.authenticationMethod().isEmpty()) {
++ packet.append(char(0x15));
++ packet.append(properties.authenticationMethod().toUtf8());
++ }
++ // 3.15.2.2.3
++ if (!properties.authenticationData().isEmpty()) {
++ packet.append(char(0x16));
++ packet.append(properties.authenticationData());
++ }
++
++ // 3.15.2.2.4
++ if (!properties.reason().isEmpty()) {
++ packet.append(char(0x1F));
++ packet.append(properties.reason().toUtf8());
++ }
++
++ // 3.15.2.2.5
++ auto userProperties = properties.userProperties();
++ if (!userProperties.isEmpty()) {
++ qCDebug(lcMqttConnectionVerbose) << "Unsubscription Properties: specify user properties";
++ for (const auto &prop : userProperties) {
++ packet.append(char(0x26));
++ packet.append(prop.name().toUtf8());
++ packet.append(prop.value().toUtf8());
++ }
++ }
++
++ return packet.serializePayload();
++ }
++
++ void QMqttConnection::finalize_auth()
++ {
++ qCDebug(lcMqttConnectionVerbose) << "Finalize AUTH";
++
++ quint8 authReason = 0;
++ QMqttAuthenticationProperties authProperties;
++ // 3.15.2.1 - The Reason Code and Property Length can be omitted if the Reason Code
++ // is 0x00 (Success) and there are no Properties. In this case the AUTH has a
++ // Remaining Length of 0.
++ if (m_missingData > 0) {
++ authReason = readBufferTyped<quint8>(&m_missingData);
++ readAuthProperties(authProperties);
++ }
++
++ // 3.15.2.1
++ switch (QMqtt::ReasonCode(authReason)) {
++ case QMqtt::ReasonCode::Success:
++ emit m_clientPrivate->m_client->authenticationFinished(authProperties);
++ break;
++ case QMqtt::ReasonCode::ContinueAuthentication:
++ case QMqtt::ReasonCode::ReAuthenticate:
++ emit m_clientPrivate->m_client->authenticationRequested(authProperties);
++ break;
++ default:
++ qCDebug(lcMqttConnection) << "Received illegal AUTH reason code:" << authReason;
++ closeConnection(QMqttClient::ProtocolViolation);
++ break;
++ }
++ }
++
++ void QMqttConnection::finalize_connack()
++ {
++ qCDebug(lcMqttConnectionVerbose) << "Finalize CONNACK";
++
++ const quint8 ackFlags = readBufferTyped<quint8>(&m_missingData);
++
++ if (ackFlags > 1) { // MQTT-3.2.2.1
++ qCDebug(lcMqttConnection) << "Unexpected CONNACK Flags specified:" << QString::number(ackFlags);
++ readBuffer(quint64(m_missingData));
++ m_missingData = 0;
++ closeConnection(QMqttClient::ProtocolViolation);
++ return;
++ }
++ bool sessionPresent = ackFlags == 1;
++
++ // MQTT-3.2.2-1 & MQTT-3.2.2-2
++ if (sessionPresent) {
++ emit m_clientPrivate->m_client->brokerSessionRestored();
++ if (m_clientPrivate->m_cleanSession)
++ qCDebug(lcMqttConnection) << "Connected with a clean session, ack contains session present.";
++ } else {
++ // MQTT-4.1.0.-1 MQTT-4.1.0-2 Session not stored on broker side
++ // regardless whether cleanSession is false
++ cleanSubscriptions();
++ }
++
++ quint8 connectResultValue = readBufferTyped<quint8>(&m_missingData);
++ QMqttServerConnectionProperties serverProp;
++ serverProp.serverData->reasonCode = QMqtt::ReasonCode(connectResultValue);
++ m_clientPrivate->m_serverConnectionProperties = serverProp;
++ if (connectResultValue != 0 && m_clientPrivate->m_protocolVersion != QMqttClient::MQTT_5_0) {
++ qCDebug(lcMqttConnection) << "Connection has been rejected.";
++ closeConnection(static_cast<QMqttClient::ClientError>(connectResultValue));
++ return;
++ }
++
++ // MQTT 5.0 has variable part != 2 in the header
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
++ readConnackProperties(m_clientPrivate->m_serverConnectionProperties);
++ // readConnackProperties() might have closed the connection due to a malformed
++ // property length. Do not overwrite the state of a connection which is gone.
++ if (m_internalState == BrokerDisconnected)
++ return;
++ // MQTT 5.0, 3.1.2.11.5 and 3.2.2.3.8: the two Topic Alias Maximum values are
++ // directional. The value this client sent in CONNECT bounds the aliases the
++ // server may use towards us (m_receiveAliases); the value the server sent in
++ // CONNACK bounds the aliases we may use towards it (m_publishAliases).
++ // Mappings live only for the lifetime of a network connection, so discard any
++ // left over from a previous one instead of resizing in place.
++ m_receiveAliases.clear();
++ m_receiveAliases.resize(m_clientPrivate->m_connectionProperties.maximumTopicAlias());
++ m_publishAliases.clear();
++ m_publishAliases.resize(m_clientPrivate->m_serverConnectionProperties.maximumTopicAlias());
++
++ // 3.2.2.2
++ switch (QMqtt::ReasonCode(connectResultValue)) {
++ case QMqtt::ReasonCode::Success:
++ break;
++ case QMqtt::ReasonCode::MalformedPacket:
++ case QMqtt::ReasonCode::ProtocolError:
++ closeConnection(QMqttClient::ProtocolViolation);
++ return;
++ case QMqtt::ReasonCode::UnsupportedProtocolVersion:
++ closeConnection(QMqttClient::InvalidProtocolVersion);
++ return;
++ case QMqtt::ReasonCode::InvalidClientId:
++ closeConnection(QMqttClient::IdRejected);
++ return;
++ case QMqtt::ReasonCode::ServerNotAvailable:
++ case QMqtt::ReasonCode::ServerBusy:
++ case QMqtt::ReasonCode::UseAnotherServer:
++ case QMqtt::ReasonCode::ServerMoved:
++ closeConnection(QMqttClient::ServerUnavailable);
++ return;
++ case QMqtt::ReasonCode::InvalidUserNameOrPassword:
++ closeConnection(QMqttClient::BadUsernameOrPassword);
++ return;
++ case QMqtt::ReasonCode::NotAuthorized:
++ closeConnection(QMqttClient::NotAuthorized);
++ return;
++ case QMqtt::ReasonCode::UnspecifiedError:
++ closeConnection(QMqttClient::UnknownError);
++ return;
++ case QMqtt::ReasonCode::ImplementationSpecificError:
++ case QMqtt::ReasonCode::ClientBanned:
++ case QMqtt::ReasonCode::InvalidAuthenticationMethod:
++ case QMqtt::ReasonCode::InvalidTopicName:
++ case QMqtt::ReasonCode::PacketTooLarge:
++ case QMqtt::ReasonCode::QuotaExceeded:
++ case QMqtt::ReasonCode::InvalidPayloadFormat:
++ case QMqtt::ReasonCode::RetainNotSupported:
++ case QMqtt::ReasonCode::QoSNotSupported:
++ case QMqtt::ReasonCode::ExceededConnectionRate:
++ closeConnection(QMqttClient::Mqtt5SpecificError);
++ return;
++ default:
++ qCDebug(lcMqttConnection) << "Received illegal CONNACK reason code:" << connectResultValue;
++ closeConnection(QMqttClient::ProtocolViolation);
++ return;
++ }
++ }
++
++ m_internalState = BrokerConnected;
++ m_clientPrivate->setStateAndError(QMqttClient::Connected);
++
++ if (m_clientPrivate->m_autoKeepAlive)
++ m_pingTimer.start(m_clientPrivate->m_keepAlive * 1000, this);
++ }
++
++ void QMqttConnection::finalize_suback()
++ {
++ const quint16 id = readBufferTyped<quint16>(&m_missingData);
++
++ auto sub = m_pendingSubscriptionAck.take(id);
++ if (Q_UNLIKELY(sub == nullptr)) {
++ qCDebug(lcMqttConnection) << "Received SUBACK for unknown subscription request.";
++ return;
++ }
++
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
++ readSubscriptionProperties(sub);
++
++ // 3.9.3 - The Payload contains a list of Reason Codes. Each Reason Code corresponds to a Topic Filter in the SUBSCRIBE packet being acknowledged.
++ // Whereas 3.8.3 states "The Payload MUST contain at least one Topic Filter and Subscription Options pair. A SUBSCRIBE packet with no Payload is a Protocol Error."
++ do {
++ quint8 reason = readBufferTyped<quint8>(&m_missingData);
++
++ sub->d_func()->m_reasonCode = QMqtt::ReasonCode(reason);
++
++ // 3.9.3
++ switch (QMqtt::ReasonCode(reason)) {
++ case QMqtt::ReasonCode::SubscriptionQoSLevel0:
++ case QMqtt::ReasonCode::SubscriptionQoSLevel1:
++ case QMqtt::ReasonCode::SubscriptionQoSLevel2:
++ qCDebug(lcMqttConnectionVerbose) << "Finalize SUBACK: id:" << id << "qos:" << reason;
++ // The broker might have a different support level for QoS than what
++ // the client requested
++ if (reason != sub->qos()) {
++ sub->setQos(reason);
++ emit sub->qosChanged(reason);
++ }
++ sub->setState(QMqttSubscription::Subscribed);
++ break;
++ case QMqtt::ReasonCode::UnspecifiedError:
++ qCWarning(lcMqttConnection) << "Subscription for id " << id << " failed. Reason Code:" << reason;
++ sub->setState(QMqttSubscription::Error);
++ break;
++ case QMqtt::ReasonCode::ImplementationSpecificError:
++ case QMqtt::ReasonCode::NotAuthorized:
++ case QMqtt::ReasonCode::InvalidTopicFilter:
++ case QMqtt::ReasonCode::MessageIdInUse:
++ case QMqtt::ReasonCode::QuotaExceeded:
++ case QMqtt::ReasonCode::SharedSubscriptionsNotSupported:
++ case QMqtt::ReasonCode::SubscriptionIdsNotSupported:
++ case QMqtt::ReasonCode::WildCardSubscriptionsNotSupported:
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
++ qCWarning(lcMqttConnection) << "Subscription for id " << id << " failed. Reason Code:" << reason;
++ sub->setState(QMqttSubscription::Error);
++ break;
++ }
++ Q_FALLTHROUGH();
++ default:
++ qCWarning(lcMqttConnection) << "Received illegal SUBACK reason code:" << reason;
++ closeConnection(QMqttClient::ProtocolViolation);
++ break;
++ }
++ } while (m_missingData > 0);
++ }
++
++ void QMqttConnection::finalize_unsuback()
++ {
++ const quint16 id = readBufferTyped<quint16>(&m_missingData);
++ qCDebug(lcMqttConnectionVerbose) << "Finalize UNSUBACK: " << id;
++
++ auto sub = m_pendingUnsubscriptions.take(id);
++ if (Q_UNLIKELY(sub == nullptr)) {
++ qCDebug(lcMqttConnection) << "Received UNSUBACK for unknown request.";
++ return;
++ }
++
++ m_activeSubscriptions.remove(sub->topic(), sub);
++
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
++ readSubscriptionProperties(sub);
++ } else {
++ // 3.11.3 - The UNSUBACK Packet has no payload.
++ // emulate successful unsubscription
++ sub->d_func()->m_reasonCode = QMqtt::ReasonCode::Success;
++ sub->setState(QMqttSubscription::Unsubscribed);
++ return;
++ }
++
++ // 3.1.3 - The Payload contains a list of Reason Codes. Each Reason Code corresponds to a Topic Filter in the UNSUBSCRIBE packet being acknowledged.
++ // Whereas 3.10.3 states "The Payload of an UNSUBSCRIBE packet MUST contain at least one Topic Filter. An UNSUBSCRIBE packet with no Payload is a Protocol Error."
++ do {
++ const quint8 reasonCode = readBufferTyped<quint8>(&m_missingData);
++ sub->d_func()->m_reasonCode = QMqtt::ReasonCode(reasonCode);
++
++ // 3.11.3
++ switch (QMqtt::ReasonCode(reasonCode)) {
++ case QMqtt::ReasonCode::Success:
++ sub->setState(QMqttSubscription::Unsubscribed);
++ break;
++ case QMqtt::ReasonCode::NoSubscriptionExisted:
++ case QMqtt::ReasonCode::ImplementationSpecificError:
++ case QMqtt::ReasonCode::NotAuthorized:
++ case QMqtt::ReasonCode::InvalidTopicFilter:
++ case QMqtt::ReasonCode::MessageIdInUse:
++ case QMqtt::ReasonCode::UnspecifiedError:
++ qCWarning(lcMqttConnection) << "Unsubscription for id " << id << " failed. Reason Code:" << reasonCode;
++ sub->setState(QMqttSubscription::Error);
++ break;
++ default:
++ qCWarning(lcMqttConnection) << "Received illegal UNSUBACK reason code:" << reasonCode;
++ closeConnection(QMqttClient::ProtocolViolation);
++ break;
++ }
++ } while (m_missingData > 0);
++ }
++
++ void QMqttConnection::finalize_publish()
++ {
++ // String topic
++ QMqttTopicName topic = readBufferTyped<QString>(&m_missingData);
++ const int topicLength = topic.name().size();
++
++ quint16 id = 0;
++ if (m_currentPublish.qos > 0)
++ id = readBufferTyped<quint16>(&m_missingData);
++
++ QMqttPublishProperties publishProperties;
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
++ readPublishProperties(publishProperties);
++
++ if (publishProperties.availableProperties() & QMqttPublishProperties::TopicAlias) {
++ const quint16 topicAlias = publishProperties.topicAlias();
++ // m_receiveAliases is sized from the maximum this client declared in CONNECT,
++ // so bounds-check against the list itself rather than against a property value.
++ if (topicAlias == 0 || topicAlias > m_receiveAliases.size()) {
++ qCDebug(lcMqttConnection) << "TopicAlias receive: overflow.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return;
++ }
++ if (topicLength == 0) { // New message on existing topic alias
++ topic = m_receiveAliases.at(topicAlias - 1);
++ if (topic.name().isEmpty()) {
++ qCDebug(lcMqttConnection) << "TopicAlias receive: alias for unknown topic.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return;
++ }
++ qCDebug(lcMqttConnectionVerbose) << "TopicAlias receive: Using " << topicAlias;
++ } else { // Resetting a topic alias
++ qCDebug(lcMqttConnection) << "TopicAlias receive: Resetting:" << topic.name() << " : " << topicAlias;
++ m_receiveAliases[topicAlias - 1] = topic;
++ }
++ }
++
++ // message
++ const quint64 payloadLength = quint64(m_missingData);
++ const QByteArray message = readBuffer(payloadLength);
++ m_missingData -= payloadLength;
++
++ qCDebug(lcMqttConnectionVerbose) << "Finalize PUBLISH: topic:" << topic
++ << " payloadLength:" << payloadLength;
++
++ emit m_clientPrivate->m_client->messageReceived(message, topic);
++
++ QMqttMessage qmsg(topic, message, id, m_currentPublish.qos,
++ m_currentPublish.dup, m_currentPublish.retain);
++ qmsg.d->m_publishProperties = publishProperties;
++
++ if (id != 0) {
++ QMqttMessageStatusProperties statusProp;
++ statusProp.data->userProperties = publishProperties.userProperties();
++ emit m_clientPrivate->m_client->messageStatusChanged(id, QMqtt::MessageStatus::Published, statusProp);
++ }
++
++ // Store subscriptions in a temporary container as each messageReceived is allowed to subscribe
++ // again and thus invalid the iterator of the loop.
++ QList<QMqttSubscription *> subscribers;
++ for (const auto [key, value] : m_activeSubscriptions.asKeyValueRange()) {
++ if (key.match(topic))
++ subscribers.append(value);
++ }
++ for (const auto &s : subscribers)
++ emit s->messageReceived(qmsg);
++
++ if (m_currentPublish.qos == 1)
++ sendControlPublishAcknowledge(id);
++ else if (m_currentPublish.qos == 2)
++ sendControlPublishReceive(id);
++ }
++
++ void QMqttConnection::finalize_pubAckRecRelComp()
++ {
++ qCDebug(lcMqttConnectionVerbose) << "Finalize PUBACK/REC/REL/COMP";
++ const quint16 id = readBufferTyped<quint16>(&m_missingData);
++
++ QMqttMessageStatusProperties properties;
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0 && m_missingData > 0) {
++ // Reason Code (1byte)
++ const quint8 reasonCode = readBufferTyped<quint8>(&m_missingData);
++ properties.data->reasonCode = QMqtt::ReasonCode(reasonCode);
++
++ if ((m_currentPacket & 0xF0) == QMqttControlPacket::PUBACK || (m_currentPacket & 0xF0) == QMqttControlPacket::PUBREC) {
++ // 3.4.2.1, 3.5.2.1
++ switch (QMqtt::ReasonCode(reasonCode)) {
++ case QMqtt::ReasonCode::Success:
++ case QMqtt::ReasonCode::NoMatchingSubscriber:
++ case QMqtt::ReasonCode::UnspecifiedError:
++ case QMqtt::ReasonCode::ImplementationSpecificError:
++ case QMqtt::ReasonCode::NotAuthorized:
++ case QMqtt::ReasonCode::InvalidTopicName:
++ case QMqtt::ReasonCode::MessageIdInUse:
++ case QMqtt::ReasonCode::QuotaExceeded:
++ case QMqtt::ReasonCode::InvalidPayloadFormat:
++ break;
++ default:
++ qCWarning(lcMqttConnection) << "Received illegal PUBACK/REC reason code:" << reasonCode;
++ closeConnection(QMqttClient::ProtocolViolation);
++ return;
++ }
++ } else {
++ // 3.6.2.1, 3.7.2.1
++ switch (QMqtt::ReasonCode(reasonCode)) {
++ case QMqtt::ReasonCode::Success:
++ case QMqtt::ReasonCode::MessageIdNotFound:
++ break;
++ default:
++ qCWarning(lcMqttConnection) << "Received illegal PUBREL/COMP reason code:" << reasonCode;
++ closeConnection(QMqttClient::ProtocolViolation);
++ return;
++ }
++ }
++ if (m_missingData > 0)
++ readMessageStatusProperties(properties);
++ }
++
++ if ((m_currentPacket & 0xF0) == QMqttControlPacket::PUBREL) {
++ qCDebug(lcMqttConnectionVerbose) << " PUBREL:" << id;
++ emit m_clientPrivate->m_client->messageStatusChanged(id, QMqtt::MessageStatus::Released, properties);
++ sendControlPublishComp(id);
++ return;
++ }
++
++ if ((m_currentPacket & 0xF0) == QMqttControlPacket::PUBCOMP) {
++ qCDebug(lcMqttConnectionVerbose) << " PUBCOMP:" << id;
++ auto pendingRelease = m_pendingReleaseMessages.take(id);
++ if (!pendingRelease)
++ qCDebug(lcMqttConnection) << "Received PUBCOMP for unknown released message.";
++ emit m_clientPrivate->m_client->messageStatusChanged(id, QMqtt::MessageStatus::Completed, properties);
++ emit m_clientPrivate->m_client->messageSent(id);
++ return;
++ }
++
++ auto pendingMsg = m_pendingMessages.take(id);
++ if (!pendingMsg) {
++ qCDebug(lcMqttConnection) << "Received PUBACK for unknown message: " << id;
++ return;
++ }
++ if ((m_currentPacket & 0xF0) == QMqttControlPacket::PUBREC) {
++ qCDebug(lcMqttConnectionVerbose) << " PUBREC:" << id;
++ m_pendingReleaseMessages.insert(id, pendingMsg);
++ emit m_clientPrivate->m_client->messageStatusChanged(id, QMqtt::MessageStatus::Received, properties);
++ sendControlPublishRelease(id);
++ } else {
++ qCDebug(lcMqttConnectionVerbose) << " PUBACK:" << id;
++ emit m_clientPrivate->m_client->messageStatusChanged(id, QMqtt::MessageStatus::Acknowledged, properties);
++ emit m_clientPrivate->m_client->messageSent(id);
++ }
++ }
++
++ void QMqttConnection::finalize_pingresp()
++ {
++ qCDebug(lcMqttConnectionVerbose) << "Finalize PINGRESP";
++ const quint8 v = readBufferTyped<quint8>(&m_missingData);
++
++ if (v != 0) {
++ qCDebug(lcMqttConnection) << "Received a PINGRESP including payload.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return;
++ }
++
++ // Unexpected PINGRESP from the network should not push the counter below zero.
++ if (m_pingTimeout > 0)
++ m_pingTimeout--;
++ emit m_clientPrivate->m_client->pingResponseReceived();
++ }
++
++ bool QMqttConnection::processDataHelper()
++ {
++ if (m_missingData > 0) {
++ // MQTT-2.1.4 The remaining length in the fixed header says where the message ends.
++ const qint64 packetEnd = qint64(m_readPosition) + m_missingData;
++ if (m_internalState == BrokerConnected || m_internalState == BrokerWaitForConnectAck) {
++ const quint32 maxSize = m_clientPrivate->m_connectionProperties.maximumPacketSize();
++ if (maxSize > 0 && packetEnd > maxSize) {
++ qCDebug(lcMqttConnection) << "Packet size exceeds the maximum packet size.";
++ if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
++ // MQTT 5: 3.1.2.11.4 Maximum Packet Size
++ // If client receives a packet whose size exceeds this limit,
++ // this is a Protocol Error, the Client uses DISCONNECT with
++ // Reason Code 0x95 (Packet too large).
++ m_clientPrivate->m_client->setError(QMqttClient::ProtocolViolation);
++ sendControlDisconnect(QMqtt::ReasonCode::PacketTooLarge);
++ } else {
++ // MQTT 3.1 and 3.1.1 do not define a maximum packet size property, but we
++ // can still enforce the limit locally and close the connection if the broker
++ // attempts to send a packet that exceeds it.
++ // There are no reason codes before MQTT 5.0, and a DISCONNECT is not
++ // required. Just close the connection.
++ closeConnection(QMqttClient::ProtocolViolation);
++ }
++ return false;
++ }
++ }
++
++ if ((m_readBuffer.size() - m_readPosition) < m_missingData)
++ return false;
++
++ switch (m_currentPacket & 0xF0) {
++ case QMqttControlPacket::AUTH:
++ finalize_auth();
++ break;
++ case QMqttControlPacket::CONNACK:
++ finalize_connack();
++ break;
++ case QMqttControlPacket::SUBACK:
++ finalize_suback();
++ break;
++ case QMqttControlPacket::UNSUBACK:
++ finalize_unsuback();
++ break;
++ case QMqttControlPacket::PUBLISH:
++ finalize_publish();
++ break;
++ case QMqttControlPacket::PUBACK:
++ case QMqttControlPacket::PUBREC:
++ case QMqttControlPacket::PUBREL:
++ case QMqttControlPacket::PUBCOMP:
++ finalize_pubAckRecRelComp();
++ break;
++ case QMqttControlPacket::PINGRESP:
++ finalize_pingresp();
++ break;
++ default:
++ qCDebug(lcMqttConnection) << "Unknown packet to finalize.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ break;
++ }
++
++ if (m_internalState == BrokerDisconnected)
++ return false;
++
++ // Report bytes that no handler read.
++ if (m_readPosition != packetEnd) {
++ qCDebug(lcMqttConnection) << "Packet type" << (m_currentPacket & 0xF0)
++ << "left" << (packetEnd - m_readPosition)
++ << "bytes unread; resynchronizing.";
++ }
++
++ m_missingData = 0;
++ // A handler may stop reading early, so continue at the end of the message.
++ m_readBuffer = m_readBuffer.mid(packetEnd);
++ m_readPosition = 0;
++ }
++
++ // MQTT-2.2 A fixed header of a control packet must be at least 2 bytes. If the payload is
++ // longer than 127 bytes the header can be up to 5 bytes long.
++ switch (m_readBuffer.size()) {
++ case 0:
++ case 1:
++ return false;
++ case 2:
++ if ((m_readBuffer.at(1) & 128) != 0)
++ return false;
++ break;
++ case 3:
++ if ((m_readBuffer.at(1) & 128) != 0 && (m_readBuffer.at(2) & 128) != 0)
++ return false;
++ break;
++ case 4:
++ if ((m_readBuffer.at(1) & 128) != 0 && (m_readBuffer.at(2) & 128) != 0 && (m_readBuffer.at(3) & 128) != 0)
++ return false;
++ break;
++ default:
++ break;
++ }
++
++ readBuffer(reinterpret_cast<char *>(&m_currentPacket), 1);
++
++ switch (m_currentPacket & 0xF0) {
++ case QMqttControlPacket::CONNACK: {
++ qCDebug(lcMqttConnectionVerbose) << "Received CONNACK";
++ if (m_internalState != BrokerWaitForConnectAck) {
++ qCDebug(lcMqttConnection) << "Received CONNACK at unexpected time.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++
++ qint32 payloadSize = readVariableByteInteger();
++ if (payloadSize < 0) // Connection closed inside readVariableByteInteger
++ return false;
++ if (m_clientPrivate->m_protocolVersion != QMqttClient::MQTT_5_0) {
++ if (payloadSize != 2) {
++ qCDebug(lcMqttConnection) << "Unexpected FRAME size in CONNACK.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ }
++ m_missingData = payloadSize;
++ break;
++ }
++ case QMqttControlPacket::SUBACK: {
++ qCDebug(lcMqttConnectionVerbose) << "Received SUBACK";
++ // We cannot use quint8 here, because quint8 is one byte while MQTT 3.1.1 section
++ // 3.9.1 says the remaining length of a SUBACK is "the length of variable header
++ // (2 bytes) plus the length of the payload". The payload carries one reason code
++ // per topic filter, so 126 filters already exceed one byte. MQTT 5.0 section
++ // 3.9.1 adds properties on top, which exceed it on their own.
++ const qint32 remaining = readVariableByteInteger();
++ if (remaining < 0)
++ return false; // Connection closed inside readVariableByteInteger
++ // The remaining length counts the bytes that follow the fixed header, so the
++ // smallest value a SUBACK may declare is the size of its shortest legal body:
++ // 2 bytes packet identifier MQTT-3.9.2
++ // + 1 byte property length, MQTT 5.0 only, mandatory
++ // even when there are no properties MQTT-3.9.2.1
++ // + 1 byte reason code, at least one is required MQTT-3.9.3
++ // = 3 for MQTT 3.1.1, 4 for MQTT 5.0
++ // A smaller value leaves the reason code loop in finalize_suback() with
++ // nothing to read.
++ const qint32 minimumRemainingLength =
++ m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0 ? 4 : 3;
++ if (remaining < minimumRemainingLength) {
++ qCDebug(lcMqttConnection) << "Received SUBACK with invalid remaining length.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ m_missingData = remaining;
++ break;
++ }
++ case QMqttControlPacket::PUBLISH: {
++ qCDebug(lcMqttConnectionVerbose) << "Received PUBLISH";
++ m_currentPublish.dup = m_currentPacket & 0x08;
++ m_currentPublish.qos = (m_currentPacket & 0x06) >> 1;
++ m_currentPublish.retain = m_currentPacket & 0x01;
++ if ((m_currentPublish.qos == 0 && m_currentPublish.dup != 0)
++ || m_currentPublish.qos > 2) {
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++
++ m_missingData = readVariableByteInteger();
++ if (m_missingData == -1)
++ return false; // Connection closed inside readVariableByteInteger
++ break;
++ }
++ case QMqttControlPacket::PINGRESP:
++ qCDebug(lcMqttConnectionVerbose) << "Received PINGRESP";
++ m_missingData = 1;
++ break;
++
++
++ case QMqttControlPacket::PUBREL: {
++ qCDebug(lcMqttConnectionVerbose) << "Received PUBREL";
++ // We cannot use quint8 here, because quint8 is one byte while the remaining
++ // length is a variable byte integer of up to four, MQTT 5.0 section 2.1.4. In
++ // MQTT 3.1.1 a PUBREL carried nothing but the packet identifier, so the value
++ // was always 2 (section 3.6.1) and one byte was always enough. MQTT 5.0 changed
++ // that by adding an optional reason code and properties, section 3.6.2.
++ const qint32 remaining = readVariableByteInteger();
++ if (remaining < 0)
++ return false; // Connection closed inside readVariableByteInteger
++ if (m_clientPrivate->m_protocolVersion != QMqttClient::MQTT_5_0) {
++ // MQTT-3.6.1 Always exactly the packet identifier.
++ if (remaining != 0x02) {
++ qCDebug(lcMqttConnection) << "Received 2 byte message with invalid remaining length.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ } else if (remaining < 2) {
++ // MQTT-3.6.2.1 The reason code and the property length may both be omitted
++ // when the reason code would be Success and there are no properties, and
++ // only then is the value 2. A longer PUBREL is legal and must be accepted;
++ // rejecting it broke interoperability with compliant MQTT 5.0 brokers.
++ qCDebug(lcMqttConnection) << "Received PUBREL with invalid remaining length.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ // MQTT-3.6.1-1 The fixed header flags of a PUBREL are reserved and must be 0010.
++ if ((m_currentPacket & 0x0F) != 0x02) {
++ qCDebug(lcMqttConnection) << "Malformed fixed header for PUBREL.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ m_missingData = remaining;
++ break;
++ }
++
++ case QMqttControlPacket::UNSUBACK:
++ case QMqttControlPacket::PUBACK:
++ case QMqttControlPacket::PUBREC:
++ case QMqttControlPacket::PUBCOMP: {
++ qCDebug(lcMqttConnectionVerbose) << "Received UNSUBACK/PUBACK/PUBREC/PUBCOMP";
++ if ((m_currentPacket & 0x0F) != 0) {
++ qCDebug(lcMqttConnection) << "Malformed fixed header for UNSUBACK/PUBACK/PUBREC/PUBCOMP.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ // We cannot use quint8 here, because quint8 is one byte while the remaining
++ // length is a variable byte integer of up to four, MQTT 5.0 section 2.1.4. In
++ // MQTT 3.1.1 these packets carried nothing but the packet identifier, so the
++ // value was always 2 (sections 3.4.1, 3.5.1, 3.7.1 and 3.11.1) and one byte was
++ // always enough. MQTT 5.0 changed that by adding a reason code and properties,
++ // and a reason string or user properties push the value past 127.
++ const qint32 remaining = readVariableByteInteger();
++ if (remaining < 0)
++ return false; // Connection closed inside readVariableByteInteger
++ if (m_clientPrivate->m_protocolVersion != QMqttClient::MQTT_5_0) {
++ // MQTT-3.4.1, MQTT-3.5.1, MQTT-3.7.1, MQTT-3.11.1 Always exactly the packet
++ // identifier.
++ if (remaining != 0x02) {
++ qCDebug(lcMqttConnection) << "Received 2 byte message with invalid remaining length.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ } else {
++ // The smallest remaining length the specification permits, i.e. the size of
++ // the shortest legal body:
++ // PUBACK/PUBREC/PUBCOMP: 2 bytes packet identifier. The reason code and
++ // property length may both be omitted when the reason code would be
++ // Success and there are no properties. MQTT-3.4.2.1
++ // UNSUBACK: 2 bytes packet identifier
++ // + 1 byte property length, mandatory even when
++ // there are no properties MQTT-3.11.2.1
++ // + 1 byte reason code, at least one is required MQTT-3.11.3
++ // = 4 bytes. Anything shorter leaves the reason code loop in
++ // finalize_unsuback() with nothing to read.
++ const qint32 minimumRemainingLength =
++ (m_currentPacket & 0xF0) == QMqttControlPacket::UNSUBACK ? 4 : 2;
++ if (remaining < minimumRemainingLength) {
++ qCDebug(lcMqttConnection) << "Received UNSUBACK/PUBACK/PUBREC/PUBCOMP with "
++ "invalid remaining length.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ }
++ m_missingData = remaining;
++ break;
++ }
++ case QMqttControlPacket::AUTH:
++ if (m_clientPrivate->m_protocolVersion != QMqttClient::MQTT_5_0) {
++ qCDebug(lcMqttConnection) << "Received unknown command.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ qCDebug(lcMqttConnectionVerbose) << "Received AUTH";
++ if ((m_currentPacket & 0x0F) != 0) {
++ qCDebug(lcMqttConnection) << "Malformed fixed header for AUTH.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ m_missingData = readVariableByteInteger();
++ if (m_missingData == -1)
++ return false; // Connection closed inside readVariableByteInteger
++ break;
++ case QMqttControlPacket::DISCONNECT:
++ if (m_clientPrivate->m_protocolVersion != QMqttClient::MQTT_5_0) {
++ qCDebug(lcMqttConnection) << "Received unknown command.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ qCDebug(lcMqttConnectionVerbose) << "Received DISCONNECT";
++ if ((m_currentPacket & 0x0F) != 0) {
++ qCDebug(lcMqttConnection) << "Malformed fixed header for DISCONNECT.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ if (m_internalState != BrokerConnected) {
++ qCDebug(lcMqttConnection) << "Received DISCONNECT at unexpected time.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++ closeConnection(QMqttClient::NoError);
++ return false;
++ default:
++ qCDebug(lcMqttConnection) << "Received unknown command.";
++ closeConnection(QMqttClient::ProtocolViolation);
++ return false;
++ }
++
++ /* set current command CONNACK - PINGRESP */
++ /* read command size */
++ /* calculate missing_data */
++ return true; // reiterate. implicitly finishes and enqueues
++ }
++
++ void QMqttConnection::processData()
++ {
++ while (processDataHelper())
++ ;
++ }
++
++ bool QMqttConnection::writePacketToTransport(const QMqttControlPacket &p)
++ {
++ const QByteArray writeData = p.serialize();
++ qCDebug(lcMqttConnectionVerbose) << Q_FUNC_INFO << " DataSize:" << writeData.size();
++ const qint64 res = m_transport->write(writeData.constData(), writeData.size());
++ if (Q_UNLIKELY(res == -1)) {
++ qCDebug(lcMqttConnection) << "Could not write frame to transport.";
++ return false;
++ }
++ return true;
++ }
++
++ QT_END_NAMESPACE