MQTT5: Introduce support for topic aliases

Task-number: QTPM-1490
Change-Id: Ie666d4c0dd75c08083a6860f345144ee1efb3d83
Reviewed-by: hjk <hjk@qt.io>

src/mqtt/qmqttconnection.cpp

@@ -361,7 +361,45 @@ qint32 QMqttConnection::sendControlPublish(const QMqttTopicName &topic,
         header |= 0x01;
 
     QSharedPointer<QMqttControlPacket> packet(new QMqttControlPacket(header));
-    packet->append(topic.name().toUtf8());
+    // topic alias
+    QMqttPublishProperties publishProperties(properties);
+    if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
+        const quint16 topicAlias = publishProperties.topicAlias();
+        if (topicAlias > 0) { // User specified topic Alias
+            if (topicAlias > m_clientPrivate->m_serverConnectionProperties.maximumTopicAlias()) {
+                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();
@@ -370,7 +408,7 @@ qint32 QMqttConnection::sendControlPublish(const QMqttTopicName &topic,
     }
 
     if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
-        packet->appendRaw(writePublishProperties(properties));
+        packet->appendRaw(writePublishProperties(publishProperties));
 
     packet->appendRaw(message);
 
@@ -549,6 +587,9 @@ bool QMqttConnection::sendControlDisconnect()
 
     m_activeSubscriptions.clear();
 
+    m_receiveAliases.clear();
+    m_publishAliases.clear();
+
     const QMqttControlPacket packet(QMqttControlPacket::DISCONNECT);
     if (!writePacketToTransport(packet)) {
         qCDebug(lcMqttConnection) << "Failed to write DISCONNECT to transport.";
@@ -1340,8 +1381,11 @@ void QMqttConnection::finalize_connack()
     }
 
     // MQTT 5.0 has variable part != 2 in the header
-    if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
+    if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
         readConnackProperties();
+        m_receiveAliases.resize(m_clientPrivate->m_serverConnectionProperties.maximumTopicAlias());
+        m_publishAliases.resize(m_clientPrivate->m_connectionProperties.maximumTopicAlias());
+    }
 
     m_internalState = BrokerConnected;
     m_clientPrivate->setStateAndError(QMqttClient::Connected);
@@ -1445,7 +1489,8 @@ void QMqttConnection::finalize_unsuback()
 void QMqttConnection::finalize_publish()
 {
     // String topic
-    const QMqttTopicName topic = readBufferTyped<QString>(&m_missingData);
+    QMqttTopicName topic = readBufferTyped<QString>(&m_missingData);
+    const int topicLength = topic.name().length();
 
     quint16 id = 0;
     if (m_currentPublish.qos > 0)
@@ -1455,6 +1500,24 @@ void QMqttConnection::finalize_publish()
     if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0)
         readPublishProperties(publishProperties);
 
+    const quint16 topicAlias = publishProperties.topicAlias();
+    if (topicAlias > 0) {
+        if (topicAlias > m_clientPrivate->m_serverConnectionProperties.maximumTopicAlias()) {
+            qCDebug(lcMqttConnection) << "TopicAlias receive: overflow.";
+            closeConnection(QMqttClient::ProtocolViolation);
+        } else if (topicLength == 0) { // New message on existing topic alias
+            if (m_receiveAliases.at(topicAlias - 1).name().isEmpty()) {
+                qCDebug(lcMqttConnection) << "TopicAlias receive: alias for unknown topic.";
+                closeConnection(QMqttClient::ProtocolViolation);
+            }
+            qCDebug(lcMqttConnectionVerbose) << "TopicAlias receive: Using " << topicAlias;
+            topic = m_receiveAliases.at(topicAlias - 1);
+        } 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);
@@ -1724,6 +1787,7 @@ void QMqttConnection::processData()
 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.";

src/mqtt/qmqttconnection_p.h

@@ -152,6 +152,9 @@ private:
     QMap<quint16, QSharedPointer<QMqttControlPacket>> m_pendingReleaseMessages;
     InternalConnectionState m_internalState{BrokerDisconnected};
     QTimer m_pingTimer;
+
+    QVector<QMqttTopicName> m_receiveAliases;
+    QVector<QMqttTopicName> m_publishAliases;
 };
 
 QT_END_NAMESPACE

tests/auto/qmqttconnectionproperties/tst_qmqttconnectionproperties.cpp

@@ -28,6 +28,7 @@
 
 #include "broker_connection.h"
 
+#include <QtCore/QLoggingCategory>
 #include <QtCore/QString>
 #include <QtTest/QtTest>
 #include <QtMqtt/QMqttClient>
@@ -49,6 +50,7 @@ private Q_SLOTS:
     void receiveServerProperties();
     void maximumPacketSize();
     void maximumTopicAlias();
+    void maximumTopicAliasReceive();
     void assignedClientId();
     void userProperties();
 private:
@@ -228,8 +230,139 @@ void tst_QMqttConnectionProperties::maximumPacketSize()
 
 void tst_QMqttConnectionProperties::maximumTopicAlias()
 {
-    // ### TODO: implement
-    QSKIP("Needs to be implemented.");
+    const QByteArray msgContent("SomeContent");
+    QMqttClient client;
+    client.setProtocolVersion(QMqttClient::MQTT_5_0);
+    client.setHostname(m_testBroker);
+    client.setPort(m_port);
+
+    const int userMaxTopicAlias = 9;
+    QMqttConnectionProperties userProperties;
+    userProperties.setMaximumTopicAlias(userMaxTopicAlias);
+    client.setConnectionProperties(userProperties);
+
+    client.connectToHost();
+    QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker.");
+
+    int publishTransportSize = 0;
+    auto transport = client.transport();
+    QSignalSpy transportSpy(transport, SIGNAL(bytesWritten(qint64))); //&QIODevice::bytesWritten);
+
+    const auto serverProperties = client.serverConnectionProperties();
+    const quint16 serverMaxAlias = serverProperties.maximumTopicAlias();
+    if (serverMaxAlias == 0)
+        QSKIP("Need to skip this test due to topic aliases not supported on server");
+
+    //qDebug() << "Server Max Alias:" << serverMaxAlias;
+    //QLoggingCategory::setFilterRules("qt.mqtt.connection*=true");
+
+    // Fill up the internal publish vector
+    const QLatin1String topicBase("Qt/connprop/alias/top");
+    for (quint16 i = 0; i < serverMaxAlias; ++i) {
+        QSignalSpy publishSpy(&client, SIGNAL(messageSent(qint32)));
+
+        QMqttTopicName topic(topicBase + QString::number(i));
+        client.publish(topic, msgContent, 1);
+        QTRY_VERIFY(publishSpy.count() == 1);
+
+        QVERIFY(transportSpy.count() == 1);
+        const int dataSize = transportSpy.at(0).at(0).toInt();
+        if (publishTransportSize == 0) {
+            publishTransportSize = dataSize;
+        } else {
+            QCOMPARE(dataSize, publishTransportSize);
+        }
+        transportSpy.clear();
+    }
+
+    // Verify non auto assignable defaults to old behavior
+    QSignalSpy fullSpy(&client, SIGNAL(messageSent(qint32)));
+    transportSpy.clear();
+    client.publish(QLatin1String("Qt/connprop/alias/full/with/long/topic/to/verify/bigger/size"), msgContent, 1);
+    QTRY_VERIFY(fullSpy.count() == 1);
+    QVERIFY(transportSpy.count() == 1);
+    QVERIFY(transportSpy.at(0).at(0).toInt() > publishTransportSize);
+
+    // Verify alias is used at sending second time
+    transportSpy.clear();
+    fullSpy.clear();
+
+    client.publish(topicBase + QLatin1String("0"), msgContent, 1);
+    QTRY_VERIFY(fullSpy.count() == 1);
+    QVERIFY(transportSpy.count() == 1);
+    int usageSize = transportSpy.at(0).at(0).toInt();
+    QVERIFY(usageSize < publishTransportSize);
+
+    // Manually overwrite topic alias 1
+    const QMqttTopicName overwrite(QLatin1String("Qt/connprop/alias/overwrite/with/long/topic/to/verify/reset"));
+    QMqttPublishProperties overProp;
+    overProp.setTopicAlias(2);
+    fullSpy.clear();
+    transportSpy.clear();
+    client.publish(overwrite, overProp, msgContent, 1);
+    QTRY_VERIFY(fullSpy.count() == 1);
+    QVERIFY(transportSpy.count() == 1);
+    const int overwriteSize = transportSpy.at(0).at(0).toInt();
+    QVERIFY(overwriteSize > publishTransportSize);
+    // After resend new alias should be used and msg size reduced
+    fullSpy.clear();
+    transportSpy.clear();
+    client.publish(overwrite, msgContent, 1);
+    QTRY_VERIFY(fullSpy.count() == 1);
+    QVERIFY(transportSpy.count() == 1);
+    usageSize = transportSpy.at(0).at(0).toInt();
+    QVERIFY(usageSize < overwriteSize);
+}
+
+void createTopicAliasClient(QMqttClient &client, const QString &hostname, quint16 port)
+{
+    client.setProtocolVersion(QMqttClient::MQTT_5_0);
+    client.setHostname(hostname);
+    client.setPort(port);
+
+    const int userMaxTopicAlias = 9;
+    QMqttConnectionProperties userProperties;
+    userProperties.setMaximumTopicAlias(userMaxTopicAlias);
+    client.setConnectionProperties(userProperties);
+}
+
+void tst_QMqttConnectionProperties::maximumTopicAliasReceive()
+{
+    const QByteArray msgContent("SomeContent");
+    const QString topic("Qt/connprop/receive/alias");
+
+    QMqttClient client;
+    createTopicAliasClient(client, m_testBroker, m_port);
+
+    client.connectToHost();
+    QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker.");
+
+    QMqttClient subscriber;
+    createTopicAliasClient(subscriber, m_testBroker, m_port);
+
+    subscriber.connectToHost();
+    QTRY_VERIFY2(subscriber.state() == QMqttClient::Connected, "Could not connect to broker.");
+
+    auto sub = subscriber.subscribe(topic + "/#", 1);
+
+    int receiveCounter = 0;
+    connect(sub, &QMqttSubscription::messageReceived, [&receiveCounter](QMqttMessage msg) {
+        qDebug() << "Received message with alias:" << msg.publishProperties().topicAlias();
+        receiveCounter++;
+    });
+
+    QTRY_VERIFY2(sub->state() == QMqttSubscription::Subscribed, "Could not subscribe");
+
+    QSignalSpy publishSpy(&client, &QMqttClient::messageSent);
+    //QLoggingCategory::setFilterRules("qt.mqtt.connection*=true");
+    client.publish(topic, msgContent, 1);
+    QTRY_VERIFY2(publishSpy.count() == 1, "Could not publish");
+    QTRY_VERIFY2(receiveCounter == 1, "Did not receive initial message");
+
+    publishSpy.clear();
+    client.publish(topic, msgContent, 1);
+    QTRY_VERIFY2(publishSpy.count() == 1, "Could not publish");
+    QTRY_VERIFY2(receiveCounter == 2, "Did not receive second aliases message");
 }
 
 void tst_QMqttConnectionProperties::assignedClientId()