From 5c7472838721ac81b48884314510b66b18ba579d Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun, 6 Nov 2022 13:28:41 +0100 Subject: [PATCH] Fix client session takeover --- .../Mockups/ConnectionContextMockup.cs | 26 - .../Mockups/ConnectionHandlerMockup.cs | 22 +- .../Mockups/DuplexPipeMockup.cs | 7 +- .../Mockups/LimitedMemoryPool.cs | 2 +- .../Mockups/MemoryOwner.cs | 8 +- .../MqttConnectionContextTest.cs | 115 ++- .../ReaderExtensionsTest.cs | 2 +- ...MqttPacketSerialization_V3_Binary_Tests.cs | 3 +- .../MqttPacketSerialization_V3_Tests.cs | 2 +- .../MqttPacketSerialization_V5_Tests.cs | 2 +- .../Server/MqttSubscriptionsManager_Tests.cs | 4 +- .../Server/Subscription_TopicHash_Tests.cs | 722 +++++++++--------- Source/MQTTnet.Tests/TopicGenerator.cs | 3 +- .../Formatter/MqttUnsubAckPacketFactory.cs | 8 +- Source/MQTTnet/Internal/AsyncEvent.cs | 2 +- Source/MQTTnet/Server/Internal/MqttClient.cs | 35 +- .../Internal/MqttClientSessionsManager.cs | 63 +- .../MqttClientSubscriptionsManager.cs | 66 +- Source/MQTTnet/Server/Internal/MqttSession.cs | 50 +- .../Server/Internal/MqttSubscription.cs | 2 +- ...ubscribeResult.cs => UnsubscribeResult.cs} | 2 +- 21 files changed, 588 insertions(+), 558 deletions(-) delete mode 100644 Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs rename Source/MQTTnet/Server/Internal/{MqttUnsubscribeResult.cs => UnsubscribeResult.cs} (90%) diff --git a/Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs b/Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs deleted file mode 100644 index 0ec946de..00000000 --- a/Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs +++ /dev/null @@ -1,26 +0,0 @@ -// Licensed to the .NET Foundation under one or more agreements. -// The .NET Foundation licenses this file to you under the MIT license. -// See the LICENSE file in the project root for more information. - -using System.Collections.Generic; -using System.IO.Pipelines; -using Microsoft.AspNetCore.Connections; -using Microsoft.AspNetCore.Http.Features; - -namespace MQTTnet.AspNetCore.Tests.Mockups -{ - public class ConnectionContextMockup : ConnectionContext - { - public override string ConnectionId { get; set; } - - public override IFeatureCollection Features { get; } - - public override IDictionary Items { get; set; } - public override IDuplexPipe Transport { get; set; } - - public ConnectionContextMockup() - { - //Transport = new DefaultConnectionContext - } - } -} diff --git a/Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionHandlerMockup.cs b/Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionHandlerMockup.cs index 5f372c99..387e0bfb 100644 --- a/Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionHandlerMockup.cs +++ b/Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionHandlerMockup.cs @@ -2,20 +2,24 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. -using Microsoft.AspNetCore.Connections; -using MQTTnet.Adapter; -using MQTTnet.Formatter; -using MQTTnet.Server; using System; using System.Threading.Tasks; +using Microsoft.AspNetCore.Connections; +using MQTTnet.Adapter; using MQTTnet.Diagnostics; +using MQTTnet.Formatter; +using MQTTnet.Server; namespace MQTTnet.AspNetCore.Tests.Mockups { - public class ConnectionHandlerMockup: IMqttServerAdapter + public sealed class ConnectionHandlerMockup : IMqttServerAdapter { - public TaskCompletionSource Context { get; } = new TaskCompletionSource(); public Func ClientHandler { get; set; } + public TaskCompletionSource Context { get; } = new TaskCompletionSource(); + + public void Dispose() + { + } public async Task OnConnectedAsync(ConnectionContext connection) { @@ -42,9 +46,5 @@ namespace MQTTnet.AspNetCore.Tests.Mockups { return Task.CompletedTask; } - - public void Dispose() - { - } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs b/Source/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs index 24f85529..831b7b72 100644 --- a/Source/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs +++ b/Source/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs @@ -6,7 +6,7 @@ using System.IO.Pipelines; namespace MQTTnet.AspNetCore.Tests.Mockups { - public class DuplexPipeMockup : IDuplexPipe + public sealed class DuplexPipeMockup : IDuplexPipe { public DuplexPipeMockup() { @@ -20,9 +20,8 @@ namespace MQTTnet.AspNetCore.Tests.Mockups PipeWriter IDuplexPipe.Output => Send.Writer; + public Pipe Receive { get; } - public Pipe Receive { get; set; } - - public Pipe Send { get; set; } + public Pipe Send { get; } } } diff --git a/Source/MQTTnet.AspNetCore.Tests/Mockups/LimitedMemoryPool.cs b/Source/MQTTnet.AspNetCore.Tests/Mockups/LimitedMemoryPool.cs index c2f40502..242ba421 100644 --- a/Source/MQTTnet.AspNetCore.Tests/Mockups/LimitedMemoryPool.cs +++ b/Source/MQTTnet.AspNetCore.Tests/Mockups/LimitedMemoryPool.cs @@ -6,7 +6,7 @@ using System.Buffers; namespace MQTTnet.AspNetCore.Tests.Mockups { - public class LimitedMemoryPool : MemoryPool + public sealed class LimitedMemoryPool : MemoryPool { protected override void Dispose(bool disposing) { diff --git a/Source/MQTTnet.AspNetCore.Tests/Mockups/MemoryOwner.cs b/Source/MQTTnet.AspNetCore.Tests/Mockups/MemoryOwner.cs index 55e4e318..aad5e1c6 100644 --- a/Source/MQTTnet.AspNetCore.Tests/Mockups/MemoryOwner.cs +++ b/Source/MQTTnet.AspNetCore.Tests/Mockups/MemoryOwner.cs @@ -7,9 +7,9 @@ using System.Buffers; namespace MQTTnet.AspNetCore.Tests.Mockups { - public class MemoryOwner : IMemoryOwner + public sealed class MemoryOwner : IMemoryOwner { - private readonly byte[] _raw; + readonly byte[] _raw; public MemoryOwner(int size) { @@ -27,11 +27,11 @@ namespace MQTTnet.AspNetCore.Tests.Mockups Memory = _raw; } + public Memory Memory { get; } + public void Dispose() { ArrayPool.Shared.Return(_raw); } - - public Memory Memory { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs b/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs index 190ff485..a9b08a87 100644 --- a/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs +++ b/Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs @@ -2,22 +2,21 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. -using Microsoft.AspNetCore.Builder; -using Microsoft.AspNetCore.Connections; -using Microsoft.AspNetCore.Hosting; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.VisualStudio.TestTools.UnitTesting; - using System; using System.Net; using System.Threading; using System.Threading.Tasks; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Connections; +using Microsoft.AspNetCore.Hosting; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Adapter; using MQTTnet.AspNetCore.Tests.Mockups; +using MQTTnet.Client; using MQTTnet.Exceptions; using MQTTnet.Formatter; using MQTTnet.Packets; -using MQTTnet.Client; using MQTTnet.Protocol; using MQTTnet.Tests.Extensions; @@ -26,20 +25,6 @@ namespace MQTTnet.AspNetCore.Tests [TestClass] public class MqttConnectionContextTest { - [TestMethod] - public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted() - { - var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535)); - var pipe = new DuplexPipeMockup(); - var connection = new DefaultConnectionContext(); - connection.Transport = pipe; - var ctx = new MqttConnectionContext(serializer, connection); - - pipe.Receive.Writer.Complete(); - - await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None)); - } - [TestMethod] public async Task TestCorruptedConnectPacket() { @@ -49,13 +34,43 @@ namespace MQTTnet.AspNetCore.Tests var connection = new DefaultConnectionContext(); connection.Transport = pipe; var ctx = new MqttConnectionContext(serializer, connection); - + await pipe.Receive.Writer.WriteAsync(writer.AddMqttHeader(MqttControlPacketType.Connect, new byte[0])); await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None)); // the first exception should complete the pipes so if someone tries to use the connection after that it should throw immidiatly - await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None)); + await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None)); + } + + [TestMethod] + public async Task TestEndpoint() + { + var mockup = new ConnectionHandlerMockup(); + + + using (var host = new WebHostBuilder().UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx)))) + .UseStartup() + .ConfigureServices( + (hostContext, services) => + { + services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint()); + services.AddSingleton(mockup); + }) + .Build()) + + using (var client = new MqttFactory().CreateMqttClient()) + { + host.Start(); + await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build(), CancellationToken.None); + + var ctx = await mockup.Context.Task; +#if NETCOREAPP3_1 + var ep = IPEndPoint.Parse(ctx.Endpoint); + Assert.IsNotNull(ep); +#endif + Assert.IsNotNull(ctx); + } } // COMMENTED OUT DUE TO DEAD LOCK? OR VERY VERY SLOW PERFORMANCE ON LOCAL DEV MACHINE. TEST WAS STILL RUNNING AFTER SEVERAL MINUTES! @@ -78,7 +93,7 @@ namespace MQTTnet.AspNetCore.Tests // await Task.WhenAll(tasks).ConfigureAwait(false); //} - + [TestMethod] public async Task TestLargePacket() { @@ -94,43 +109,25 @@ namespace MQTTnet.AspNetCore.Tests Assert.IsTrue(readResult.Buffer.Length > 20000); } - private class Startup + [TestMethod] + public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted() + { + var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535)); + var pipe = new DuplexPipeMockup(); + var connection = new DefaultConnectionContext(); + connection.Transport = pipe; + var ctx = new MqttConnectionContext(serializer, connection); + + pipe.Receive.Writer.Complete(); + + await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None)); + } + + class Startup { public void Configure(IApplicationBuilder app) - { + { } } - - [TestMethod] - public async Task TestEndpoint() - { - var mockup = new ConnectionHandlerMockup(); - - - using (var host = new WebHostBuilder() - .UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx)))) - .UseStartup() - .ConfigureServices((hostContext, services) => - { - services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint()); - services.AddSingleton(mockup); - }) - .Build()) - - using (var client = new MqttFactory().CreateMqttClient()) - { - host.Start(); - await client.ConnectAsync(new MqttClientOptionsBuilder() - .WithTcpServer("localhost") - .Build(), CancellationToken.None); - - var ctx = await mockup.Context.Task; -#if NETCOREAPP3_1 - var ep = IPEndPoint.Parse(ctx.Endpoint); - Assert.IsNotNull(ep); -#endif - Assert.IsNotNull(ctx); - } - } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs b/Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs index 056a6abd..d711067b 100644 --- a/Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs +++ b/Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs @@ -12,7 +12,7 @@ using MQTTnet.Packets; namespace MQTTnet.AspNetCore.Tests { [TestClass] - public class ReaderExtensionsTest + public sealed class ReaderExtensionsTest { [TestMethod] public void TestTryDeserialize() diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs index 8260b3f4..673e46b3 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs @@ -13,13 +13,12 @@ using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Formatter; -using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Tests.Extensions; using MQTTnet.Tests.Mockups; -namespace MQTTnet.Tests +namespace MQTTnet.Tests.Formatter { [TestClass] public sealed class MqttPacketSerialization_V3_Binary_Tests diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs index 5ae86cc1..e1135529 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs @@ -10,7 +10,7 @@ using MQTTnet.Formatter; using MQTTnet.Packets; using MQTTnet.Protocol; -namespace MQTTnet.Tests +namespace MQTTnet.Tests.Formatter { [TestClass] public sealed class MqttPacketSerialization_V3_Tests diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs index 1d9e9740..fd7a70e2 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs @@ -9,7 +9,7 @@ using MQTTnet.Formatter; using MQTTnet.Packets; using MQTTnet.Protocol; -namespace MQTTnet.Tests +namespace MQTTnet.Tests.Formatter { [TestClass] public sealed class MqttPacketSerialization_V5_Tests diff --git a/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs b/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs index ea3e743c..19a4f427 100644 --- a/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs +++ b/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs @@ -130,9 +130,7 @@ namespace MQTTnet.Tests.Server CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId) { - ulong topicHashMask; // not needed - bool hasWildcard; // not needed - MqttSubscription.CalculateTopicHash(topic, out var topicHash, out topicHashMask, out hasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out _, out _); return _subscriptionsManager.CheckSubscriptions(topic, topicHash, applicationMessageQoSLevel, senderClientId); } } diff --git a/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs b/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs index 70e9bffd..31e1ab05 100644 --- a/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs @@ -1,9 +1,11 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Text; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Tests.Mockups; @@ -13,43 +15,299 @@ namespace MQTTnet.Tests.Server [TestClass] public sealed class SubscriptionTopicHashTests { - enum TopicHashSelector { NoWildcard, SingleWildcard, MultiWildcard }; - MqttSession _clientSession; + /// + /// Even fairly regularly named topics as generated by the topic generator should result in shallow hash buckets + /// + /// [TestMethod] - public void Match_Hash_Test_SingleWildCard() + public void Check_Hash_Bucket_Depth() { - var l0 = "pub0"; - var l1 = "topic1"; - var l2 = "+"; - var l3 = "prop1"; - var topic = $"{l0}/{l1}/{l2}/{l3}"; + const int numPublishers = 5000; + const int numTopicsPerPublisher = 10; + + TopicGenerator.Generate( + numPublishers, + numTopicsPerPublisher, + out var topicsByPublisher, + out _, + out _); + + // There will be many 'similar' topics ending with, i.e. "sensor100", "sensor101", ... + // Hash bucket depths should remain low. + var bucketDepths = new Dictionary(); + var maxBucketDepth = 0; + ulong maxBucketDepthHash = 0; + + var topicsByHash = new Dictionary>(); + + foreach (var t in topicsByPublisher) + { + var topics = t.Value; + foreach (var topic in topics) + { + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard); + + bucketDepths.TryGetValue(topicHash, out var currentValue); + ++currentValue; + bucketDepths[topicHash] = currentValue; + + if (currentValue > maxBucketDepth) + { + maxBucketDepth = currentValue; + maxBucketDepthHash = topicHash; + } + + if (!topicsByHash.TryGetValue(topicHash, out var topicList)) + { + topicList = new List(); + topicsByHash.Add(topicHash, topicList); + } + + topicList.Add(topic); + } + } + + var maxDepthTopics = topicsByHash[maxBucketDepthHash]; + + Console.Write("Max bucket depth is " + maxBucketDepth); + + // for the test case the bucket depth should be less than 100 + Assert.IsTrue(maxBucketDepth < 100, "Unexpected high topic hash bucket depth"); + } + + + [TestMethod] + public void Check_Selected_Topic_Hashes() + { + CheckTopicHash("client1/building1/level1/sensor1", 0x655D4AF100000000, 0xFFFFFFFFFFFFFFFF); + CheckTopicHash("client1/building1/+/sensor1", 0x655D00F100000000, 0xFFFF00FFFFFFFFFF); + CheckTopicHash("client1/+/level1/+", 0x65004A0000000000, 0xFF00FF00FFFFFFFF); + CheckTopicHash("client1/building1/level1/#", 0x655D4A0000000000, 0xFFFFFF0000000000); + CheckTopicHash("client1/+/level1/#", 0x65004A0000000000, 0xFF00FF0000000000); + } + + /// + /// Test long topic name with last level being # + /// + [TestMethod] + public void Match_Hash_Test_LongTopic_DetectMultiLevelWildcard() + { + var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/#"; + + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); + + Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); + } + + + /// + /// Test long topic name with last level being + + /// + [TestMethod] + public void Match_Hash_Test_LongTopic_DetectSingleLevelWildcard() + { + var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/+"; + + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); + + Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); + } + + /// + /// Test long topic name with last level being # + /// + [TestMethod] + public void Match_Hash_Test_LongTopic_MultiWildcard() + { + var sb = new StringBuilder(); + const int numLevels = 8; + var levelNames = new string[numLevels]; + for (var i = 0; i < numLevels; ++i) + { + if (i > 0) + { + sb.Append("/"); + } + + string levelName; + if (i == numLevels - 1) + { + // last one is # + levelName = "#"; + } + else + { + levelName = "level" + i; + } + + levelNames[i] = levelName; + sb.Append(levelName); + } + + var topic = sb.ToString(); + + // UInt64 is limited to 8 levels MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); var hashBytes = GetBytes(topicHash); - Assert.AreNotEqual(hashBytes[0], 0, "checksum 0 mismatch"); - Assert.AreNotEqual(hashBytes[1], 0, "checksum 1 mismatch"); - Assert.AreEqual(hashBytes[2], 0, "checksum 2 mismatch"); - Assert.AreNotEqual(hashBytes[3], 0, "checksum 3 mismatch"); - Assert.AreEqual(hashBytes[4], 0, "checksum 4 mismatch"); - Assert.AreEqual(hashBytes[5], 0, "checksum 5 mismatch"); - Assert.AreEqual(hashBytes[6], 0, "checksum 6 mismatch"); - Assert.AreEqual(hashBytes[7], 0, "checksum 7 mismatch"); + // all bytes should contain checksum + var count = 0; + foreach (var h in hashBytes) + { + if (count < 7) + { + Assert.AreNotEqual(h, 0, "checksum mismatch"); + } + else + { + Assert.AreEqual(h, 0, "checksum mismatch"); + } - // The mask should have zeroes where the wildcard and ff at the end + ++count; + } + + // The mask should have ff except for last level var hashMaskBytes = GetBytes(topicHashMask); - Assert.AreEqual(hashMaskBytes[0], 0xff, "mask 0 mismatch"); - Assert.AreEqual(hashMaskBytes[1], 0xff, "mask 1 mismatch"); - Assert.AreEqual(hashMaskBytes[2], 0, "mask 2 mismatch"); - Assert.AreEqual(hashMaskBytes[3], 0xff, "mask 3 mismatch"); - Assert.AreEqual(hashMaskBytes[4], 0xff, "mask 4 mismatch"); - Assert.AreEqual(hashMaskBytes[5], 0xff, "mask 5 mismatch"); - Assert.AreEqual(hashMaskBytes[6], 0xff, "mask 6 mismatch"); - Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch"); + count = 0; + foreach (var h in hashMaskBytes) + { + if (count < 7) + { + Assert.AreEqual(h, 0xff, "mask mismatch"); + } + else + { + Assert.AreEqual(h, 0, "last mask mismatch"); + } + + ++count; + } + } + + /// + /// Test long topic name exceeding 8 levels + /// + [TestMethod] + public void Match_Hash_Test_LongTopic_NoWildCard() + { + var sb = new StringBuilder(); + const int numLevels = 9; + var levelNames = new string[numLevels]; + for (var i = 0; i < numLevels; ++i) + { + if (i > 0) + { + sb.Append("/"); + } + + var levelName = "level" + i; + levelNames[i] = levelName; + sb.Append(levelName); + } + + var topic = sb.ToString(); + + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); + + Assert.IsFalse(topicHasWildcard, "Wildcard detected when not present"); + + + var hashBytes = GetBytes(topicHash); + // all bytes should contain checksum + var count = 0; + foreach (var h in hashBytes) + { + Assert.AreNotEqual(h, 0, "checksum mismatch"); + ++count; + } + + // The mask should have ff + var hashMaskBytes = GetBytes(topicHashMask); + count = 0; + foreach (var h in hashMaskBytes) + { + Assert.AreEqual(h, 0xff, "mask mismatch"); + ++count; + } + } + + /// + /// Test long topic name with last level being + + /// + [TestMethod] + public void Match_Hash_Test_LongTopic_SingleWildCard() + { + var sb = new StringBuilder(); + const int numLevels = 8; + var levelNames = new string[numLevels]; + for (var i = 0; i < numLevels; ++i) + { + if (i > 0) + { + sb.Append("/"); + } + + string levelName; + if (i == numLevels - 1) + { + // last one is + + levelName = "+"; + } + else + { + levelName = "level" + i; + } + + levelNames[i] = levelName; + sb.Append(levelName); + } + + var topic = sb.ToString(); + + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); + + Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); + + + var hashBytes = GetBytes(topicHash); + // all bytes should contain checksum + var count = 0; + foreach (var h in hashBytes) + { + if (count < 7) + { + Assert.AreNotEqual(h, 0, "checksum mismatch"); + } + else + { + // wildcard position + Assert.AreEqual(h, 0, "checksum mismatch"); + } + + ++count; + } + + // The mask should have ff + var hashMaskBytes = GetBytes(topicHashMask); + count = 0; + foreach (var h in hashMaskBytes) + { + if (count < 7) + { + Assert.AreEqual(h, 0xff, "mask mismatch"); + } + else + { + Assert.AreEqual(h, 0, "last mask mismatch"); + } + + ++count; + } } [TestMethod] @@ -123,205 +381,13 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch"); } - /// - /// Test long topic name with last level being # - /// [TestMethod] - public void Match_Hash_Test_LongTopic_MultiWildcard() + public async Task Match_Hash_Test_Search_MultiWildcard() { - var sb = new StringBuilder(); - const int numLevels = 8; - var levelNames = new string[numLevels]; - for (var i = 0; i < numLevels; ++i) - { - if (i > 0) - sb.Append("/"); - string levelName; - if (i == numLevels - 1) - { - // last one is # - levelName = "#"; - } - else - { - levelName = "level" + i; - } - levelNames[i] = levelName; - sb.Append(levelName); - } - var topic = sb.ToString(); - - // UInt64 is limited to 8 levels - - MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); - - Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); - - var hashBytes = GetBytes(topicHash); - // all bytes should contain checksum - int count = 0; - foreach (var h in hashBytes) - { - if (count < 7) - { - Assert.AreNotEqual(h, 0, "checksum mismatch"); - } - else - { - Assert.AreEqual(h, 0, "checksum mismatch"); - } - ++count; - } - // The mask should have ff except for last level - var hashMaskBytes = GetBytes(topicHashMask); - count = 0; - foreach (var h in hashMaskBytes) - { - if (count < 7) - { - Assert.AreEqual(h, 0xff, "mask mismatch"); - } - else - { - Assert.AreEqual(h, 0, "last mask mismatch"); - } - ++count; - } - } - - /// - /// Test long topic name with last level being + - /// - [TestMethod] - public void Match_Hash_Test_LongTopic_SingleWildCard() - { - var sb = new StringBuilder(); - const int numLevels = 8; - var levelNames = new string[numLevels]; - for (var i = 0; i < numLevels; ++i) - { - if (i > 0) - sb.Append("/"); - string levelName; - if (i == numLevels - 1) - { - // last one is + - levelName = "+"; - } - else - { - levelName = "level" + i; - } - levelNames[i] = levelName; - sb.Append(levelName); - } - var topic = sb.ToString(); - - MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); - - Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); - - - var hashBytes = GetBytes(topicHash); - // all bytes should contain checksum - int count = 0; - foreach (var h in hashBytes) - { - if (count < 7) - { - Assert.AreNotEqual(h, 0, "checksum mismatch"); - } - else - { - // wildcard position - Assert.AreEqual(h, 0, "checksum mismatch"); - } - ++count; - } - // The mask should have ff - var hashMaskBytes = GetBytes(topicHashMask); - count = 0; - foreach (var h in hashMaskBytes) - { - if (count < 7) - { - Assert.AreEqual(h, 0xff, "mask mismatch"); - } - else - { - Assert.AreEqual(h, 0, "last mask mismatch"); - } - ++count; - } - } - - /// - /// Test long topic name exceeding 8 levels - /// - [TestMethod] - public void Match_Hash_Test_LongTopic_NoWildCard() - { - var sb = new StringBuilder(); - const int numLevels = 9; - var levelNames = new string[numLevels]; - for(var i=0; i < numLevels; ++i) - { - if (i > 0) - sb.Append("/"); - var levelName = "level" + i; - levelNames[i] = levelName; - sb.Append(levelName); - } - var topic = sb.ToString(); - - MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); - - Assert.IsFalse(topicHasWildcard, "Wildcard detected when not present"); - - - var hashBytes = GetBytes(topicHash); - // all bytes should contain checksum - int count = 0; - foreach(var h in hashBytes) - { - Assert.AreNotEqual(h, 0, "checksum mismatch"); - ++count; - } - // The mask should have ff - var hashMaskBytes = GetBytes(topicHashMask); - count = 0; - foreach (var h in hashMaskBytes) - { - Assert.AreEqual(h, 0xff, "mask mismatch"); - ++count; - } - } - - - /// - /// Test long topic name with last level being + - /// - [TestMethod] - public void Match_Hash_Test_LongTopic_DetectSingleLevelWildcard() - { - var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/+"; - - MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); - - Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); - } - - /// - /// Test long topic name with last level being # - /// - [TestMethod] - public void Match_Hash_Test_LongTopic_DetectMultiLevelWildcard() - { - var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/#"; - - MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); - - Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); + var topics = await PrepareTopicHashSubscriptions(TopicHashSelector.MultiWildcard); + var matchCount = CheckTopicSubscriptions(_clientSession, topics, "Multi Wildcard"); + // Should match all topics + Assert.AreEqual(topics.Count, matchCount, "Topics not matched"); } [TestMethod] @@ -373,77 +439,38 @@ namespace MQTTnet.Tests.Server } [TestMethod] - public async Task Match_Hash_Test_Search_MultiWildcard() + public void Match_Hash_Test_SingleWildCard() { - var topics = await PrepareTopicHashSubscriptions(TopicHashSelector.MultiWildcard); - var matchCount = CheckTopicSubscriptions(_clientSession, topics, "Multi Wildcard"); - // Should match all topics - Assert.AreEqual(topics.Count, matchCount, "Topics not matched"); - } + var l0 = "pub0"; + var l1 = "topic1"; + var l2 = "+"; + var l3 = "prop1"; + var topic = $"{l0}/{l1}/{l2}/{l3}"; - /// - /// Even fairly regularly named topics as generated by the topic generator should result in shallow hash buckets - /// - /// - [TestMethod] - public void Check_Hash_Bucket_Depth() - { - const int numPublishers = 5000; - const int numTopicsPerPublisher = 10; + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); - TopicGenerator.Generate(numPublishers, numTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher); + Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); - // There will be many 'similar' topics ending with, i.e. "sensor100", "sensor101", ... - // Hash bucket depths should remain low. - var bucketDepths = new Dictionary(); - int maxBucketDepth = 0; - UInt64 maxBucketDepthHash = 0; + var hashBytes = GetBytes(topicHash); + Assert.AreNotEqual(hashBytes[0], 0, "checksum 0 mismatch"); + Assert.AreNotEqual(hashBytes[1], 0, "checksum 1 mismatch"); + Assert.AreEqual(hashBytes[2], 0, "checksum 2 mismatch"); + Assert.AreNotEqual(hashBytes[3], 0, "checksum 3 mismatch"); + Assert.AreEqual(hashBytes[4], 0, "checksum 4 mismatch"); + Assert.AreEqual(hashBytes[5], 0, "checksum 5 mismatch"); + Assert.AreEqual(hashBytes[6], 0, "checksum 6 mismatch"); + Assert.AreEqual(hashBytes[7], 0, "checksum 7 mismatch"); - var topicsByHash = new Dictionary>(); - - foreach (var t in topicsByPublisher) - { - var topics = t.Value; - foreach (var topic in topics) - { - MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard); - - bucketDepths.TryGetValue(topicHash, out var currentValue); - ++currentValue; - bucketDepths[topicHash] = currentValue; - - if (currentValue > maxBucketDepth) - { - maxBucketDepth = currentValue; - maxBucketDepthHash = topicHash; - } - - if (!topicsByHash.TryGetValue(topicHash, out var topicList)) - { - topicList = new List(); - topicsByHash.Add(topicHash, topicList); - } - topicList.Add(topic); - } - } - - var maxDepthTopics = topicsByHash[maxBucketDepthHash]; - - Console.Write("Max bucket depth is " + maxBucketDepth); - - // for the test case the bucket depth should be less than 100 - Assert.IsTrue(maxBucketDepth < 100, "Unexpected high topic hash bucket depth"); - } - - - [TestMethod] - public void Check_Selected_Topic_Hashes() - { - CheckTopicHash("client1/building1/level1/sensor1", 0x655D4AF100000000, 0xFFFFFFFFFFFFFFFF); - CheckTopicHash("client1/building1/+/sensor1", 0x655D00F100000000, 0xFFFF00FFFFFFFFFF); - CheckTopicHash("client1/+/level1/+", 0x65004A0000000000, 0xFF00FF00FFFFFFFF); - CheckTopicHash("client1/building1/level1/#", 0x655D4A0000000000, 0xFFFFFF0000000000); - CheckTopicHash("client1/+/level1/#", 0x65004A0000000000, 0xFF00FF0000000000); + // The mask should have zeroes where the wildcard and ff at the end + var hashMaskBytes = GetBytes(topicHashMask); + Assert.AreEqual(hashMaskBytes[0], 0xff, "mask 0 mismatch"); + Assert.AreEqual(hashMaskBytes[1], 0xff, "mask 1 mismatch"); + Assert.AreEqual(hashMaskBytes[2], 0, "mask 2 mismatch"); + Assert.AreEqual(hashMaskBytes[3], 0xff, "mask 3 mismatch"); + Assert.AreEqual(hashMaskBytes[4], 0xff, "mask 4 mismatch"); + Assert.AreEqual(hashMaskBytes[5], 0xff, "mask 5 mismatch"); + Assert.AreEqual(hashMaskBytes[6], 0xff, "mask 6 mismatch"); + Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch"); } void CheckTopicHash(string topic, ulong expectedHash, ulong expectedHashMask) @@ -459,13 +486,66 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(expectedHashMask, hashMask, "Topic hash mask not as expected"); } + int CheckTopicSubscriptions(MqttSession clientSession, List topicsToFind, string subject) + { + var matchCount = 0; + + { + var resultCount = 0; + + var stopWatch = new Stopwatch(); + stopWatch.Start(); + var countUp = 0; + var countDown = topicsToFind.Count - 1; + for (; countUp < topicsToFind.Count; ++countUp, --countDown) + { + var topicToFind = topicsToFind[countDown]; + + MqttSubscription.CalculateTopicHash(topicToFind, out var topicHash, out _, out _); + + clientSession.TryCheckSubscriptions(topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient", out var result); + if (result.IsSubscribed) + { + ++matchCount; + } + + ++resultCount; + } + + stopWatch.Stop(); + + Console.Write("Match count: " + matchCount + "; "); + + Console.WriteLine(subject + " lookup milliseconds: " + stopWatch.ElapsedMilliseconds); + } + + return matchCount; + } + + byte[] GetBytes(ulong value) + { + var bytes = BitConverter.GetBytes(value); + // Ensure that highest byte comes first for comparison left to right + if (BitConverter.IsLittleEndian) + { + return bytes.Reverse().ToArray(); + } + + return bytes; + } + async Task> PrepareTopicHashSubscriptions(TopicHashSelector selector) { const int numPublishers = 1; const int numTopicsPerPublisher = 10000; - TopicGenerator.Generate(numPublishers, numTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher); + TopicGenerator.Generate( + numPublishers, + numTopicsPerPublisher, + out var topicsByPublisher, + out var singleWildcardTopicsByPublisher, + out var multiWildcardTopicsByPublisher); var topics = topicsByPublisher.FirstOrDefault().Value; var singleWildcardTopics = singleWildcardTopicsByPublisher.FirstOrDefault().Value; @@ -477,15 +557,7 @@ namespace MQTTnet.Tests.Server var eventContainer = new MqttServerEventContainer(); var retainedMessagesManager = new MqttRetainedMessagesManager(eventContainer, logger); var sessionManager = new MqttClientSessionsManager(serverOptions, retainedMessagesManager, eventContainer, logger); - _clientSession = new MqttSession( - clientId, - false, - new Dictionary(), - serverOptions, - eventContainer, - retainedMessagesManager, - sessionManager - ); + _clientSession = new MqttSession(clientId, false, new Dictionary(), serverOptions, eventContainer, retainedMessagesManager, sessionManager); List topicsToSubscribe; @@ -501,62 +573,26 @@ namespace MQTTnet.Tests.Server topicsToSubscribe = topics; break; } + foreach (var t in topicsToSubscribe) { - var subPacket = new Packets.MqttSubscribePacket(); - var filter = new Packets.MqttTopicFilter + var subPacket = new MqttSubscribePacket(); + var filter = new MqttTopicFilter { Topic = t }; subPacket.TopicFilters.Add(filter); - await _clientSession.SubscriptionsManager.Subscribe(subPacket, default); + await _clientSession.Subscribe(subPacket, default); } return topics; } - int CheckTopicSubscriptions(MqttSession clientSession, List topicsToFind, string subject) + enum TopicHashSelector { - var matchCount = 0; - - { - int resultCount = 0; - - var stopWatch = new System.Diagnostics.Stopwatch(); - stopWatch.Start(); - var countUp = 0; - var countDown = topicsToFind.Count - 1; - for (; countUp < topicsToFind.Count; ++countUp, --countDown) - { - var topicToFind = topicsToFind[countDown]; - - MqttSubscription.CalculateTopicHash(topicToFind, out var topicHash, out var hashMask, out var hasWildcard); - - var result = clientSession.SubscriptionsManager.CheckSubscriptions(topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient"); - if (result.IsSubscribed) - { - ++matchCount; - } - ++resultCount; - } - - stopWatch.Stop(); - - Console.Write("Match count: " + matchCount + "; "); - - Console.WriteLine(subject + " lookup milliseconds: " + stopWatch.ElapsedMilliseconds); - } - - return matchCount; - } - - byte[] GetBytes(UInt64 value) - { - var bytes = BitConverter.GetBytes(value); - // Ensure that highest byte comes first for comparison left to right - if (BitConverter.IsLittleEndian) - return bytes.Reverse().ToArray(); - return bytes; + NoWildcard, + SingleWildcard, + MultiWildcard } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet.Tests/TopicGenerator.cs b/Source/MQTTnet.Tests/TopicGenerator.cs index 64ae1144..ddfbe709 100644 --- a/Source/MQTTnet.Tests/TopicGenerator.cs +++ b/Source/MQTTnet.Tests/TopicGenerator.cs @@ -1,8 +1,7 @@ using System; using System.Collections.Generic; -using System.Text; -namespace MQTTnet.Tests.Server +namespace MQTTnet.Tests { public class TopicGenerator { diff --git a/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs index 18da46a1..88cbbbc6 100644 --- a/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs @@ -10,16 +10,16 @@ namespace MQTTnet.Formatter { public sealed class MqttUnsubAckPacketFactory { - public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, MqttUnsubscribeResult mqttUnsubscribeResult) + public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult) { if (unsubscribePacket == null) { throw new ArgumentNullException(nameof(unsubscribePacket)); } - if (mqttUnsubscribeResult == null) + if (unsubscribeResult == null) { - throw new ArgumentNullException(nameof(mqttUnsubscribeResult)); + throw new ArgumentNullException(nameof(unsubscribeResult)); } var unsubAckPacket = new MqttUnsubAckPacket @@ -28,7 +28,7 @@ namespace MQTTnet.Formatter }; // MQTTv5.0.0 only. - unsubAckPacket.ReasonCodes = mqttUnsubscribeResult.ReasonCodes; + unsubAckPacket.ReasonCodes = unsubscribeResult.ReasonCodes; return unsubAckPacket; } diff --git a/Source/MQTTnet/Internal/AsyncEvent.cs b/Source/MQTTnet/Internal/AsyncEvent.cs index 8e988d75..e7adc7b5 100644 --- a/Source/MQTTnet/Internal/AsyncEvent.cs +++ b/Source/MQTTnet/Internal/AsyncEvent.cs @@ -123,7 +123,7 @@ namespace MQTTnet.Internal } catch (Exception exception) { - logger.Warning(exception, $"Error while invoking event ({typeof(TEventArgs)})."); + logger.Warning(exception, $"Error while invoking event with arguments of type {typeof(TEventArgs)}."); } } } diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 1bc51152..674a262f 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -33,6 +33,7 @@ namespace MQTTnet.Server readonly Dictionary _topicAlias = new Dictionary(); CancellationTokenSource _cancellationToken; + bool _disconnectPacketSent; public MqttClient( MqttConnectPacket connectPacket, @@ -146,19 +147,22 @@ namespace MQTTnet.Server { IsRunning = false; - // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+. - if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) + if (!_disconnectPacketSent) { - // The Client or Server MAY send a DISCONNECT packet before closing the Network Connection. - // This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding - // a configuration option is requested in the future. - if (reason != MqttDisconnectReasonCode.NormalDisconnection) + // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+. + if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) { - // Is is very important to send the DISCONNECT packet here BEFORE cancelling the - // token because the entire connection is closed (disposed) as soon as the cancellation - // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive - // at the client! - await TrySendDisconnectPacket(reason).ConfigureAwait(false); + // The Client or Server MAY send a DISCONNECT packet before closing the Network Connection. + // This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding + // a configuration option is requested in the future. + if (reason != MqttDisconnectReasonCode.NormalDisconnection) + { + // Is is very important to send the DISCONNECT packet here BEFORE cancelling the + // token because the entire connection is closed (disposed) as soon as the cancellation + // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive + // at the client! + await TrySendDisconnectPacket(reason).ConfigureAwait(false); + } } } @@ -282,7 +286,7 @@ namespace MQTTnet.Server async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { - var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false); + var subscribeResult = await Session.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false); var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult); @@ -306,7 +310,7 @@ namespace MQTTnet.Server async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { - var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); + var unsubscribeResult = await Session.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult); @@ -484,7 +488,7 @@ namespace MQTTnet.Server try { - while (!cancellationToken.IsCancellationRequested) + while (!cancellationToken.IsCancellationRequested && !IsTakenOver && IsRunning) { packetBusItem = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false); @@ -566,6 +570,9 @@ namespace MQTTnet.Server { try { + // This also indicates that it was tried at least! + _disconnectPacketSent = true; + var disconnectPacket = _packetFactories.Disconnect.Create(reasonCode); using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout)) diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 87fcdd34..186a15ca 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -150,11 +150,17 @@ namespace MQTTnet.Server foreach (var session in subscriberSessions) { - var checkSubscriptionsResult = session.SubscriptionsManager.CheckSubscriptions( + if(!session.TryCheckSubscriptions( applicationMessage.Topic, topicHash, applicationMessage.QualityOfServiceLevel, - senderId); + senderId, + out var checkSubscriptionsResult)) + { + // Checking the subscriptions has failed for the session. The session + // will be ignored. + continue; + } if (!checkSubscriptionsResult.IsSubscribed) { @@ -301,7 +307,7 @@ namespace MQTTnet.Server channelAdapter.Endpoint, client.Session.Items); - await _eventContainer.ClientConnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + await _eventContainer.ClientConnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false); } await client.RunAsync().ConfigureAwait(false); @@ -414,7 +420,7 @@ namespace MQTTnet.Server var clientSession = GetClientSession(clientId); - var subscribeResult = await clientSession.SubscriptionsManager.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false); + var subscribeResult = await clientSession.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false); if (subscribeResult.RetainedMessages != null) { @@ -441,7 +447,7 @@ namespace MQTTnet.Server var fakeUnsubscribePacket = new MqttUnsubscribePacket(); fakeUnsubscribePacket.TopicFilters.AddRange(topicFilters); - return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket, CancellationToken.None); + return GetClientSession(clientId).Unsubscribe(fakeUnsubscribePacket, CancellationToken.None); } MqttClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session) @@ -484,10 +490,14 @@ namespace MQTTnet.Server using (await _createConnectionSyncRoot.EnterAsync().ConfigureAwait(false)) { - MqttSession session; MqttSession oldSession; + MqttClient oldClient; + lock (_sessionsManagementLock) { + MqttSession session; + + // Create a new session (if required). if (!_sessions.TryGetValue(connectPacket.ClientId, out oldSession)) { session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist); @@ -505,6 +515,7 @@ namespace MQTTnet.Server _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId); session = oldSession; oldSession = null; + // Session persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval session.IsPersistent = sessionShouldPersist; connAckPacket.IsSessionPresent = true; @@ -513,39 +524,41 @@ namespace MQTTnet.Server } _sessions[connectPacket.ClientId] = session; + + // Create a new client (always required). + _clients.TryGetValue(connectPacket.ClientId, out oldClient); + if (oldClient != null) + { + // This will stop the current client from sending and receiving but remains the connection active + // for a later DISCONNECT packet. + oldClient.IsTakenOver = true; + } + + client = CreateClient(connectPacket, channelAdapter, session); + + _clients[connectPacket.ClientId] = client; } if (!connAckPacket.IsSessionPresent) { // TODO: This event is not yet final. It can already be used but restoring sessions from storage will be added later! var preparingSessionEventArgs = new PreparingSessionEventArgs(); - await _eventContainer.PreparingSessionEvent.InvokeAsync(preparingSessionEventArgs).ConfigureAwait(false); + await _eventContainer.PreparingSessionEvent.TryInvokeAsync(preparingSessionEventArgs, _logger).ConfigureAwait(false); } - - MqttClient existingClient; - - lock (_clients) + + if (oldClient != null) { - _clients.TryGetValue(connectPacket.ClientId, out existingClient); - client = CreateClient(connectPacket, channelAdapter, session); - - _clients[connectPacket.ClientId] = client; - } - - if (existingClient != null) - { - existingClient.IsTakenOver = true; - await existingClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false); + await oldClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false); if (_eventContainer.ClientDisconnectedEvent.HasHandlers) { var eventArgs = new ClientDisconnectedEventArgs( - existingClient.Id, + oldClient.Id, MqttClientDisconnectType.Takeover, - existingClient.Endpoint, - existingClient.Session.Items); + oldClient.Endpoint, + oldClient.Session.Items); - await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + await _eventContainer.ClientDisconnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false); } } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index a6542d14..6fb45371 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -7,6 +7,7 @@ using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -30,7 +31,7 @@ namespace MQTTnet.Server readonly Dictionary _subscriptions = new Dictionary(); // Use subscription lock to maintain consistency across subscriptions and topic hash dictionaries - readonly SemaphoreSlim _subscriptionsLock = new SemaphoreSlim(1); + readonly AsyncLock _subscriptionsLock = new AsyncLock(); readonly Dictionary _wildcardSubscriptionsByTopicHash = new Dictionary(); public MqttClientSubscriptionsManager( @@ -45,13 +46,12 @@ namespace MQTTnet.Server _subscriptionChangedNotification = subscriptionChangedNotification; } - public CheckSubscriptionsResult CheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId) + public CheckSubscriptionsResult CheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel qualityOfServiceLevel, string senderId) { var possibleSubscriptions = new List(); // Check for possible subscriptions. They might have collisions but this is fine. - _subscriptionsLock.Wait(); - try + using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult()) { if (_noWildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var noWildcardSubscriptions)) { @@ -63,18 +63,14 @@ namespace MQTTnet.Server var wildcardSubscriptions = wcs.Value; var subscriptionHash = wcs.Key; var subscriptionHashMask = wildcardSubscriptions.HashMask; - + if ((topicHash & subscriptionHashMask) == subscriptionHash) { possibleSubscriptions.AddRange(wildcardSubscriptions.Subscriptions.ToList()); } } } - finally - { - _subscriptionsLock.Release(); - } - + // The pre check has evaluated that nothing is subscribed. // If there were some possible candidates they get checked below // again to avoid collisions. @@ -83,7 +79,7 @@ namespace MQTTnet.Server return CheckSubscriptionsResult.NotSubscribed; } - var senderIsReceiver = string.Equals(senderClientId, _session.Id); + var senderIsReceiver = string.Equals(senderId, _session.Id); var maxQoSLevel = -1; // Not subscribed. HashSet subscriptionIdentifiers = null; @@ -136,7 +132,7 @@ namespace MQTTnet.Server SubscriptionIdentifiers = subscriptionIdentifiers?.ToList() ?? EmptySubscriptionIdentifiers, // Start with the same QoS as the publisher. - QualityOfServiceLevel = applicationMessageQoSLevel + QualityOfServiceLevel = qualityOfServiceLevel }; // Now downgrade if required. @@ -148,7 +144,7 @@ namespace MQTTnet.Server // Subscribing to a Topic Filter at QoS 2 is equivalent to saying "I would like to receive Messages matching this filter at the QoS with which they were published". // This means a publisher is responsible for determining the maximum QoS a Message can be delivered at, but a subscriber is able to require that the Server // downgrades the QoS to one more suitable for its usage. - if (maxQoSLevel < (int)applicationMessageQoSLevel) + if (maxQoSLevel < (int)qualityOfServiceLevel) { result.QualityOfServiceLevel = (MqttQualityOfServiceLevel)maxQoSLevel; } @@ -209,7 +205,7 @@ namespace MQTTnet.Server // This call will add the new subscription to the internal storage. // So the event _ClientSubscribedTopicEvent_ must be called afterwards. _subscriptionChangedNotification?.OnSubscriptionsAdded(_session, addedSubscriptions); - + if (_eventContainer.ClientSubscribedTopicEvent.HasHandlers) { foreach (var finalTopicFilter in finalTopicFilters) @@ -218,23 +214,22 @@ namespace MQTTnet.Server await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } } - + return result; } - public async Task Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) + public async Task Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { if (unsubscribePacket == null) { throw new ArgumentNullException(nameof(unsubscribePacket)); } - var result = new MqttUnsubscribeResult(); + var result = new UnsubscribeResult(); var removedSubscriptions = new List(); - await _subscriptionsLock.WaitAsync(cancellationToken).ConfigureAwait(false); - try + using (await _subscriptionsLock.EnterAsync(cancellationToken).ConfigureAwait(false)) { foreach (var topicFilter in unsubscribePacket.TopicFilters) { @@ -265,7 +260,7 @@ namespace MQTTnet.Server if (existingSubscription != null) { var topicHash = existingSubscription.TopicHash; - + if (existingSubscription.TopicHasWildcard) { if (_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions)) @@ -294,11 +289,8 @@ namespace MQTTnet.Server } } } - finally - { - _subscriptionsLock.Release(); - _subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions); - } + + _subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions); if (_eventContainer.ClientUnsubscribedTopicEvent.HasHandlers) { @@ -345,8 +337,7 @@ namespace MQTTnet.Server // Add to subscriptions and maintain topic hash dictionaries - _subscriptionsLock.Wait(); - try + using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult()) { MqttSubscription.CalculateTopicHash(topicFilter.Topic, out var topicHash, out var topicHashMask, out var hasWildcard); @@ -396,10 +387,6 @@ namespace MQTTnet.Server subscriptions.Add(subscription); } } - finally - { - _subscriptionsLock.Release(); - } return new CreateSubscriptionResult { @@ -493,23 +480,20 @@ namespace MQTTnet.Server async Task InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken) { - var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, topicFilter, _session.Items, topicFilter); - - if (mqttSubscription == null) + var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, topicFilter, _session.Items, topicFilter) { - clientUnsubscribingTopicEventArgs.Response.ReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted; - } - else - { - clientUnsubscribingTopicEventArgs.Response.ReasonCode = MqttUnsubscribeReasonCode.Success; - } + Response = + { + ReasonCode = mqttSubscription == null ? MqttUnsubscribeReasonCode.NoSubscriptionExisted : MqttUnsubscribeReasonCode.Success + } + }; await _eventContainer.InterceptingUnsubscriptionEvent.InvokeAsync(clientUnsubscribingTopicEventArgs).ConfigureAwait(false); return clientUnsubscribingTopicEventArgs; } - public sealed class CreateSubscriptionResult + sealed class CreateSubscriptionResult { public bool IsNewSubscription { get; set; } diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs index de9ff74d..a5a045bf 100644 --- a/Source/MQTTnet/Server/Internal/MqttSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttSession.cs @@ -23,6 +23,7 @@ namespace MQTTnet.Server readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); readonly MqttServerOptions _serverOptions; + readonly MqttClientSubscriptionsManager _subscriptionsManager; readonly ConcurrentDictionary _unacknowledgedPublishPackets = new ConcurrentDictionary(); @@ -45,7 +46,7 @@ namespace MQTTnet.Server _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager)); - SubscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager); + _subscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager); } public DateTime CreatedTimestamp { get; } = DateTime.UtcNow; @@ -67,18 +68,8 @@ namespace MQTTnet.Server public long PendingDataPacketsCount => _packetBus.PartitionItemsCount(MqttPacketBusPartition.Data); - public MqttClientSubscriptionsManager SubscriptionsManager { get; } - public bool WillMessageSent { get; set; } - public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier) - { - // This will only return the matching PUBLISH packet but does not remove it. - // This is required for QoS 2. - _unacknowledgedPublishPackets.TryGetValue(packetIdentifier, out var publishPacket); - return publishPacket; - } - public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier) { _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket); @@ -107,8 +98,8 @@ namespace MQTTnet.Server public void Dispose() { - _packetBus?.Dispose(); - SubscriptionsManager.Dispose(); + _packetBus.Dispose(); + _subscriptionsManager.Dispose(); } public void EnqueueControlPacket(MqttPacketBusItem packetBusItem) @@ -150,6 +141,14 @@ namespace MQTTnet.Server _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health); } + public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier) + { + // This will only return the matching PUBLISH packet but does not remove it. + // This is required for QoS 2. + _unacknowledgedPublishPackets.TryGetValue(packetIdentifier, out var publishPacket); + return publishPacket; + } + public void Recover() { // TODO: Keep the bus and only insert pending items again. @@ -167,5 +166,30 @@ namespace MQTTnet.Server { _subscribedTopics?.Remove(topic); } + + public Task Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) + { + return _subscriptionsManager.Subscribe(subscribePacket, cancellationToken); + } + + public bool TryCheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel qualityOfServiceLevel, string senderId, out CheckSubscriptionsResult result) + { + result = null; + + try + { + result = _subscriptionsManager.CheckSubscriptions(topic, topicHash, qualityOfServiceLevel, senderId); + return true; + } + catch + { + return false; + } + } + + public Task Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) + { + return _subscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken); + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttSubscription.cs b/Source/MQTTnet/Server/Internal/MqttSubscription.cs index ec532425..930d3c72 100644 --- a/Source/MQTTnet/Server/Internal/MqttSubscription.cs +++ b/Source/MQTTnet/Server/Internal/MqttSubscription.cs @@ -179,7 +179,7 @@ namespace MQTTnet.Server RetainAsPublished = retainAsPublished; GrantedQualityOfServiceLevel = qualityOfServiceLevel; Identifier = identifier; - + CalculateTopicHash(Topic, out var hash, out var hashMask, out var hasWildcard); TopicHash = hash; TopicHashMask = hashMask; diff --git a/Source/MQTTnet/Server/Internal/MqttUnsubscribeResult.cs b/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs similarity index 90% rename from Source/MQTTnet/Server/Internal/MqttUnsubscribeResult.cs rename to Source/MQTTnet/Server/Internal/UnsubscribeResult.cs index 9e482dcf..5a93d997 100644 --- a/Source/MQTTnet/Server/Internal/MqttUnsubscribeResult.cs +++ b/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs @@ -7,7 +7,7 @@ using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class MqttUnsubscribeResult + public sealed class UnsubscribeResult { public List ReasonCodes { get; } = new List(128);