From ac666587dd32e3b130295ca9084a407e12482dfc Mon Sep 17 00:00:00 2001 From: Maurice Kalinowski Date: Mon, 19 Mar 2018 15:55:58 +0100 Subject: [PATCH] MQTT5: Add initial support With this patch it is possible to set the version to MQTT5 and use the same features as against version 3.1.1. No MQTT5 specific features are enabled. Change-Id: I129ff8a2c996d7bcc582565fccda9e38edc48c62 Reviewed-by: Maurice Kalinowski --- src/mqtt/qmqttclient.cpp | 3 +- src/mqtt/qmqttclient.h | 3 +- src/mqtt/qmqttclient_p.h | 2 - src/mqtt/qmqttconnection.cpp | 89 +++++++++++++++++++--- tests/auto/conformance/tst_conformance.cpp | 17 ++++- 5 files changed, 98 insertions(+), 16 deletions(-) diff --git a/src/mqtt/qmqttclient.cpp b/src/mqtt/qmqttclient.cpp index f4a6dac..dcfb213 100644 --- a/src/mqtt/qmqttclient.cpp +++ b/src/mqtt/qmqttclient.cpp @@ -579,8 +579,7 @@ void QMqttClient::setProtocolVersion(ProtocolVersion protocolVersion) if (d->m_protocolVersion == protocolVersion) return; - // Only MQTT 3 and 4 are supported - if (protocolVersion < 3 || protocolVersion > 4) + if (protocolVersion < 3 || protocolVersion > 5) return; d->m_protocolVersion = protocolVersion; diff --git a/src/mqtt/qmqttclient.h b/src/mqtt/qmqttclient.h index 9b06ae7..6a26733 100644 --- a/src/mqtt/qmqttclient.h +++ b/src/mqtt/qmqttclient.h @@ -71,7 +71,8 @@ public: }; enum ProtocolVersion { MQTT_3_1 = 3, - MQTT_3_1_1 = 4 + MQTT_3_1_1 = 4, + MQTT_5_0 = 5 }; private: diff --git a/src/mqtt/qmqttclient_p.h b/src/mqtt/qmqttclient_p.h index 2bec37e..007293c 100644 --- a/src/mqtt/qmqttclient_p.h +++ b/src/mqtt/qmqttclient_p.h @@ -63,8 +63,6 @@ public: QMqttConnection m_connection; QString m_clientId; // auto-generated quint16 m_keepAlive{60}; - // 3 == MQTT Standard 3.1 - // 4 == MQTT Standard 3.1.1 QMqttClient::ProtocolVersion m_protocolVersion{QMqttClient::MQTT_3_1_1}; QMqttClient::ClientState m_state{QMqttClient::Disconnected}; QMqttClient::ClientError m_error{QMqttClient::NoError}; diff --git a/src/mqtt/qmqttconnection.cpp b/src/mqtt/qmqttconnection.cpp index 4d60ed0..f555d23 100644 --- a/src/mqtt/qmqttconnection.cpp +++ b/src/mqtt/qmqttconnection.cpp @@ -188,6 +188,10 @@ bool QMqttConnection::sendControlConnect() 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; default: qCWarning(lcMqttConnection) << "Illegal MQTT Version"; m_clientPrivate->setStateAndError(QMqttClient::Disconnected, QMqttClient::InvalidProtocolVersion); @@ -224,6 +228,10 @@ bool QMqttConnection::sendControlConnect() // 3.1.2.10 Keep Alive packet.append(m_clientPrivate->m_keepAlive); + if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) { + packet.append(char(0)); // No properties + // ### TODO: Idea is to introduce QMqttConnectionProperties struct + } // 3.1.3 Payload // 3.1.3.1 Client Identifier const QByteArray clientStringArray = m_clientPrivate->m_clientId.toUtf8(); @@ -235,6 +243,11 @@ bool QMqttConnection::sendControlConnect() } if (!m_clientPrivate->m_willMessage.isEmpty()) { + if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) { + // ### TODO: Add Will properties + packet.append(char(0)); + } + packet.append(m_clientPrivate->m_willTopic.toUtf8()); packet.append(m_clientPrivate->m_willMessage); } @@ -279,6 +292,12 @@ qint32 QMqttConnection::sendControlPublish(const QMqttTopicName &topic, const QB packet->append(identifier); m_pendingMessages.insert(identifier, packet); } + + if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) { + // ### TODO: Add publish properties + packet->append(char(0)); // No properties + } + packet->appendRaw(message); const bool written = writePacketToTransport(*packet.data()); @@ -340,6 +359,10 @@ QMqttSubscription *QMqttConnection::sendControlSubscribe(const QMqttTopicFilter packet.append(identifier); + if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) { + // ### TODO: Add properties + packet.append(char(0)); // No properties + } // Overflow protection if (!topic.isValid()) { qWarning("Subscribed topic filter is not valid."); @@ -574,8 +597,13 @@ QByteArray QMqttConnection::readBuffer(qint64 size) void QMqttConnection::finalize_connack() { qCDebug(lcMqttConnectionVerbose) << "Finalize CONNACK"; + + qint64 variableLength = m_missingData; + quint8 ackFlags; readBuffer((char*)&ackFlags, 1); + + variableLength--; if (ackFlags > 1) { // MQTT-3.2.2.1 qWarning("Unexpected CONNACK Flags set"); closeConnection(QMqttClient::ProtocolViolation); @@ -596,6 +624,7 @@ void QMqttConnection::finalize_connack() quint8 connectResultValue; readBuffer((char*)&connectResultValue, 1); + variableLength--; if (connectResultValue != 0) { qWarning("Connection has been rejected"); // MQTT-3.2.2-5 @@ -606,6 +635,13 @@ void QMqttConnection::finalize_connack() m_clientPrivate->setStateAndError(QMqttClient::Disconnected, static_cast(connectResultValue)); return; } + + // MQTT 5.0 has variable part != 2 in the header + if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) { + readBuffer(variableLength); + // ### TODO: Read variable content / connectionProperties + } + m_internalState = BrokerConnected; m_clientPrivate->setStateAndError(QMqttClient::Connected); @@ -622,7 +658,15 @@ void QMqttConnection::finalize_suback() qWarning("Received SUBACK for unknown subscription request"); return; } + + if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) { + // ### TODO: SubAck Properties + const quint32 propertySize = readVariableByteInteger(); + readBuffer(propertySize); + m_missingData = 0; + } quint8 result; + // ### TODO: In MQTT5 this might be a list of reason codes, not just granted QoS, see 3.9.3 readBuffer((char*)&result, 1); auto sub = m_pendingSubscriptionAck.take(id); qCDebug(lcMqttConnectionVerbose) << "Finalize SUBACK: id:" << id << "qos:" << result; @@ -669,6 +713,17 @@ void QMqttConnection::finalize_publish() id = qFromBigEndian(reinterpret_cast(readBuffer(2).constData())); } + if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) { + quint8 propertyLength; + readBuffer(reinterpret_cast(&propertyLength), 1); + m_missingData -= 1; + // ### TODO: Handle publish properties + if (propertyLength > 0) { + readBuffer(propertyLength); + m_missingData -= propertyLength; + } + } + // message qint64 payloadLength = m_missingData - (topicLength + 2) - (m_currentPublish.qos > 0 ? 2 : 0); const QByteArray message = readBuffer(payloadLength); @@ -698,7 +753,21 @@ void QMqttConnection::finalize_pubAckRecComp() quint16 id; readBuffer((char*)&id, 2); id = qFromBigEndian(id); + m_missingData -= 2; + if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0 && m_missingData > 0) { + // Reason Code (1byte) + quint8 reasonCode; + readBuffer(reinterpret_cast(&reasonCode), 1); + m_missingData--; + // Property Length (Variable Int) + qint32 byteCount = 0; + const qint32 propertyLength = readVariableByteInteger(&byteCount); + m_missingData -= byteCount; + // ### TODO: Publish ACK/REC/COMP property handling + if (propertyLength > 0) + readBuffer(propertyLength); + } if ((m_currentPacket & 0xF0) == QMqttControlPacket::PUBCOMP) { qCDebug(lcMqttConnectionVerbose) << " PUBCOMP:" << id; auto pendingRelease = m_pendingReleaseMessages.take(id); @@ -822,12 +891,14 @@ void QMqttConnection::processData() quint8 payloadSize; readBuffer((char*)&payloadSize, 1); - if (payloadSize != 2) { - qWarning("Unexpected FRAME size in CONNACK"); - closeConnection(QMqttClient::ProtocolViolation); - return; + if (m_clientPrivate->m_protocolVersion != QMqttClient::MQTT_5_0) { + if (payloadSize != 2) { + qWarning("Unexpected FRAME size in CONNACK"); + closeConnection(QMqttClient::ProtocolViolation); + return; + } } - m_missingData = 2; + m_missingData = payloadSize; break; } case QMqttControlPacket::SUBACK: { @@ -887,14 +958,14 @@ void QMqttConnection::processData() closeConnection(QMqttClient::ProtocolViolation); return; } - char remaining; - readBuffer(&remaining, 1); - if (remaining != 0x02) { + quint8 remaining; + readBuffer(reinterpret_cast(&remaining), 1); + if (m_clientPrivate->m_protocolVersion != QMqttClient::MQTT_5_0 && remaining != 0x02) { qWarning("Received 2 byte message with invalid remaining length"); closeConnection(QMqttClient::ProtocolViolation); return; } - m_missingData = 2; + m_missingData = remaining; break; } default: diff --git a/tests/auto/conformance/tst_conformance.cpp b/tests/auto/conformance/tst_conformance.cpp index ca3cc61..fcca6db 100644 --- a/tests/auto/conformance/tst_conformance.cpp +++ b/tests/auto/conformance/tst_conformance.cpp @@ -101,7 +101,7 @@ void Tst_MqttConformance::basic_test() const QString topic(QLatin1String("Qt/conformance")); - auto sub = client.subscribe(topic, 2); + auto sub = client.subscribe(topic, 1); QTRY_VERIFY2(sub->state() == QMqttSubscription::Subscribed, "Could not subscribe"); int msgCount = 0; @@ -110,6 +110,12 @@ void Tst_MqttConformance::basic_test() msgCount++; }); + connect(&client, &QMqttClient::messageReceived, this, [](const QByteArray &message, const QMqttTopicName &topic) + { + Q_UNUSED(message) + Q_UNUSED(topic) + }); + client.publish(topic, "qos 0", 0); client.publish(topic, "qos 1", 1); client.publish(topic, "qos 2", 2); @@ -242,7 +248,14 @@ void Tst_MqttConformance::zero_length_clientid_test() QVERIFY2(client.state() == QMqttClient::Connecting, "Could not set state to connecting."); if (!session) { - QTRY_VERIFY2(client.state() == QMqttClient::Disconnected, "Sessions with empty client should not be allowed."); + if (client.protocolVersion() == QMqttClient::MQTT_5_0) { + // For MQTT 5 the broker creates an ID and returns it in CONNACK + QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker."); + // ### TODO: Enable this check: + //QVERIFY(!client.clientId().isEmpty()) + } else { + QTRY_VERIFY2(client.state() == QMqttClient::Disconnected, "Sessions with empty client should not be allowed."); + } } else { QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker."); client.disconnectFromHost();