Make sure unsubscribe followed by subscribe returns new subscription

Since we could only have one subscription for each topic it was
not possible to return a new subscription when a subscription
request comes in on a topic that is scheduled for destruction.

We solve this by switching to a QMultiHash for the active
subscriptions, allowing more then one subscription for a
given topic. However, at most one of these subscriptions
are not present in the list of subscriptions scheduled to
be unsubscribed

This patch also changes multiple unsubscribes to the
same topic. Earlier all would return true, and
m_pendingUnsubscriptions would be updated for all
requests.
The new behavior is that only the first unsubscribe
is acted upon. Furthere requests will return false.

Fixes: QTBUG-135653
Change-Id: I5a79efb68b3191e185325a89cc2042c773bb5c12
Reviewed-by: Maurice Kalinowski <maurice.kalinowski@qt.io>
(cherry picked from commit c383083270)
Reviewed-by: Qt Cherry-pick Bot <cherrypick_bot@qt-project.org>
(cherry picked from commit 71bc423d11)
(cherry picked from commit c6a708a71a)

src/mqtt/qmqttconnection.cpp

@@ -56,6 +56,25 @@ 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)
 {
 }
@@ -456,18 +475,18 @@ QMqttSubscription *QMqttConnection::sendControlSubscribe(const QMqttTopicFilter
         const QString sharedSubscriptionName = topic.sharedSubscriptionName();
         if (!sharedSubscriptionName.isEmpty()) {
             const QMqttTopicFilter filter(topic.filter().section(QLatin1Char('/'), 2));
-            auto it = m_activeSubscriptions.constFind(filter);
-            if (it != m_activeSubscriptions.cend() && (*it)->sharedSubscriptionName() == sharedSubscriptionName)
-                return *it;
+            auto sub = findActiveSubscription(filter);
+            if (sub && (sub->sharedSubscriptionName() == sharedSubscriptionName))
+                return sub;
         } else {
-            auto it = m_activeSubscriptions.constFind(topic);
-            if (it != m_activeSubscriptions.cend() && !(*it)->isSharedSubscription())
-                return *it;
+            auto sub = findActiveSubscription(topic);
+            if (sub && !sub->isSharedSubscription())
+                return sub;
         }
     } else {
-        auto it = m_activeSubscriptions.constFind(topic);
-        if (it != m_activeSubscriptions.cend())
-            return *it;
+        auto sub = findActiveSubscription(topic);
+        if (sub)
+            return sub;
     }
 
     // has to have 0010 as bits 3-0, maybe update SUBSCRIBE instead?
@@ -519,11 +538,13 @@ bool QMqttConnection::sendControlUnsubscribe(const QMqttTopicFilter &topic, cons
     if (!topic.isValid())
         return false;
 
-    if (!m_activeSubscriptions.contains(topic))
+    auto sub = findActiveSubscription(topic);
+    if (!sub)
+        // Already unsubscribed
         return false;
 
     if (m_internalState != QMqttConnection::BrokerConnected) {
-        m_activeSubscriptions.remove(topic);
+        m_activeSubscriptions.remove(topic, sub);
         return true;
     }
 
@@ -542,7 +563,6 @@ bool QMqttConnection::sendControlUnsubscribe(const QMqttTopicFilter &topic, cons
     }
 
     packet.append(topic.filter().toUtf8());
-    auto sub = m_activeSubscriptions[topic];
     sub->setState(QMqttSubscription::UnsubscriptionPending);
 
     if (!writePacketToTransport(packet))
@@ -1559,7 +1579,7 @@ void QMqttConnection::finalize_unsuback()
         return;
     }
 
-    m_activeSubscriptions.remove(sub->topic());
+    m_activeSubscriptions.remove(sub->topic(), sub);
 
     if (m_clientPrivate->m_protocolVersion == QMqttClient::MQTT_5_0) {
         readSubscriptionProperties(sub);

src/mqtt/qmqttconnection_p.h

@@ -116,6 +116,9 @@ private:
     void closeConnection(QMqttClient::ClientError error);
     QByteArray readBuffer(quint64 size);
     template<typename T> T readBufferTyped(qint64 *dataSize = nullptr);
+    bool isPendingUnsubscribe(QMqttSubscription *) const;
+    QMqttSubscription *findActiveSubscription(const QMqttTopicFilter &topic) const;
+
     QByteArray m_readBuffer;
     int m_readPosition{0};
     qint64 m_missingData{0};
@@ -128,9 +131,13 @@ private:
     QMqttControlPacket::PacketType m_currentPacket{QMqttControlPacket::UNKNOWN};
 
     bool writePacketToTransport(const QMqttControlPacket &p);
+
+    // I: There can be many occurences of different QMqttSubscription * with
+    //    the same QMqttTopicFilter, but only one that is not in
+    //    m_pendingUnsubscriptions.
     QHash<quint16, QMqttSubscription *> m_pendingSubscriptionAck;
     QHash<quint16, QMqttSubscription *> m_pendingUnsubscriptions;
-    QHash<QMqttTopicFilter, QMqttSubscription *> m_activeSubscriptions;
+    QMultiHash<QMqttTopicFilter, QMqttSubscription *> m_activeSubscriptions;
     QHash<quint16, QSharedPointer<QMqttControlPacket>> m_pendingMessages;
     QHash<quint16, QSharedPointer<QMqttControlPacket>> m_pendingReleaseMessages;
     InternalConnectionState m_internalState{BrokerDisconnected};

tests/auto/qmqttsubscription/tst_qmqttsubscription.cpp

@@ -31,6 +31,9 @@ private Q_SLOTS:
     void noLocal();
     void qtbug_106203();
     void qtbug_104478();
+    void qtbug_135653_connected();
+    void qtbug_135653_not_connected();
+    void qtbug_135653_not_unsubscribed();
 private:
     void createAndSubscribe(QMqttClient *c, QMqttSubscription **sub, const QString &topic);
     QProcess m_brokerProcess;
@@ -482,6 +485,89 @@ void Tst_QMqttSubscription::qtbug_104478()
     QTRY_VERIFY2(client.state() == QMqttClient::Disconnected, "Could not disconnect from broker.");
 }
 
