From db00682c75c4a9a3fb7333deafbeb5032306a217 Mon Sep 17 00:00:00 2001 From: Christian Date: Sat, 2 Oct 2021 20:44:12 +0200 Subject: [PATCH] Improve topic comparer. --- Build/MQTTnet.nuspec | 2 + .../Internal/MqttClientSessionsManager.cs | 2 - .../MqttClientSubscriptionsManager.cs | 102 ++++--- .../Internal/MqttTopicFilterCompareResult.cs | 13 + .../Internal/MqttTopicFilterComparer.cs | 168 +++++++---- .../MQTTnet/Server/Internal/Subscription.cs | 2 +- .../MqttSubscriptionInterceptorContext.cs | 39 ++- .../TopicFilterComparerBenchmark.cs | 14 +- Tests/MQTTnet.Core.Tests/Server/General.cs | 243 +-------------- .../Server/Shared_Subscriptions_Tests.cs | 19 ++ .../Server/Subscribe_Tests.cs | 282 ++++++++++++++++++ .../Server/TopicFilterComparer_Tests.cs | 108 ++++--- Tests/MQTTnet.TestApp.NetCore/ServerTest.cs | 2 +- 13 files changed, 598 insertions(+), 398 deletions(-) create mode 100644 Source/MQTTnet/Server/Internal/MqttTopicFilterCompareResult.cs create mode 100644 Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 6893d807..159418ea 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -29,6 +29,8 @@ * [Server] Processing all pending messages before stopping (thanks to @AblEdge, #1234). * [Server] Added support for a custom exception handler in _MqttServerMultiThreadedApplicationMessageInterceptorDelegate_. * [Server] Removed logger from _MqttServerMultiThreadedApplicationMessageInterceptorDelegate_ (BREAKING CHANGE!). +* [Server] Added support for returning individual subscription errors (#80 thanks to @jimch) +* [Server] Improved topic filter comparisons (support for $). * [MQTTnet.Server] Moved server project to a dedicated GitHub repository. Git commit: $gitCommit diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 33627e9b..5743f693 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -234,8 +234,6 @@ namespace MQTTnet.Server.Internal if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - if (topicFilters is null) throw new ArgumentNullException(nameof(topicFilters)); - var fakeSubscribePacket = new MqttSubscribePacket { TopicFilters = topicFilters.ToList() diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index b6c1c393..0dd5d2ac 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -44,22 +44,22 @@ namespace MQTTnet.Server.Internal foreach (var originalTopicFilter in subscribePacket.TopicFilters.OrderByDescending(f => f.QualityOfServiceLevel)) { var interceptorContext = await InterceptSubscribe(originalTopicFilter).ConfigureAwait(false); - var finalTopicFilter = interceptorContext?.TopicFilter ?? originalTopicFilter; - var acceptSubscription = interceptorContext?.AcceptSubscription ?? true; - var closeConnection = interceptorContext?.CloseConnection ?? false; - + var finalTopicFilter = interceptorContext.TopicFilter; + var acceptSubscription = interceptorContext.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; + if (string.IsNullOrEmpty(finalTopicFilter.Topic) || !acceptSubscription) { + // Return codes is for MQTT < 5 and Reason Code is for MQTT > 5. result.ReturnCodes.Add(MqttSubscribeReturnCode.Failure); - result.ReasonCodes.Add(MqttSubscribeReasonCode.UnspecifiedError); + result.ReasonCodes.Add(interceptorContext.ReasonCode); } else { - result.ReturnCodes.Add(ConvertToSubscribeReturnCode(finalTopicFilter.QualityOfServiceLevel)); - result.ReasonCodes.Add(ConvertToSubscribeReasonCode(finalTopicFilter.QualityOfServiceLevel)); + result.ReturnCodes.Add(interceptorContext.ReturnCode); + result.ReasonCodes.Add(interceptorContext.ReasonCode); } - if (closeConnection) + if (interceptorContext.CloseConnection) { result.CloseConnection = true; } @@ -69,7 +69,10 @@ namespace MQTTnet.Server.Internal continue; } - var subscription = CreateSubscription(finalTopicFilter, subscribePacket); + var subscription = CreateSubscription( + finalTopicFilter, + subscribePacket.Properties?.SubscriptionIdentifier ?? 0, + interceptorContext.ReasonCode); await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false); @@ -148,12 +151,12 @@ namespace MQTTnet.Server.Internal retainAsPublished = true; } - if (!MqttTopicFilterComparer.IsMatch(topic, subscription.Topic)) + if (MqttTopicFilterComparer.Compare(topic, subscription.Topic) != MqttTopicFilterCompareResult.IsMatch) { continue; } - qosLevels.Add(subscription.QualityOfServiceLevel); + qosLevels.Add(subscription.GrantedQualityOfServiceLevel); if (subscription.Identifier > 0) { @@ -175,7 +178,7 @@ namespace MQTTnet.Server.Internal }; } - Subscription CreateSubscription(MqttTopicFilter topicFilter, MqttSubscribePacket subscribePacket) + Subscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier, MqttSubscribeReasonCode reasonCode) { var subscription = new Subscription { @@ -183,10 +186,26 @@ namespace MQTTnet.Server.Internal NoLocal = topicFilter.NoLocal, RetainHandling = topicFilter.RetainHandling, RetainAsPublished = topicFilter.RetainAsPublished, - QualityOfServiceLevel = topicFilter.QualityOfServiceLevel, - Identifier = subscribePacket.Properties?.SubscriptionIdentifier ?? 0 + Identifier = subscriptionIdentifier }; + if (reasonCode == MqttSubscribeReasonCode.GrantedQoS0) + { + subscription.GrantedQualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce; + } + else if (reasonCode == MqttSubscribeReasonCode.GrantedQoS1) + { + subscription.GrantedQualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce; + } + else if (reasonCode == MqttSubscribeReasonCode.GrantedQoS2) + { + subscription.GrantedQualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce; + } + else + { + throw new InvalidOperationException(); + } + _subscriptionsLock.EnterWriteLock(); try { @@ -223,7 +242,7 @@ namespace MQTTnet.Server.Internal continue; } - if (!MqttTopicFilterComparer.IsMatch(retainedApplicationMessage.Topic, subscription.Topic)) + if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, subscription.Topic) != MqttTopicFilterCompareResult.IsMatch) { continue; } @@ -232,7 +251,7 @@ namespace MQTTnet.Server.Internal { ApplicationMessage = retainedApplicationMessage, IsRetainedMessage = true, - SubscriptionQualityOfServiceLevel = subscription.QualityOfServiceLevel + SubscriptionQualityOfServiceLevel = subscription.GrantedQualityOfServiceLevel }; if (subscription.Identifier > 0) @@ -245,25 +264,9 @@ namespace MQTTnet.Server.Internal retainedApplicationMessages[i] = null; } } - - static MqttSubscribeReturnCode ConvertToSubscribeReturnCode(MqttQualityOfServiceLevel qualityOfServiceLevel) - { - return (MqttSubscribeReturnCode)(int)qualityOfServiceLevel; - } - - static MqttSubscribeReasonCode ConvertToSubscribeReasonCode(MqttQualityOfServiceLevel qualityOfServiceLevel) - { - return (MqttSubscribeReasonCode)(int)qualityOfServiceLevel; - } - + async Task InterceptSubscribe(MqttTopicFilter topicFilter) { - var interceptor = _options.SubscriptionInterceptor; - if (interceptor == null) - { - return null; - } - var context = new MqttSubscriptionInterceptorContext { ClientId = _clientSession.ClientId, @@ -271,8 +274,39 @@ namespace MQTTnet.Server.Internal SessionItems = _clientSession.Items }; - await interceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false); + if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) + { + context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS0; + context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS0; + } + else if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) + { + context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS1; + context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS1; + } + else if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) + { + context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS2; + context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS2; + } + context.DefaultReturnCode = context.ReturnCode; + context.DefaultReasonCode = context.ReasonCode; + + if (topicFilter.Topic.StartsWith("$share/")) + { + context.ReasonCode = MqttSubscribeReasonCode.SharedSubscriptionsNotSupported; + context.ReturnCode = MqttSubscribeReturnCode.Failure; + } + else + { + var interceptor = _options.SubscriptionInterceptor; + if (interceptor != null) + { + await interceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false); + } + } + return context; } diff --git a/Source/MQTTnet/Server/Internal/MqttTopicFilterCompareResult.cs b/Source/MQTTnet/Server/Internal/MqttTopicFilterCompareResult.cs new file mode 100644 index 00000000..118aa82b --- /dev/null +++ b/Source/MQTTnet/Server/Internal/MqttTopicFilterCompareResult.cs @@ -0,0 +1,13 @@ +namespace MQTTnet.Server.Internal +{ + public enum MqttTopicFilterCompareResult + { + NoMatch, + + IsMatch, + + FilterInvalid, + + TopicInvalid + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttTopicFilterComparer.cs b/Source/MQTTnet/Server/Internal/MqttTopicFilterComparer.cs index 1fe976e9..0c2ebff3 100644 --- a/Source/MQTTnet/Server/Internal/MqttTopicFilterComparer.cs +++ b/Source/MQTTnet/Server/Internal/MqttTopicFilterComparer.cs @@ -1,128 +1,164 @@ -using System; - -namespace MQTTnet.Server.Internal +namespace MQTTnet.Server.Internal { public static class MqttTopicFilterComparer { const char LevelSeparator = '/'; const char MultiLevelWildcard = '#'; const char SingleLevelWildcard = '+'; + const char ReservedTopicPrefix = '$'; - public static bool IsMatch(string topic, string filter) + public static MqttTopicFilterCompareResult Compare(string topic, string filter) { - if (topic == null) throw new ArgumentNullException(nameof(topic)); - if (filter == null) throw new ArgumentNullException(nameof(filter)); - - var sPos = 0; - var sLen = filter.Length; - var tPos = 0; - var tLen = topic.Length; - - while (sPos < sLen && tPos < tLen) + if (string.IsNullOrEmpty(topic)) { - if (filter[sPos] == topic[tPos]) + return MqttTopicFilterCompareResult.TopicInvalid; + } + + if (string.IsNullOrEmpty(filter)) + { + return MqttTopicFilterCompareResult.FilterInvalid; + } + + var filterOffset = 0; + var filterLength = filter.Length; + + var topicOffset = 0; + var topicLength = topic.Length; + + var isMultiLevelFilter = filter[filterLength - 1] == MultiLevelWildcard; + var isReservedTopic = topic[0] == ReservedTopicPrefix; + + if (isReservedTopic && filterLength == 1 && isMultiLevelFilter) + { + // It is not allowed to receive i.e. '$foo/bar' with filter '#'. + return MqttTopicFilterCompareResult.NoMatch; + } + + if (isReservedTopic && filter[0] == SingleLevelWildcard) + { + // It is not allowed to receive i.e. '$SYS/monitor/Clients' with filter '+/monitor/Clients'. + return MqttTopicFilterCompareResult.NoMatch; + } + + if (filterLength == 1 && isMultiLevelFilter) + { + // Filter '#' matches basically everything. + return MqttTopicFilterCompareResult.IsMatch; + } + + // Go through the filter char by char. + while (filterOffset < filterLength && topicOffset < topicLength) + { + // Check if the current char is a multi level wildcard. The char is only allowed + // at the very las position. + if (filter[filterOffset] == MultiLevelWildcard && filterOffset != filterLength - 1) { - if (tPos == tLen - 1) + return MqttTopicFilterCompareResult.FilterInvalid; + } + + if (filter[filterOffset] == topic[topicOffset]) + { + if (topicOffset == topicLength - 1) { - // Check for e.g. foo matching foo/# - if (sPos == sLen - 3 - && filter[sPos + 1] == LevelSeparator - && filter[sPos + 2] == MultiLevelWildcard) + // Check for e.g. "foo" matching "foo/#" + if (filterOffset == filterLength - 3 && + filter[filterOffset + 1] == LevelSeparator && + isMultiLevelFilter) { - return true; + return MqttTopicFilterCompareResult.IsMatch; } - // Check for e.g. foo/ matching foo/# - if (sPos == sLen - 2 - && filter[sPos] == LevelSeparator - && filter[sPos + 1] == MultiLevelWildcard) + + // Check for e.g. "foo/" matching "foo/#" + if (filterOffset == filterLength - 2 && + filter[filterOffset] == LevelSeparator && + isMultiLevelFilter) { - return true; + return MqttTopicFilterCompareResult.IsMatch; } } - sPos++; - tPos++; + filterOffset++; + topicOffset++; - if (sPos == sLen && tPos == tLen) + // Check if the end was reached and i.e. "foo/bar" matches "foo/bar" + if (filterOffset == filterLength && topicOffset == topicLength) { - return true; + return MqttTopicFilterCompareResult.IsMatch; } - if (tPos == tLen && sPos == sLen - 1 && filter[sPos] == SingleLevelWildcard) + var endOfTopic = topicOffset == topicLength; + + if (endOfTopic && filterOffset == filterLength - 1 && filter[filterOffset] == SingleLevelWildcard) { - if (sPos > 0 && filter[sPos - 1] != LevelSeparator) + if (filterOffset > 0 && filter[filterOffset - 1] != LevelSeparator) { - // Invalid filter string - return false; + // Invalid filter. + return MqttTopicFilterCompareResult.FilterInvalid; } - return true; + return MqttTopicFilterCompareResult.IsMatch; } } else { - if (filter[sPos] == SingleLevelWildcard) + if (filter[filterOffset] == SingleLevelWildcard) { - // Check for bad "+foo" or "a/+foo" subscription - if (sPos > 0 && filter[sPos - 1] != LevelSeparator) + // Check for invalid "+foo" or "a/+foo" subscription + if (filterOffset > 0 && filter[filterOffset - 1] != LevelSeparator) { - // Invalid filter string - return false; + return MqttTopicFilterCompareResult.FilterInvalid; } // Check for bad "foo+" or "foo+/a" subscription - if (sPos < sLen - 1 && filter[sPos + 1] != LevelSeparator) + if (filterOffset < filterLength - 1 && filter[filterOffset + 1] != LevelSeparator) { - // Invalid filter string - return false; + return MqttTopicFilterCompareResult.FilterInvalid; } - sPos++; - while (tPos < tLen && topic[tPos] != LevelSeparator) + filterOffset++; + while (topicOffset < topicLength && topic[topicOffset] != LevelSeparator) { - tPos++; + topicOffset++; } - if (tPos == tLen && sPos == sLen) + if (topicOffset == topicLength && filterOffset == filterLength) { - return true; + return MqttTopicFilterCompareResult.IsMatch; } } - else if (filter[sPos] == MultiLevelWildcard) + else if (filter[filterOffset] == MultiLevelWildcard) { - if (sPos > 0 && filter[sPos - 1] != LevelSeparator) + if (filterOffset > 0 && filter[filterOffset - 1] != LevelSeparator) { - // Invalid filter string - return false; + return MqttTopicFilterCompareResult.FilterInvalid; } - if (sPos + 1 != sLen) + if (filterOffset + 1 != filterLength) { - // Invalid filter string - return false; + return MqttTopicFilterCompareResult.FilterInvalid; } - return true; + return MqttTopicFilterCompareResult.IsMatch; } else { - // Check for e.g. foo/bar matching foo/+/# - if (sPos > 0 - && sPos + 2 == sLen - && tPos == tLen - && filter[sPos - 1] == SingleLevelWildcard - && filter[sPos] == LevelSeparator - && filter[sPos + 1] == MultiLevelWildcard) + // Check for e.g. "foo/bar" matching "foo/+/#". + if (filterOffset > 0 && + filterOffset + 2 == filterLength && + topicOffset == topicLength && + filter[filterOffset - 1] == SingleLevelWildcard && + filter[filterOffset] == LevelSeparator && + isMultiLevelFilter) { - return true; + return MqttTopicFilterCompareResult.IsMatch; } - return false; + return MqttTopicFilterCompareResult.NoMatch; } } } - return false; + return MqttTopicFilterCompareResult.NoMatch; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/Subscription.cs b/Source/MQTTnet/Server/Internal/Subscription.cs index 965cce67..9dcd4657 100644 --- a/Source/MQTTnet/Server/Internal/Subscription.cs +++ b/Source/MQTTnet/Server/Internal/Subscription.cs @@ -12,7 +12,7 @@ namespace MQTTnet.Server.Internal public bool RetainAsPublished { get; set; } - public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; } + public MqttQualityOfServiceLevel GrantedQualityOfServiceLevel { get; set; } public uint Identifier { get; set; } diff --git a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs index 025a6504..28b8deb5 100644 --- a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs @@ -1,4 +1,6 @@ -using System.Collections.Generic; +using System; +using System.Collections.Generic; +using MQTTnet.Protocol; namespace MQTTnet.Server { @@ -21,8 +23,39 @@ namespace MQTTnet.Server /// public IDictionary SessionItems { get; internal set; } - public bool AcceptSubscription { get; set; } = true; + // Will be removed together with "AcceptSubscription". It only stores the default value when setting "AcceptSubscription" to true. + internal MqttSubscribeReasonCode DefaultReasonCode { get; set; } + internal MqttSubscribeReturnCode DefaultReturnCode { get; set; } + + [Obsolete("Please use a proper value for 'ReasonCode' instead. This property will be removed in the future.")] + public bool AcceptSubscription + { + get => ReasonCode >= MqttSubscribeReasonCode.GrantedQoS0 && ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2; + set + { + if (value) + { + ReasonCode = DefaultReasonCode; + ReturnCode = DefaultReturnCode; + } + else + { + ReturnCode = MqttSubscribeReturnCode.Failure; + ReasonCode = MqttSubscribeReasonCode.UnspecifiedError; + } + } + } + + /// + /// Gets or sets the reason code which is sent to the client. + /// The subscription is skipped when the value is not GrantedQoS_. + /// MQTTv5 only. + /// + public MqttSubscribeReasonCode ReasonCode { get; set; } + + // MQTT < 5 only! + internal MqttSubscribeReturnCode ReturnCode { get; set; } public bool CloseConnection { get; set; } } -} +} \ No newline at end of file diff --git a/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs b/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs index e6a09aa4..02924f51 100644 --- a/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs @@ -37,13 +37,13 @@ namespace MQTTnet.Benchmarks { for (var i = 0; i < 10000; i++) { - MqttTopicFilterComparer.IsMatch("sport/tennis/player1", "sport/#"); - MqttTopicFilterComparer.IsMatch("sport/tennis/player1/ranking", "sport/#/ranking"); - MqttTopicFilterComparer.IsMatch("sport/tennis/player1/score/wimbledon", "sport/+/player1/#"); - MqttTopicFilterComparer.IsMatch("sport/tennis/player1", "sport/tennis/+"); - MqttTopicFilterComparer.IsMatch("/finance", "+/+"); - MqttTopicFilterComparer.IsMatch("/finance", "/+"); - MqttTopicFilterComparer.IsMatch("/finance", "+"); + MqttTopicFilterComparer.Compare("sport/tennis/player1", "sport/#"); + MqttTopicFilterComparer.Compare("sport/tennis/player1/ranking", "sport/#/ranking"); + MqttTopicFilterComparer.Compare("sport/tennis/player1/score/wimbledon", "sport/+/player1/#"); + MqttTopicFilterComparer.Compare("sport/tennis/player1", "sport/tennis/+"); + MqttTopicFilterComparer.Compare("/finance", "+/+"); + MqttTopicFilterComparer.Compare("/finance", "/+"); + MqttTopicFilterComparer.Compare("/finance", "+"); } } diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index f39b5215..6c190a8f 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -202,248 +202,7 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(1, receivedMessagesCount); } } - - [TestMethod] - public async Task Intercept_Subscription() - { - using (var testEnvironment = new TestEnvironment(TestContext)) - { - await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithSubscriptionInterceptor( - c => - { - // Set the topic to "a" regards what the client wants to subscribe. - c.TopicFilter.Topic = "a"; - })); - - var topicAReceived = false; - var topicBReceived = false; - - var client = await testEnvironment.ConnectClient(); - client.UseApplicationMessageReceivedHandler(c => - { - if (c.ApplicationMessage.Topic == "a") - { - topicAReceived = true; - } - else if (c.ApplicationMessage.Topic == "b") - { - topicBReceived = true; - } - }); - - await client.SubscribeAsync("b"); - - await client.PublishAsync("a"); - - await Task.Delay(500); - - Assert.IsTrue(topicAReceived); - Assert.IsFalse(topicBReceived); - } - } - - [TestMethod] - public async Task Subscribe_Unsubscribe() - { - using (var testEnvironment = new TestEnvironment(TestContext)) - { - var receivedMessagesCount = 0; - - var server = await testEnvironment.StartServer(); - - var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("c1")); - c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); - - var c2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("c2")); - - var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); - await c2.PublishAsync(message); - - await Task.Delay(500); - Assert.AreEqual(0, receivedMessagesCount); - - var subscribeEventCalled = false; - server.ClientSubscribedTopicHandler = new MqttServerClientSubscribedTopicHandlerDelegate(e => - { - subscribeEventCalled = e.TopicFilter.Topic == "a" && e.ClientId == c1.Options.ClientId; - }); - - await c1.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); - await Task.Delay(250); - Assert.IsTrue(subscribeEventCalled, "Subscribe event not called."); - - await c2.PublishAsync(message); - await Task.Delay(250); - Assert.AreEqual(1, receivedMessagesCount); - - var unsubscribeEventCalled = false; - server.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(e => - { - unsubscribeEventCalled = e.TopicFilter == "a" && e.ClientId == c1.Options.ClientId; - }); - - await c1.UnsubscribeAsync("a"); - await Task.Delay(250); - Assert.IsTrue(unsubscribeEventCalled, "Unsubscribe event not called."); - - await c2.PublishAsync(message); - await Task.Delay(500); - Assert.AreEqual(1, receivedMessagesCount); - - await Task.Delay(500); - - Assert.AreEqual(1, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Subscribe_Multiple_In_Single_Request() - { - using (var testEnvironment = new TestEnvironment(TestContext)) - { - var receivedMessagesCount = 0; - - await testEnvironment.StartServer(); - - var c1 = await testEnvironment.ConnectClient(); - c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); - await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() - .WithTopicFilter("a") - .WithTopicFilter("b") - .WithTopicFilter("c") - .Build()); - - var c2 = await testEnvironment.ConnectClient(); - - await c2.PublishAsync("a"); - await Task.Delay(100); - Assert.AreEqual(receivedMessagesCount, 1); - - await c2.PublishAsync("b"); - await Task.Delay(100); - Assert.AreEqual(receivedMessagesCount, 2); - - await c2.PublishAsync("c"); - await Task.Delay(100); - Assert.AreEqual(receivedMessagesCount, 3); - } - } - - [TestMethod] - public async Task Subscribe_Lots_In_Single_Request() - { - using (var testEnvironment = new TestEnvironment(TestContext)) - { - var receivedMessagesCount = 0; - - await testEnvironment.StartServer(); - - var c1 = await testEnvironment.ConnectClient(); - c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); - - var optionsBuilder = new MqttClientSubscribeOptionsBuilder(); - for (var i = 0; i < 500; i++) - { - optionsBuilder.WithTopicFilter(i.ToString(), MqttQualityOfServiceLevel.AtMostOnce); - } - - await c1.SubscribeAsync(optionsBuilder.Build()).ConfigureAwait(false); - - var c2 = await testEnvironment.ConnectClient(); - - var messageBuilder = new MqttApplicationMessageBuilder(); - for (var i = 0; i < 500; i++) - { - messageBuilder.WithTopic(i.ToString()); - - await c2.PublishAsync(messageBuilder.Build()).ConfigureAwait(false); - } - - SpinWait.SpinUntil(() => receivedMessagesCount == 500, TimeSpan.FromSeconds(20)); - - Assert.AreEqual(500, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Subscribe_Lots_In_Multiple_Requests() - { - using (var testEnvironment = new TestEnvironment(TestContext)) - { - var receivedMessagesCount = 0; - - await testEnvironment.StartServer(); - - var c1 = await testEnvironment.ConnectClient(); - c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); - - for (var i = 0; i < 500; i++) - { - var so = new MqttClientSubscribeOptionsBuilder() - .WithTopicFilter(i.ToString()).Build(); - - await c1.SubscribeAsync(so).ConfigureAwait(false); - - await Task.Delay(10); - } - - var c2 = await testEnvironment.ConnectClient(); - - var messageBuilder = new MqttApplicationMessageBuilder(); - for (var i = 0; i < 500; i++) - { - messageBuilder.WithTopic(i.ToString()); - - await c2.PublishAsync(messageBuilder.Build()).ConfigureAwait(false); - - await Task.Delay(10); - } - - SpinWait.SpinUntil(() => receivedMessagesCount == 500, 5000); - - Assert.AreEqual(500, receivedMessagesCount); - } - } - - [TestMethod] - public async Task Subscribe_Multiple_In_Multiple_Request() - { - using (var testEnvironment = new TestEnvironment(TestContext)) - { - var receivedMessagesCount = 0; - - await testEnvironment.StartServer(); - - var c1 = await testEnvironment.ConnectClient(); - c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); - await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() - .WithTopicFilter("a") - .Build()); - - await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() - .WithTopicFilter("b") - .Build()); - - await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() - .WithTopicFilter("c") - .Build()); - - var c2 = await testEnvironment.ConnectClient(); - - await c2.PublishAsync("a"); - await Task.Delay(100); - Assert.AreEqual(receivedMessagesCount, 1); - - await c2.PublishAsync("b"); - await Task.Delay(100); - Assert.AreEqual(receivedMessagesCount, 2); - - await c2.PublishAsync("c"); - await Task.Delay(100); - Assert.AreEqual(receivedMessagesCount, 3); - } - } - + [TestMethod] public async Task Publish_From_Server() { diff --git a/Tests/MQTTnet.Core.Tests/Server/Shared_Subscriptions_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Shared_Subscriptions_Tests.cs index d510e7a8..0c615531 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Shared_Subscriptions_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Shared_Subscriptions_Tests.cs @@ -1,6 +1,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; +using MQTTnet.Client.Subscribing; using MQTTnet.Formatter; namespace MQTTnet.Tests.Server @@ -23,5 +24,23 @@ namespace MQTTnet.Tests.Server Assert.IsFalse(connectResult.SharedSubscriptionAvailable); } } + + [TestMethod] + public async Task Subscription_Of_Shared_Subscription_Is_Denied() + { + using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500)) + { + await testEnvironment.StartServer(); + + var client = testEnvironment.CreateClient(); + await client.ConnectAsync(testEnvironment.Factory.CreateClientOptionsBuilder() + .WithProtocolVersion(MqttProtocolVersion.V500) + .WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build()); + + var subscribeResult = await client.SubscribeAsync("$share/A"); + + Assert.AreEqual(MqttClientSubscribeResultCode.SharedSubscriptionsNotSupported, subscribeResult.Items[0].ResultCode); + } + } } } \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs new file mode 100644 index 00000000..a6d035cc --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs @@ -0,0 +1,282 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Client.Options; +using MQTTnet.Client.Subscribing; +using MQTTnet.Formatter; +using MQTTnet.Protocol; +using MQTTnet.Server; + +namespace MQTTnet.Tests.Server +{ + [TestClass] + public sealed class Subscribe_Tests : BaseTestClass + { + [TestMethod] + public async Task Intercept_Subscription() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithSubscriptionInterceptor( + c => + { + // Set the topic to "a" regards what the client wants to subscribe. + c.TopicFilter.Topic = "a"; + })); + + var topicAReceived = false; + var topicBReceived = false; + + var client = await testEnvironment.ConnectClient(); + client.UseApplicationMessageReceivedHandler(c => + { + if (c.ApplicationMessage.Topic == "a") + { + topicAReceived = true; + } + else if (c.ApplicationMessage.Topic == "b") + { + topicBReceived = true; + } + }); + + await client.SubscribeAsync("b"); + + await client.PublishAsync("a"); + + await Task.Delay(500); + + Assert.IsTrue(topicAReceived); + Assert.IsFalse(topicBReceived); + } + } + + [TestMethod] + public async Task Subscribe_Unsubscribe() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + var server = await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("c1")); + c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); + + var c2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("c2")); + + var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); + await c2.PublishAsync(message); + + await Task.Delay(500); + Assert.AreEqual(0, receivedMessagesCount); + + var subscribeEventCalled = false; + server.ClientSubscribedTopicHandler = new MqttServerClientSubscribedTopicHandlerDelegate(e => + { + subscribeEventCalled = e.TopicFilter.Topic == "a" && e.ClientId == c1.Options.ClientId; + }); + + await c1.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); + await Task.Delay(250); + Assert.IsTrue(subscribeEventCalled, "Subscribe event not called."); + + await c2.PublishAsync(message); + await Task.Delay(250); + Assert.AreEqual(1, receivedMessagesCount); + + var unsubscribeEventCalled = false; + server.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(e => + { + unsubscribeEventCalled = e.TopicFilter == "a" && e.ClientId == c1.Options.ClientId; + }); + + await c1.UnsubscribeAsync("a"); + await Task.Delay(250); + Assert.IsTrue(unsubscribeEventCalled, "Unsubscribe event not called."); + + await c2.PublishAsync(message); + await Task.Delay(500); + Assert.AreEqual(1, receivedMessagesCount); + + await Task.Delay(500); + + Assert.AreEqual(1, receivedMessagesCount); + } + } + + [TestMethod] + public async Task Subscribe_Multiple_In_Single_Request() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(); + c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); + await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() + .WithTopicFilter("a") + .WithTopicFilter("b") + .WithTopicFilter("c") + .Build()); + + var c2 = await testEnvironment.ConnectClient(); + + await c2.PublishAsync("a"); + await Task.Delay(100); + Assert.AreEqual(receivedMessagesCount, 1); + + await c2.PublishAsync("b"); + await Task.Delay(100); + Assert.AreEqual(receivedMessagesCount, 2); + + await c2.PublishAsync("c"); + await Task.Delay(100); + Assert.AreEqual(receivedMessagesCount, 3); + } + } + + [TestMethod] + public async Task Subscribe_Lots_In_Single_Request() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(); + c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); + + var optionsBuilder = new MqttClientSubscribeOptionsBuilder(); + for (var i = 0; i < 500; i++) + { + optionsBuilder.WithTopicFilter(i.ToString()); + } + + await c1.SubscribeAsync(optionsBuilder.Build()).ConfigureAwait(false); + + var c2 = await testEnvironment.ConnectClient(); + + var messageBuilder = new MqttApplicationMessageBuilder(); + for (var i = 0; i < 500; i++) + { + messageBuilder.WithTopic(i.ToString()); + + await c2.PublishAsync(messageBuilder.Build()).ConfigureAwait(false); + } + + SpinWait.SpinUntil(() => receivedMessagesCount == 500, TimeSpan.FromSeconds(20)); + + Assert.AreEqual(500, receivedMessagesCount); + } + } + + [TestMethod] + public async Task Subscribe_Lots_In_Multiple_Requests() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(); + c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); + + for (var i = 0; i < 500; i++) + { + var so = new MqttClientSubscribeOptionsBuilder() + .WithTopicFilter(i.ToString()).Build(); + + await c1.SubscribeAsync(so).ConfigureAwait(false); + + await Task.Delay(10); + } + + var c2 = await testEnvironment.ConnectClient(); + + var messageBuilder = new MqttApplicationMessageBuilder(); + for (var i = 0; i < 500; i++) + { + messageBuilder.WithTopic(i.ToString()); + + await c2.PublishAsync(messageBuilder.Build()).ConfigureAwait(false); + + await Task.Delay(10); + } + + SpinWait.SpinUntil(() => receivedMessagesCount == 500, 5000); + + Assert.AreEqual(500, receivedMessagesCount); + } + } + + [TestMethod] + public async Task Subscribe_Multiple_In_Multiple_Request() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var receivedMessagesCount = 0; + + await testEnvironment.StartServer(); + + var c1 = await testEnvironment.ConnectClient(); + c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); + await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() + .WithTopicFilter("a") + .Build()); + + await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() + .WithTopicFilter("b") + .Build()); + + await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() + .WithTopicFilter("c") + .Build()); + + var c2 = await testEnvironment.ConnectClient(); + + await c2.PublishAsync("a"); + await Task.Delay(100); + Assert.AreEqual(receivedMessagesCount, 1); + + await c2.PublishAsync("b"); + await Task.Delay(100); + Assert.AreEqual(receivedMessagesCount, 2); + + await c2.PublishAsync("c"); + await Task.Delay(100); + Assert.AreEqual(receivedMessagesCount, 3); + } + } + + [TestMethod] + public async Task Deny_Invalid_Topic() + { + using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500)) + { + await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithSubscriptionInterceptor( + c => + { + if (c.TopicFilter.Topic == "not_allowed_topic") + { + c.ReasonCode = MqttSubscribeReasonCode.TopicFilterInvalid; + } + })); + + var client = await testEnvironment.ConnectClient(); + + var subscribeResult =await client.SubscribeAsync("allowed_topic"); + Assert.AreEqual(MqttClientSubscribeResultCode.GrantedQoS0, subscribeResult.Items[0].ResultCode); + + subscribeResult =await client.SubscribeAsync("not_allowed_topic"); + Assert.AreEqual(MqttClientSubscribeResultCode.TopicFilterInvalid, subscribeResult.Items[0].ResultCode); + } + } + } +} \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/Server/TopicFilterComparer_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/TopicFilterComparer_Tests.cs index e885ff02..93678df6 100644 --- a/Tests/MQTTnet.Core.Tests/Server/TopicFilterComparer_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/TopicFilterComparer_Tests.cs @@ -4,111 +4,135 @@ using MQTTnet.Server.Internal; namespace MQTTnet.Tests.Server { [TestClass] - public class TopicFilterComparer_Tests + public sealed class MqttTopicFilterComparer_Tests { [TestMethod] - public void TopicFilterComparer_Plus_Match_With_Separator_Only() + public void Plus_Match_With_Separator_Only() { //A Topic Name or Topic Filter consisting only of the ‘/’ character is valid - CompareAndAssert("A", "+", true); + CompareAndAssert("A", "+", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_Hash_Match_With_Separator_Only() + public void Hash_Match_With_Separator_Only() { //A Topic Name or Topic Filter consisting only of the ‘/’ character is valid - CompareAndAssert("/", "#", true); + CompareAndAssert("/", "#", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_DirectMatch() + public void DirectMatch() { - CompareAndAssert("A/B/C", "A/B/C", true); + CompareAndAssert("A/B/C", "A/B/C", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_DirectNoMatch() + public void DirectNoMatch() { - CompareAndAssert("A/B/X", "A/B/C", false); + CompareAndAssert("A/B/X", "A/B/C", MqttTopicFilterCompareResult.NoMatch); } [TestMethod] - public void TopicFilterComparer_MiddleOneLevelWildcardMatch() + public void MiddleOneLevelWildcardMatch() { - CompareAndAssert("A/B/C", "A/+/C", true); + CompareAndAssert("A/B/C", "A/+/C", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_MiddleOneLevelWildcardNoMatch() + public void MiddleOneLevelWildcardNoMatch() { - CompareAndAssert("A/B/C/D", "A/+/C", false); + CompareAndAssert("A/B/C/D", "A/+/C", MqttTopicFilterCompareResult.NoMatch); } [TestMethod] - public void TopicFilterComparer_BeginningOneLevelWildcardMatch() + public void BeginningOneLevelWildcardMatch() { - CompareAndAssert("A/B/C", "+/B/C", true); + CompareAndAssert("A/B/C", "+/B/C", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_EndOneLevelWildcardMatch() + public void EndOneLevelWildcardMatch() { - CompareAndAssert("A/B/C", "A/B/+", true); + CompareAndAssert("A/B/C", "A/B/+", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_EndMultipleLevelsWildcardMatch() + public void EndMultipleLevelsWildcardMatch() { - CompareAndAssert("A/B/C", "A/#", true); + CompareAndAssert("A/B/C", "A/#", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_EndMultipleLevelsWildcardNoMatch() + public void EndMultipleLevelsWildcardNoMatch() { - CompareAndAssert("A/B/C/D", "A/C/#", false); + CompareAndAssert("A/B/C/D", "A/C/#", MqttTopicFilterCompareResult.NoMatch); } [TestMethod] - public void TopicFilterComparer_EndMultipleLevelsWildcardMatchEmptyLevel() + public void EndMultipleLevelsWildcardMatchEmptyLevel() { - CompareAndAssert("A/", "A/#", true); + CompareAndAssert("A/", "A/#", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_AllLevelsWildcardMatch() + public void AllLevelsWildcardMatch() { - CompareAndAssert("A/B/C/D", "#", true); + CompareAndAssert("A/B/C/D", "#", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_MultiLevel_Sport() + public void MultiLevel_Sport() { // Tests from official MQTT spec (4.7.1.2 Multi-level wildcard) - CompareAndAssert("sport/tennis/player1", "sport/tennis/player1/#", true); - CompareAndAssert("sport/tennis/player1/ranking", "sport/tennis/player1/#", true); - CompareAndAssert("sport/tennis/player1/score/wimbledon", "sport/tennis/player1/#", true); + CompareAndAssert("sport/tennis/player1", "sport/tennis/player1/#", MqttTopicFilterCompareResult.IsMatch); + CompareAndAssert("sport/tennis/player1/ranking", "sport/tennis/player1/#", MqttTopicFilterCompareResult.IsMatch); + CompareAndAssert("sport/tennis/player1/score/wimbledon", "sport/tennis/player1/#", MqttTopicFilterCompareResult.IsMatch); - CompareAndAssert("sport/tennis/player1", "sport/tennis/+", true); - CompareAndAssert("sport/tennis/player2", "sport/tennis/+", true); - CompareAndAssert("sport/tennis/player1/ranking", "sport/tennis/+", false); + CompareAndAssert("sport/tennis/player1", "sport/tennis/+", MqttTopicFilterCompareResult.IsMatch); + CompareAndAssert("sport/tennis/player2", "sport/tennis/+", MqttTopicFilterCompareResult.IsMatch); + CompareAndAssert("sport/tennis/player1/ranking", "sport/tennis/+", MqttTopicFilterCompareResult.NoMatch); - CompareAndAssert("sport", "sport/#", true); - CompareAndAssert("sport", "sport/+", false); - CompareAndAssert("sport/", "sport/+", true); + CompareAndAssert("sport", "sport/#", MqttTopicFilterCompareResult.IsMatch); + CompareAndAssert("sport", "sport/+", MqttTopicFilterCompareResult.NoMatch); + CompareAndAssert("sport/", "sport/+", MqttTopicFilterCompareResult.IsMatch); } [TestMethod] - public void TopicFilterComparer_SingleLevel_Finance() + public void SingleLevel_Finance() { // Tests from official MQTT spec (4.7.1.3 Single level wildcard) - CompareAndAssert("/finance", "+/+", true); - CompareAndAssert("/finance", "/+", true); - CompareAndAssert("/finance", "+", false); + CompareAndAssert("/finance", "+/+", MqttTopicFilterCompareResult.IsMatch); + CompareAndAssert("/finance", "/+", MqttTopicFilterCompareResult.IsMatch); + CompareAndAssert("/finance", "+", MqttTopicFilterCompareResult.NoMatch); } - - private static void CompareAndAssert(string topic, string filter, bool expectedResult) + + [TestMethod] + public void Reserved_Multi_Level_Wildcard_Only() { - Assert.AreEqual(expectedResult, MqttTopicFilterComparer.IsMatch(topic, filter)); + CompareAndAssert("$SPECIAL/TOPIC", "#", MqttTopicFilterCompareResult.NoMatch); + } + + [TestMethod] + public void Reserved_Single_Level_Wildcard_Prefix() + { + CompareAndAssert("$SYS/monitor/Clients", "+/monitor/Clients", MqttTopicFilterCompareResult.NoMatch); + } + + [TestMethod] + public void Reserved_Single_Level_Wildcard_Suffix() + { + CompareAndAssert("$SYS/monitor/Clients", "$SYS/monitor/+", MqttTopicFilterCompareResult.IsMatch); + } + + [TestMethod] + public void Reserved_Single_Level_Wildcard() + { + CompareAndAssert("$SYS/monitor/Clients", "$SYS/#", MqttTopicFilterCompareResult.IsMatch); + } + + static void CompareAndAssert(string topic, string filter, MqttTopicFilterCompareResult expectedResult) + { + Assert.AreEqual(expectedResult, MqttTopicFilterComparer.Compare(topic, filter)); } } } diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index 21c47b7a..96841724 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -55,7 +55,7 @@ namespace MQTTnet.TestApp.NetCore ApplicationMessageInterceptor = new MqttServerApplicationMessageInterceptorDelegate(context => { - if (MqttTopicFilterComparer.IsMatch(context.ApplicationMessage.Topic, "/myTopic/WithTimestamp/#")) + if (MqttTopicFilterComparer.Compare(context.ApplicationMessage.Topic, "/myTopic/WithTimestamp/#") == MqttTopicFilterCompareResult.IsMatch) { // Replace the payload with the timestamp. But also extending a JSON // based payload with the timestamp is a suitable use case.