Add support for nolocal subscriptions

No Local subscription allow for filtering on the server side.
If a client subscribes and publishes a message on the same topic,
it will not be routed back.

Fixes: QTBUG-100862
Change-Id: I363361d0267aba740d07483c04ce0d4c30b5e14c
Reviewed-by: Oliver Wolff <oliver.wolff@qt.io>

5 files changed, 79 insertions, 1 deletions

diff --git a/src/mqtt/qmqttconnection.cpp b/src/mqtt/qmqttconnection.cpp
index 8eb0c65..1f43c2c 100644
--- a/src/mqtt/qmqttconnection.cpp
+++ b/src/mqtt/qmqttconnection.cpp
@@ -509,7 +509,10 @@ QMqttSubscription *QMqttConnection::sendControlSubscribe(const QMqttTopicFilter
         packet.appendRaw(writeSubscriptionProperties(properties));
 
     packet.append(topic.filter().toUtf8());
-    packet.append(char(qos));
+    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);
diff --git a/src/mqtt/qmqttsubscriptionproperties.cpp b/src/mqtt/qmqttsubscriptionproperties.cpp
index 08977a9..fe77653 100644
--- a/src/mqtt/qmqttsubscriptionproperties.cpp
+++ b/src/mqtt/qmqttsubscriptionproperties.cpp
@@ -66,6 +66,7 @@ class QMqttSubscriptionPropertiesData : public QSharedData
 public:
     quint32 subscriptionIdentifier{0};
     QMqttUserProperties userProperties;
+    bool noLocal{false};
 };
 
 class QMqttUnsubscriptionPropertiesData : public QSharedData
@@ -129,6 +130,27 @@ void QMqttSubscriptionProperties::setSubscriptionIdentifier(quint32 id)
 }
 
 /*!
+ * \since 6.4
+ * Returns true if the subscription shall not receive local messages on the same topic.
+ */
+bool QMqttSubscriptionProperties::noLocal() const
+{
+    return data->noLocal;
+}
+
+/*!
+ * \since 6.4
+ * Sets the subscription option to not receive local message.
+ * When a client publishes a message with the same topic as an existing local subscription
+ * the server by default sends the message back to the client. If \a noloc is set to true
+ * the broker will not send any message the same client has published.
+ */
+void QMqttSubscriptionProperties::setNoLocal(bool noloc)
+{
+    data->noLocal = noloc;
+}
+
+/*!
     \internal
 */
 QMqttUnsubscriptionProperties::QMqttUnsubscriptionProperties() : data(new QMqttUnsubscriptionPropertiesData)
diff --git a/src/mqtt/qmqttsubscriptionproperties.h b/src/mqtt/qmqttsubscriptionproperties.h
index b5a4864..6d82895 100644
--- a/src/mqtt/qmqttsubscriptionproperties.h
+++ b/src/mqtt/qmqttsubscriptionproperties.h
@@ -56,6 +56,9 @@ public:
 
     quint32 subscriptionIdentifier() const;
     void setSubscriptionIdentifier(quint32 id);
+
+    bool noLocal() const;
+    void setNoLocal(bool noloc);
 private:
     QSharedDataPointer<QMqttSubscriptionPropertiesData> data;
 };
diff --git a/tests/auto/qmqttsubscription/tst_qmqttsubscription.cpp b/tests/auto/qmqttsubscription/tst_qmqttsubscription.cpp
index 3391792..5cfcacc 100644
--- a/tests/auto/qmqttsubscription/tst_qmqttsubscription.cpp
+++ b/tests/auto/qmqttsubscription/tst_qmqttsubscription.cpp
@@ -52,6 +52,8 @@ private Q_SLOTS:
     void sharedConnection();
     void sharedNonShared_data();
     void sharedNonShared();
+    void noLocal_data();
+    void noLocal();
 private:
     void createAndSubscribe(QMqttClient *c, QMqttSubscription **sub, const QString &topic);
     QProcess m_brokerProcess;
@@ -377,6 +379,50 @@ void Tst_QMqttSubscription::sharedNonShared()
 //    QTRY_VERIFY(receivalSpy2.count() == 1);
 }
 
+void Tst_QMqttSubscription::noLocal_data()
+{
+    QTest::addColumn<QMqttClient::ProtocolVersion>("version");
+    QTest::addColumn<bool>("non");
+    QTest::newRow("3.1.1 - false") << QMqttClient::MQTT_3_1_1 << false;
+    QTest::newRow("3.1.1 - true") << QMqttClient::MQTT_3_1_1 << true;
+    QTest::newRow("5.0 - false") << QMqttClient::MQTT_5_0 << false;
+    QTest::newRow("5.0 - true") << QMqttClient::MQTT_5_0 << true;
+}
+
+void Tst_QMqttSubscription::noLocal()
+{
+    QFETCH(QMqttClient::ProtocolVersion, version);
+    QFETCH(bool, non);
+
+    QMqttClient client;
+    client.setProtocolVersion(version);
+    client.setHostname(m_testBroker);
+    client.setPort(m_port);
+    client.connectToHost();
+    QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker.");
+
+    QMqttSubscriptionProperties subProps;
+    subProps.setNoLocal(non);
+
+    const QString topic(QLatin1String("Qt/Subscription/NonLoc"));
+
+    auto sub = client.subscribe(topic, subProps, 1);
+    QTRY_VERIFY2(sub->state() == QMqttSubscription::Subscribed, "Could not subscribe to topic.");
+
+    QSignalSpy publishSpy(&client, SIGNAL(messageSent(qint32)));
+    QSignalSpy receivalSpy(sub, SIGNAL(messageReceived(QMqttMessage)));
+
+    client.publish(topic, "content", 1);
+    QTRY_VERIFY(publishSpy.count() == 1);
+
+    if (version == QMqttClient::MQTT_3_1_1 || !non) { // 3.1.1 does not know NoLocal and sends to subscription
+        QTRY_VERIFY(receivalSpy.count() == 1);
+    } else {
+        QTest::qWait(3000);
+        QCOMPARE(receivalSpy.count(), 0);
+    }
+}
+
 QTEST_MAIN(Tst_QMqttSubscription)
 
 #include "tst_qmqttsubscription.moc"
diff --git a/tests/auto/qmqttsubscriptionproperties/tst_qmqttsubscriptionproperties.cpp b/tests/auto/qmqttsubscriptionproperties/tst_qmqttsubscriptionproperties.cpp
index 00c7a3d..d48fded 100644
--- a/tests/auto/qmqttsubscriptionproperties/tst_qmqttsubscriptionproperties.cpp
+++ b/tests/auto/qmqttsubscriptionproperties/tst_qmqttsubscriptionproperties.cpp
@@ -77,6 +77,10 @@ void tst_QMqttSubscriptionProperties::getSet()
     properties.setSubscriptionIdentifier(id);
     QCOMPARE(properties.subscriptionIdentifier(), id);
 
+    QCOMPARE(properties.noLocal(), false);
+    properties.setNoLocal(true);
+    QCOMPARE(properties.noLocal(), true);
+
     const QString userKey1 = QLatin1String("UserName1");
     const QString userValue1 = QLatin1String("SomeValue");
     const QString userKey2 = QLatin1String("UserName2");