+void Tst_QMqttSubscription::qtbug_135653_connected()
+{
+    VersionClient(QMqttClient::ProtocolVersion(0), client);
+    client.setHostname(m_testBroker);
+    client.setPort(m_port);
+    client.connectToHost();
+    QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker.");
+
+    const QString topic(QLatin1String("test/foo/bar/test3"));
+
+    QMqttSubscription *subscriptions[10] = { 0 };
+    for (int i = 0; i < 10; ++i) {
+        bool received = false;
+        subscriptions[i] = client.subscribe(topic);
+        connect(subscriptions[i], &QMqttSubscription::messageReceived, subscriptions[i], [&client, topic, &received](QMqttMessage msg) {
+            received = (msg.payload() == "Hello World");
+        });
+        client.publish(topic, QString(QLatin1String("Hello World")).toLocal8Bit(), 1, true);
+        QTRY_VERIFY2(received, "Could not receive message.");
+        client.unsubscribe(topic);
+        client.unsubscribe(topic);
+    }
+
+    for (int i = 0; i < 10; ++i) {
+        QVERIFY(subscriptions[i]);
+        for (int j = i + 1; j < 10; ++j)
+            QVERIFY(subscriptions[i] != subscriptions[j]);
+    }
+}
+
+void Tst_QMqttSubscription::qtbug_135653_not_connected()
+{
+    VersionClient(QMqttClient::ProtocolVersion(0), client);
+    client.setHostname(m_testBroker);
+    client.setPort(m_port);
+
+    const QString topic(QLatin1String("test/foo/bar/test4"));
+
+    QMqttSubscription *subscriptions[10] = { 0 };
+    for (int i = 0; i < 10; ++i) {
+        bool received = false;
+        client.connectToHost();
+        QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker.");
+        subscriptions[i] = client.subscribe(topic);
+        connect(subscriptions[i], &QMqttSubscription::messageReceived, subscriptions[i], [&client, topic, &received](QMqttMessage msg) {
+            received = (msg.payload() == "Hello World");
+        });
+        client.publish(topic, QString(QLatin1String("Hello World")).toLocal8Bit(), 1, true);
+        QTRY_VERIFY2(received, "Could not receive message.");
+        client.disconnectFromHost();
+        QTRY_VERIFY2(client.state() == QMqttClient::Disconnected, "Could not disconnect from broker.");
+        client.unsubscribe(topic);
+        client.unsubscribe(topic);
+    }
+    for (int i = 0; i < 10; ++i) {
+        QVERIFY(subscriptions[i]);
+        for (int j = i + 1; j < 10; ++j)
+            QVERIFY(subscriptions[i] != subscriptions[j]);
+    }
+}
+
+void Tst_QMqttSubscription::qtbug_135653_not_unsubscribed()
+{
+
+    const QString topic(QLatin1String("test/foo/bar/test5"));
+
+    QMqttSubscription *subscriptions[10] = { 0 };
+    for (int i = 0; i < 10; ++i) {
+        bool received = false;
+        VersionClient(QMqttClient::ProtocolVersion(0), client);
+        client.setHostname(m_testBroker);
+        client.setPort(m_port);
+        client.connectToHost();
+        QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker.");
+        subscriptions[i] = client.subscribe(topic);
+        connect(subscriptions[i], &QMqttSubscription::messageReceived, subscriptions[i], [&client, topic, &received](QMqttMessage msg) {
+            received = (msg.payload() == "Hello World");
+        });
+        client.publish(topic, QString(QLatin1String("Hello World")).toLocal8Bit(), 1, true);
+        QTRY_VERIFY2(received, "Could not receive message.");
+    }
+}
+
 QTEST_MAIN(Tst_QMqttSubscription)
 
 #include "tst_qmqttsubscription.moc"