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

Fix client session takeover

Browse Source
This commit is contained in:
Christian
2022-11-06 13:28:41 +01:00
parent 97c46bea0f
commit 5c74728387
21 changed files with 588 additions and 558 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionContextMockup.cs
-26
View File
@@ -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<object, object> Items { get; set; }
public override IDuplexPipe Transport { get; set; }
public ConnectionContextMockup()
{
//Transport = new DefaultConnectionContext
}
}
}
Source/MQTTnet.AspNetCore.Tests/Mockups/ConnectionHandlerMockup.cs
+11 -11
View File
@@ -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<MqttConnectionContext> Context { get; } = new TaskCompletionSource<MqttConnectionContext>();
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public TaskCompletionSource<MqttConnectionContext> Context { get; } = new TaskCompletionSource<MqttConnectionContext>();
public void Dispose()
{
}
public async Task OnConnectedAsync(ConnectionContext connection)
{
@@ -42,9 +46,5 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
{
return Task.CompletedTask;
}
public void Dispose()
{
}
}
}
}
Source/MQTTnet.AspNetCore.Tests/Mockups/DuplexPipeMockup.cs
+3 -4
View File
@@ -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; }
}
}
Source/MQTTnet.AspNetCore.Tests/Mockups/LimitedMemoryPool.cs
+1 -1
View File
@@ -6,7 +6,7 @@ using System.Buffers;
namespace MQTTnet.AspNetCore.Tests.Mockups
{
public class LimitedMemoryPool : MemoryPool<byte>
public sealed class LimitedMemoryPool : MemoryPool<byte>
{
protected override void Dispose(bool disposing)
{
Source/MQTTnet.AspNetCore.Tests/Mockups/MemoryOwner.cs
+4 -4
View File
@@ -7,9 +7,9 @@ using System.Buffers;
namespace MQTTnet.AspNetCore.Tests.Mockups
{
public class MemoryOwner : IMemoryOwner<byte>
public sealed class MemoryOwner : IMemoryOwner<byte>
{
private readonly byte[] _raw;
readonly byte[] _raw;
public MemoryOwner(int size)
{
@@ -27,11 +27,11 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
Memory = _raw;
}
public Memory<byte> Memory { get; }
public void Dispose()
{
ArrayPool<byte>.Shared.Return(_raw);
}
public Memory<byte> Memory { get; }
}
}
Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs
+56 -59
View File
@@ -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<MqttCommunicationException>(() => 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<MqttProtocolViolationException>(() => 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<InvalidOperationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
await Assert.ThrowsExceptionAsync<InvalidOperationException>(() => 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<Startup>()
.ConfigureServices(
(hostContext, services) =>
{
services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint());
services.AddSingleton<IMqttServerAdapter>(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<MqttCommunicationException>(() => 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<Startup>()
.ConfigureServices((hostContext, services) =>
{
services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint());
services.AddSingleton<IMqttServerAdapter>(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);
}
}
}
}
}
Source/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs
+1 -1
View File
@@ -12,7 +12,7 @@ using MQTTnet.Packets;
namespace MQTTnet.AspNetCore.Tests
{
[TestClass]
public class ReaderExtensionsTest
public sealed class ReaderExtensionsTest
{
[TestMethod]
public void TestTryDeserialize()
Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs
+1 -2
View File
@@ -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
Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs
+1 -1
View File
@@ -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
Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V5_Tests.cs
+1 -1
View File
@@ -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
Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs
+1 -3
View File
@@ -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);
}
}
Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs
+379 -343
View File
@@ -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;
/// <summary>
/// Even fairly regularly named topics as generated by the topic generator should result in shallow hash buckets
/// </summary>
/// <returns></returns>
[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<ulong, int>();
var maxBucketDepth = 0;
ulong maxBucketDepthHash = 0;
var topicsByHash = new Dictionary<ulong, List<string>>();
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<string>();
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);
}
/// <summary>
/// Test long topic name with last level being #
/// </summary>
[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");
}
/// <summary>
/// Test long topic name with last level being +
/// </summary>
[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");
}
/// <summary>
/// Test long topic name with last level being #
/// </summary>
[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;
}
}
/// <summary>
/// Test long topic name exceeding 8 levels
/// </summary>
[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;
}
}
/// <summary>
/// Test long topic name with last level being +
/// </summary>
[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");
}
/// <summary>
/// Test long topic name with last level being #
/// </summary>
[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;
}
}
/// <summary>
/// Test long topic name with last level being +
/// </summary>
[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;
}
}
/// <summary>
/// Test long topic name exceeding 8 levels
/// </summary>
[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;
}
}
/// <summary>
/// Test long topic name with last level being +
/// </summary>
[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");
}
/// <summary>
/// Test long topic name with last level being #
/// </summary>
[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}";
/// <summary>
/// Even fairly regularly named topics as generated by the topic generator should result in shallow hash buckets
/// </summary>
/// <returns></returns>
[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<UInt64, int>();
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<UInt64, List<string>>();
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<string>();
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<string> 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<List<string>> 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<object, object>(),
serverOptions,
eventContainer,
retainedMessagesManager,
sessionManager
);
_clientSession = new MqttSession(clientId, false, new Dictionary<object, object>(), serverOptions, eventContainer, retainedMessagesManager, sessionManager);
List<string> 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<string> 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
}
}
}
}
Source/MQTTnet.Tests/TopicGenerator.cs
+1 -2
View File
@@ -1,8 +1,7 @@
using System;
using System.Collections.Generic;
using System.Text;
namespace MQTTnet.Tests.Server
namespace MQTTnet.Tests
{
public class TopicGenerator
{
Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs
+4 -4
View File
@@ -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;
}
Source/MQTTnet/Internal/AsyncEvent.cs
+1 -1
View File
@@ -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)}.");
}
}
}
Source/MQTTnet/Server/Internal/MqttClient.cs
+21 -14
View File
@@ -33,6 +33,7 @@ namespace MQTTnet.Server
readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
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))
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+38 -25
View File
@@ -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);
}
}
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+25 -41
View File
@@ -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<string, MqttSubscription> _subscriptions = new Dictionary<string, MqttSubscription>();
// 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<ulong, TopicHashMaskSubscriptions> _wildcardSubscriptionsByTopicHash = new Dictionary<ulong, TopicHashMaskSubscriptions>();
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<MqttSubscription>();
// 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<uint> 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<MqttUnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
public async Task<UnsubscribeResult> 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<string>();
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<InterceptingUnsubscriptionEventArgs> 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; }
Source/MQTTnet/Server/Internal/MqttSession.cs
+37 -13
View File
@@ -23,6 +23,7 @@ namespace MQTTnet.Server
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
readonly MqttServerOptions _serverOptions;
readonly MqttClientSubscriptionsManager _subscriptionsManager;
readonly ConcurrentDictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new ConcurrentDictionary<ushort, MqttPublishPacket>();
@@ -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<SubscribeResult> 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<UnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
return _subscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken);
}
}
}
Source/MQTTnet/Server/Internal/MqttSubscription.cs
+1 -1
View File
@@ -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;
Source/MQTTnet/Server/Internal/MqttUnsubscribeResult.cs → Source/MQTTnet/Server/Internal/UnsubscribeResult.cs
+1 -1
View File
@@ -7,7 +7,7 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttUnsubscribeResult
public sealed class UnsubscribeResult
{
public List<MqttUnsubscribeReasonCode> ReasonCodes { get; } = new List<MqttUnsubscribeReasonCode>(128);