From f79649d71b6544952986bac7f645482617a613ba Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Wed, 7 Dec 2022 20:59:52 +0100 Subject: [PATCH] 1612 mqttserver subscription to similar base and subtopic with wildcards fails (#1622) --- .../Server/MqttSubscriptionsManager_Tests.cs | 98 +++++++++++++++++++ .../MQTTnet.Tests/Server/Subscribe_Tests.cs | 44 +++++++++ .../MqttClientSubscriptionsManager.cs | 27 ++--- .../Internal/TopicHashMaskSubscriptions.cs | 27 +++-- 4 files changed, 178 insertions(+), 18 deletions(-) diff --git a/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs b/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs index 19a4f427..ff81c1c6 100644 --- a/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs +++ b/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs @@ -114,6 +114,104 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(result.QualityOfServiceLevel, MqttQualityOfServiceLevel.AtLeastOnce); } + [TestMethod] + public async Task MqttSubscriptionsManager_SubscribeWildcard1() + { + await SubscribeToTopic("house/+/room"); + await SubscribeToTopic("house/+/room/+"); + + CheckIsSubscribed("house/1/room"); + CheckIsSubscribed("house/1/room/bed"); + CheckIsSubscribed("house/1/room/chair"); + CheckIsSubscribed("house/2/room/bed"); + CheckIsSubscribed("house/2/room/chair"); + + CheckIsNotSubscribed("house/1/room/bed/cover"); + CheckIsNotSubscribed("house/1/study/bed"); + } + + [TestMethod] + public async Task MqttSubscriptionsManager_SubscribeWildcard2() + { + await SubscribeToTopic("house/+/room"); + await SubscribeToTopic("house/+/room/#"); + + CheckIsSubscribed("house/1/room"); + CheckIsSubscribed("house/1/room/bed"); + CheckIsSubscribed("house/2/room"); + CheckIsSubscribed("house/2/room/bed"); + CheckIsSubscribed("house/2/room/bed/cover"); + + CheckIsNotSubscribed("house/1/level"); + CheckIsNotSubscribed("house/1/level/door"); + } + + [TestMethod] + public async Task MqttSubscriptionsManager_SubscribeWildcard3() + { + await SubscribeToTopic("house/1/room"); + await SubscribeToTopic("house/1/room/+"); + + CheckIsSubscribed("house/1/room"); + CheckIsSubscribed("house/1/room/bed"); + CheckIsSubscribed("house/1/room/chair"); + + CheckIsNotSubscribed("house/2/room"); + CheckIsNotSubscribed("house/2/room/bed/cover"); + } + + [TestMethod] + public async Task MqttSubscriptionsManager_SubscribeWildcard4() + { + await SubscribeToTopic("house/1/+/+"); + + CheckIsSubscribed("house/1/room/bed"); + CheckIsSubscribed("house/1/room/chair"); + + CheckIsNotSubscribed("house/1/room"); + CheckIsNotSubscribed("house/1/room/bed/cover"); + } + + [TestMethod] + public async Task MqttSubscriptionsManager_SubscribeWildcard5() + { + await SubscribeToTopic("house/1/+/#"); + + CheckIsSubscribed("house/1/room/bed"); + CheckIsSubscribed("house/1/room/chair"); + CheckIsSubscribed("house/1/room/chair/leg"); + CheckIsSubscribed("house/1/level/window"); + CheckIsSubscribed("house/1/level/door"); + + CheckIsNotSubscribed("house/2/room/bed"); + } + + async Task SubscribeToTopic(string topic) + { + var sp = new MqttSubscribePacket + { + TopicFilters = new List + { + new MqttTopicFilter { Topic = topic, QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }, + } + }; + + await _subscriptionsManager.Subscribe(sp, CancellationToken.None); + } + + void CheckIsSubscribed(string topic) + { + var result = CheckSubscriptions(topic, MqttQualityOfServiceLevel.AtMostOnce, ""); + Assert.IsTrue(result.IsSubscribed); + Assert.AreEqual(result.QualityOfServiceLevel, MqttQualityOfServiceLevel.AtMostOnce); + } + + void CheckIsNotSubscribed(string topic) + { + var result = CheckSubscriptions(topic, MqttQualityOfServiceLevel.AtMostOnce, ""); + Assert.IsFalse(result.IsSubscribed); + } + [TestInitialize] public void TestInitialize() { diff --git a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs index d4631bc2..a4ce93b7 100644 --- a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs @@ -13,12 +13,56 @@ using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; +using MQTTnet.Tests.Helpers; namespace MQTTnet.Tests.Server { [TestClass] public sealed class Subscribe_Tests : BaseTestClass { + [DataTestMethod] + [DataRow("A", "A", true)] + [DataRow("A", "B", false)] + [DataRow("A", "#", true)] + [DataRow("A", "+", true)] + [DataRow("A/B", "A/B", true)] + [DataRow("A/B", "A/+", true)] + [DataRow("A/B", "A/#", true)] + [DataRow("A/B/C", "A/B/C", true)] + [DataRow("A/B/C", "A/+/C", true)] + [DataRow("A/B/C", "A/+/+", true)] + [DataRow("A/B/C", "A/+/#", true)] + [DataRow("A/B/C/D", "A/B/C/D", true)] + [DataRow("A/B/C/D", "A/+/C/+", true)] + [DataRow("A/B/C/D", "A/+/C/#", true)] + [DataRow("A/B/C", "A/B/+", true)] + [DataRow("A/B1/B2/C", "A/+/C", false)] + public async Task Subscription_Roundtrip(string topic, string filter, bool shouldWork) + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer().ConfigureAwait(false); + + var receiver = await testEnvironment.ConnectClient(); + await receiver.SubscribeAsync(filter).ConfigureAwait(false); + var receivedMessages = receiver.TrackReceivedMessages(); + + var sender = await testEnvironment.ConnectClient(); + await sender.PublishStringAsync(topic, "PAYLOAD").ConfigureAwait(false); + + await LongTestDelay().ConfigureAwait(false); + + if (shouldWork) + { + Assert.AreEqual(1, receivedMessages.Count, message: "The filter should work!"); + } + else + { + Assert.AreEqual(0, receivedMessages.Count, message: "The filter should not work!"); + } + } + } + [TestMethod] public async Task Deny_Invalid_Topic() { diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 6fb45371..4155331d 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -55,18 +55,21 @@ namespace MQTTnet.Server { if (_noWildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var noWildcardSubscriptions)) { - possibleSubscriptions.AddRange(noWildcardSubscriptions.ToList()); + possibleSubscriptions.AddRange(noWildcardSubscriptions); } foreach (var wcs in _wildcardSubscriptionsByTopicHash) { - var wildcardSubscriptions = wcs.Value; var subscriptionHash = wcs.Key; - var subscriptionHashMask = wildcardSubscriptions.HashMask; - - if ((topicHash & subscriptionHashMask) == subscriptionHash) + var subscriptionsByHashMask = wcs.Value.SubscriptionsByHashMask; + foreach (var shm in subscriptionsByHashMask) { - possibleSubscriptions.AddRange(wildcardSubscriptions.Subscriptions.ToList()); + var subscriptionHashMask = shm.Key; + if ((topicHash & subscriptionHashMask) == subscriptionHash) + { + var subscriptions = shm.Value; + possibleSubscriptions.AddRange(subscriptions); + } } } } @@ -265,8 +268,8 @@ namespace MQTTnet.Server { if (_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions)) { - subscriptions.Subscriptions.Remove(existingSubscription); - if (subscriptions.Subscriptions.Count == 0) + subscriptions.RemoveSubscription(existingSubscription); + if (subscriptions.SubscriptionsByHashMask.Count == 0) { _wildcardSubscriptionsByTopicHash.Remove(topicHash); } @@ -346,9 +349,9 @@ namespace MQTTnet.Server // must remove object from topic hash dictionary first if (hasWildcard) { - if (_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subs)) + if (_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions)) { - subs.Subscriptions.Remove(existingSubscription); + subscriptions.RemoveSubscription(existingSubscription); // no need to remove empty entry because we'll be adding subscription again below } } @@ -370,11 +373,11 @@ namespace MQTTnet.Server { if (!_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions)) { - subscriptions = new TopicHashMaskSubscriptions(topicHashMask); + subscriptions = new TopicHashMaskSubscriptions(); _wildcardSubscriptionsByTopicHash.Add(topicHash, subscriptions); } - subscriptions.Subscriptions.Add(subscription); + subscriptions.AddSubscription(subscription); } else { diff --git a/Source/MQTTnet/Server/Internal/TopicHashMaskSubscriptions.cs b/Source/MQTTnet/Server/Internal/TopicHashMaskSubscriptions.cs index e9a35e83..fa074995 100644 --- a/Source/MQTTnet/Server/Internal/TopicHashMaskSubscriptions.cs +++ b/Source/MQTTnet/Server/Internal/TopicHashMaskSubscriptions.cs @@ -3,17 +3,32 @@ using System.Collections.Generic; namespace MQTTnet.Server { /// - /// Helper class that stores the topic hash mask common to all contained Subscriptions for direct access. + /// Helper class that stores subscriptions by their topic hash mask. /// public sealed class TopicHashMaskSubscriptions { - public TopicHashMaskSubscriptions(ulong hashMask) + public Dictionary> SubscriptionsByHashMask { get; } = new Dictionary>(); + + public void AddSubscription(MqttSubscription subscription) { - HashMask = hashMask; + if (!SubscriptionsByHashMask.TryGetValue(subscription.TopicHashMask, out var subscriptions)) + { + subscriptions = new HashSet(); + SubscriptionsByHashMask.Add(subscription.TopicHashMask, subscriptions); + } + subscriptions.Add(subscription); } - public ulong HashMask { get; } - - public HashSet Subscriptions { get; } = new HashSet(); + public void RemoveSubscription(MqttSubscription subscription) + { + if (SubscriptionsByHashMask.TryGetValue(subscription.TopicHashMask, out var subscriptions)) + { + subscriptions.Remove(subscription); + if (subscriptions.Count == 0) + { + SubscriptionsByHashMask.Remove(subscription.TopicHashMask); + } + } + } } } \ No newline at end of file