YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-15 00:15:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

1612 mqttserver subscription to similar base and subtopic with wildcards fails (#1622)

Browse Source
This commit is contained in:
Christian
2022-12-07 20:59:52 +01:00
committed by GitHub
parent c9c27b9f01
commit f79649d71b
4 changed files with 178 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs
+98
View File
@@ -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<MqttTopicFilter>
{
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()
{
Source/MQTTnet.Tests/Server/Subscribe_Tests.cs
+44
View File
@@ -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()
{
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+15 -12
View File
@@ -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
{
Source/MQTTnet/Server/Internal/TopicHashMaskSubscriptions.cs
+21 -6
View File
@@ -3,17 +3,32 @@ using System.Collections.Generic;
namespace MQTTnet.Server
{
/// <summary>
/// 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.
/// </summary>
public sealed class TopicHashMaskSubscriptions
{
public TopicHashMaskSubscriptions(ulong hashMask)
public Dictionary<ulong, HashSet<MqttSubscription>> SubscriptionsByHashMask { get; } = new Dictionary<ulong, HashSet<MqttSubscription>>();
public void AddSubscription(MqttSubscription subscription)
{
HashMask = hashMask;
if (!SubscriptionsByHashMask.TryGetValue(subscription.TopicHashMask, out var subscriptions))
{
subscriptions = new HashSet<MqttSubscription>();
SubscriptionsByHashMask.Add(subscription.TopicHashMask, subscriptions);
}
subscriptions.Add(subscription);
}
public ulong HashMask { get; }
public HashSet<MqttSubscription> Subscriptions { get; } = new HashSet<MqttSubscription>();
public void RemoveSubscription(MqttSubscription subscription)
{
if (SubscriptionsByHashMask.TryGetValue(subscription.TopicHashMask, out var subscriptions))
{
subscriptions.Remove(subscription);
if (subscriptions.Count == 0)
{
SubscriptionsByHashMask.Remove(subscription.TopicHashMask);
}
}
}
}
}