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

Merge branch 'master' into 1681-misleading-signature-of-mqttclientconnectasync-mqttclientconnectresultcode-expected-but-throws-exception

Browse Source
This commit is contained in:
Christian
2023-04-29 12:45:17 +02:00
parent 202e299f65 4b7d035e9e
commit af1c151e8e
41 changed files with 640 additions and 433 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
+21 -18
View File
@@ -2,38 +2,34 @@
// 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.Http;
using MQTTnet.Adapter;
using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Server;
using System;
using System.Net.WebSockets;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Server;
namespace MQTTnet.AspNetCore
{
public sealed class MqttWebSocketServerAdapter : IMqttServerAdapter
{
IMqttNetLogger _logger = new MqttNetNullLogger();
IMqttNetLogger _logger = MqttNetNullLogger.Instance;
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public Task StartAsync(MqttServerOptions options, IMqttNetLogger logger)
public void Dispose()
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
return Task.CompletedTask;
}
public Task StopAsync()
{
return Task.CompletedTask;
}
public async Task RunWebSocketConnectionAsync(WebSocket webSocket, HttpContext httpContext)
{
if (webSocket == null) throw new ArgumentNullException(nameof(webSocket));
if (webSocket == null)
{
throw new ArgumentNullException(nameof(webSocket));
}
var endpoint = $"{httpContext.Connection.RemoteIpAddress}:{httpContext.Connection.RemotePort}";
@@ -48,7 +44,7 @@ namespace MQTTnet.AspNetCore
var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535));
var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection, clientCertificate);
using (var channelAdapter = new MqttChannelAdapter(channel, formatter, null, _logger))
using (var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger))
{
await clientHandler(channelAdapter).ConfigureAwait(false);
}
@@ -60,8 +56,15 @@ namespace MQTTnet.AspNetCore
}
}
public void Dispose()
public Task StartAsync(MqttServerOptions options, IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
return Task.CompletedTask;
}
public Task StopAsync()
{
return Task.CompletedTask;
}
}
}
Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs
+1 -1
View File
@@ -73,7 +73,7 @@ namespace MQTTnet.Benchmarks
var channel = new MemoryMqttChannel(_stream);
_channelAdapter = new MqttChannelAdapter(channel, serializer, null, new MqttNetEventLogger());
_channelAdapter = new MqttChannelAdapter(channel, serializer, new MqttNetEventLogger());
}
static byte[] Join(params ArraySegment<byte>[] chunks)
Source/MQTTnet.Benchmarks/SerializerBenchmark.cs
+1 -1
View File
@@ -54,7 +54,7 @@ namespace MQTTnet.Benchmarks
public void Deserialize_10000_Messages()
{
var channel = new BenchmarkMqttChannel(_serializedPacket);
var reader = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535)), null, new MqttNetEventLogger());
var reader = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(new MqttBufferWriter(4096, 65535)), new MqttNetEventLogger());
for (var i = 0; i < 10000; i++)
{
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
+10 -5
View File
@@ -353,13 +353,13 @@ namespace MQTTnet.Extensions.ManagedClient
return remainingTime < TimeSpan.Zero ? TimeSpan.Zero : remainingTime;
}
async Task HandleSubscriptionExceptionAsync(Exception exception)
async Task HandleSubscriptionExceptionAsync(Exception exception, List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
{
_logger.Warning(exception, "Synchronizing subscriptions failed.");
if (_synchronizingSubscriptionsFailedEvent.HasHandlers)
{
await _synchronizingSubscriptionsFailedEvent.InvokeAsync(new ManagedProcessFailedEventArgs(exception)).ConfigureAwait(false);
await _synchronizingSubscriptionsFailedEvent.InvokeAsync(new ManagedProcessFailedEventArgs(exception, addedSubscriptions, removedSubscriptions)).ConfigureAwait(false);
}
}
@@ -457,11 +457,13 @@ namespace MQTTnet.Extensions.ManagedClient
{
_logger.Info("Publishing subscriptions at reconnect");
List<MqttTopicFilter> topicFilters = null;
try
{
if (_reconnectSubscriptions.Any())
{
var topicFilters = new List<MqttTopicFilter>();
topicFilters = new List<MqttTopicFilter>();
foreach (var sub in _reconnectSubscriptions)
{
@@ -479,7 +481,7 @@ namespace MQTTnet.Extensions.ManagedClient
}
catch (Exception exception)
{
await HandleSubscriptionExceptionAsync(exception).ConfigureAwait(false);
await HandleSubscriptionExceptionAsync(exception, topicFilters, null).ConfigureAwait(false);
}
}
@@ -591,6 +593,9 @@ namespace MQTTnet.Extensions.ManagedClient
}
await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
//clear because these worked, maybe the subscribe below will fail, only report those
removedSubscriptions.Clear();
}
if (addedSubscriptions != null && addedSubscriptions.Any())
@@ -607,7 +612,7 @@ namespace MQTTnet.Extensions.ManagedClient
}
catch (Exception exception)
{
await HandleSubscriptionExceptionAsync(exception).ConfigureAwait(false);
await HandleSubscriptionExceptionAsync(exception, addedSubscriptions, removedSubscriptions).ConfigureAwait(false);
}
}
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs
+1 -1
View File
@@ -24,7 +24,7 @@ namespace MQTTnet.Extensions.ManagedClient
/// <summary>
/// Defines the maximum amount of topic filters which will be sent in a SUBSCRIBE/UNSUBSCRIBE packet.
/// Amazon AWS limits this number to 8. The default is int.MaxValue.
/// Amazon Web Services (AWS) limits this number to 8. The default is int.MaxValue.
/// </summary>
public int MaxTopicFiltersInSubscribeUnsubscribePackets { get; set; } = int.MaxValue;
}
Source/MQTTnet.Extensions.ManagedClient/ManagedProcessFailedEventArgs.cs
+17 -1
View File
@@ -2,17 +2,33 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using MQTTnet.Packets;
using System;
using System.Collections.Generic;
using System.Linq;
namespace MQTTnet.Extensions.ManagedClient
{
public class ManagedProcessFailedEventArgs : EventArgs
{
public ManagedProcessFailedEventArgs(Exception exception)
public ManagedProcessFailedEventArgs(Exception exception, List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
{
Exception = exception ?? throw new ArgumentNullException(nameof(exception));
if (addedSubscriptions != null)
{
AddedSubscriptions = new List<string>(addedSubscriptions.Select(item => item.Topic));
}
if (removedSubscriptions != null)
{
RemovedSubscriptions = new List<string>(removedSubscriptions);
}
}
public Exception Exception { get; }
public List<string> AddedSubscriptions { get; }
public List<string> RemovedSubscriptions { get; }
}
}
Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs
+8 -4
View File
@@ -27,8 +27,10 @@ namespace MQTTnet.Extensions.WebSocket4Net
return new MqttChannelAdapter(
new MqttTcpChannel(options),
new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(options.WriterBufferSize, options.WriterBufferSizeMax)),
packetInspector,
logger);
logger)
{
PacketInspector = packetInspector
};
}
case MqttClientWebSocketOptions webSocketOptions:
@@ -36,8 +38,10 @@ namespace MQTTnet.Extensions.WebSocket4Net
return new MqttChannelAdapter(
new WebSocket4NetMqttChannel(options, webSocketOptions),
new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(options.WriterBufferSize, options.WriterBufferSizeMax)),
packetInspector,
logger);
logger)
{
PacketInspector = packetInspector
};
}
default:
Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs
+2
View File
@@ -99,6 +99,8 @@ namespace MQTTnet.Tests.Clients.MqttClient
// Perform a clean disconnect.
await client.DisconnectAsync(disconnectOptions);
await LongTestDelay();
Assert.IsNotNull(eventArgs);
Assert.AreEqual(MqttClientDisconnectType.Clean, eventArgs.DisconnectType);
}
Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs
+1 -1
View File
@@ -30,7 +30,7 @@ namespace MQTTnet.Tests.Diagnostics
var packets = new List<string>();
mqttClient.InspectPackageAsync += eventArgs =>
mqttClient.InspectPacketAsync += eventArgs =>
{
packets.Add(eventArgs.Direction + ":" + Convert.ToBase64String(eventArgs.Buffer));
return CompletedTask.Instance;
Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Binary_Tests.cs
+2 -10
View File
@@ -566,7 +566,6 @@ namespace MQTTnet.Tests.Formatter
using (var adapter = new MqttChannelAdapter(
channel,
new MqttPacketFormatterAdapter(protocolVersion, new MqttBufferWriter(4096, 65535)),
null,
new MqttNetEventLogger()))
{
var receivedPacket = adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
@@ -582,19 +581,12 @@ namespace MQTTnet.Tests.Formatter
MqttProtocolVersion DeserializeAndDetectVersion(MqttPacketFormatterAdapter packetFormatterAdapter, byte[] buffer)
{
var channel = new MemoryMqttChannel(buffer);
var adapter = new MqttChannelAdapter(channel, packetFormatterAdapter, null, new MqttNetEventLogger());
var adapter = new MqttChannelAdapter(channel, packetFormatterAdapter, new MqttNetEventLogger());
adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
return packetFormatterAdapter.ProtocolVersion;
}
MqttBufferReader ReaderFactory(byte[] data)
{
var reader = new MqttBufferReader();
reader.SetBuffer(data, 0, data.Length);
return reader;
}
TPacket Roundtrip<TPacket>(TPacket packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311, MqttBufferWriter bufferWriter = null) where TPacket : MqttPacket
{
var writer = bufferWriter ?? WriterFactory();
@@ -603,7 +595,7 @@ namespace MQTTnet.Tests.Formatter
using (var channel = new MemoryMqttChannel(buffer.Join().ToArray()))
{
var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, new MqttBufferWriter(4096, 65535)), null, new MqttNetEventLogger());
var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, new MqttBufferWriter(4096, 65535)), new MqttNetEventLogger());
return (TPacket)adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
}
}
Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs
+1 -1
View File
@@ -31,7 +31,7 @@ namespace MQTTnet.Tests
{
var formatterAdapter = new MqttPacketFormatterAdapter(_protocolVersion, new MqttBufferWriter(4096, 65535));
var adapter = new MqttChannelAdapter(channel, formatterAdapter, null, MqttNetNullLogger.Instance);
var adapter = new MqttChannelAdapter(channel, formatterAdapter, MqttNetNullLogger.Instance);
return adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
}
}
Source/MQTTnet.Tests/Server/General.cs
+1 -1
View File
@@ -954,7 +954,7 @@ namespace MQTTnet.Tests.Server
var client = testEnvironment.CreateClient();
client.InspectPackageAsync += e =>
client.InspectPacketAsync += e =>
{
if (e.Buffer.Length > 0)
{
Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs
+1 -1
View File
@@ -228,7 +228,7 @@ namespace MQTTnet.Tests.Server
CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId)
{
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out _, out _);
MqttTopicHash.Calculate(topic, out var topicHash, out _, out _);
return _subscriptionsManager.CheckSubscriptions(topic, topicHash, applicationMessageQoSLevel, senderClientId);
}
}
Source/MQTTnet.Tests/Server/Publishing_Tests.cs
+39
View File
@@ -87,6 +87,8 @@ namespace MQTTnet.Tests.Server
await sender.PublishStringAsync("A", "Payload", MqttQualityOfServiceLevel.AtLeastOnce);
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(1);
server.InterceptingClientEnqueueAsync += e =>
@@ -103,5 +105,42 @@ namespace MQTTnet.Tests.Server
receivedMessages.AssertReceivedCountEquals(1);
}
}
[TestMethod]
public async Task Intercept_Client_Enqueue_Multiple_Clients_Subscribed_Messages_Are_Filtered()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
var sender = await testEnvironment.ConnectClient();
var receiverOne = await testEnvironment.ConnectClient(o => o.WithClientId("One"));
await receiverOne.SubscribeAsync("A");
var receivedMessagesOne = testEnvironment.CreateApplicationMessageHandler(receiverOne);
var receiverTwo = await testEnvironment.ConnectClient(o => o.WithClientId("Two"));
await receiverTwo.SubscribeAsync("A");
var receivedMessagesTwo = testEnvironment.CreateApplicationMessageHandler(receiverTwo);
var receiverThree = await testEnvironment.ConnectClient(o => o.WithClientId("Three"));
await receiverThree.SubscribeAsync("A");
var receivedMessagesThree = testEnvironment.CreateApplicationMessageHandler(receiverThree);
server.InterceptingClientEnqueueAsync += e =>
{
if (e.ReceiverClientId.Contains("Two")) e.AcceptEnqueue = false;
return CompletedTask.Instance;
};
await sender.PublishStringAsync("A", "Payload", MqttQualityOfServiceLevel.AtLeastOnce);
await LongTestDelay();
receivedMessagesOne.AssertReceivedCountEquals(1);
receivedMessagesTwo.AssertReceivedCountEquals(0);
receivedMessagesThree.AssertReceivedCountEquals(1);
}
}
}
}
Source/MQTTnet.Tests/Server/Subscribe_Tests.cs
+25 -2
View File
@@ -62,7 +62,7 @@ namespace MQTTnet.Tests.Server
}
}
}
[TestMethod]
public async Task Deny_Invalid_Topic()
{
@@ -90,6 +90,29 @@ namespace MQTTnet.Tests.Server
}
}
[TestMethod]
public async Task Intercept_Subscribe_With_User_Properties()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
var server = await testEnvironment.StartServer();
InterceptingSubscriptionEventArgs eventArgs = null;
server.InterceptingSubscriptionAsync += e =>
{
eventArgs = e;
return CompletedTask.Instance;
};
var client = await testEnvironment.ConnectClient();
var subscribeOptions = testEnvironment.Factory.CreateSubscribeOptionsBuilder().WithTopicFilter("X").WithUserProperty("A", "1").Build();
await client.SubscribeAsync(subscribeOptions);
CollectionAssert.AreEqual(subscribeOptions.UserProperties.ToList(), eventArgs.UserProperties);
}
}
[TestMethod]
[ExpectedException(typeof(MqttClientDisconnectedException))]
public async Task Disconnect_While_Subscribing()
@@ -105,7 +128,7 @@ namespace MQTTnet.Tests.Server
await client.SubscribeAsync("#");
}
}
[TestMethod]
public async Task Enqueue_Message_After_Subscription()
{
Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs
+11 -11
View File
@@ -47,7 +47,7 @@ namespace MQTTnet.Tests.Server
var topics = t.Value;
foreach (var topic in topics)
{
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var hashMask, out var hasWildcard);
bucketDepths.TryGetValue(topicHash, out var currentValue);
++currentValue;
@@ -96,7 +96,7 @@ namespace MQTTnet.Tests.Server
{
var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/#";
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
}
@@ -110,7 +110,7 @@ namespace MQTTnet.Tests.Server
{
var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/+";
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
}
@@ -150,7 +150,7 @@ namespace MQTTnet.Tests.Server
// UInt64 is limited to 8 levels
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
@@ -212,7 +212,7 @@ namespace MQTTnet.Tests.Server
var topic = sb.ToString();
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsFalse(topicHasWildcard, "Wildcard detected when not present");
@@ -269,7 +269,7 @@ namespace MQTTnet.Tests.Server
var topic = sb.ToString();
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
@@ -319,7 +319,7 @@ namespace MQTTnet.Tests.Server
var l3 = "#";
var topic = $"{l0}/{l1}/{l2}/{l3}";
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
@@ -354,7 +354,7 @@ namespace MQTTnet.Tests.Server
var l3 = "prop1";
var topic = $"{l0}/{l1}/{l2}/{l3}";
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsFalse(topicHasWildcard, "Wildcard detected when not wildcard present");
@@ -447,7 +447,7 @@ namespace MQTTnet.Tests.Server
var l3 = "prop1";
var topic = $"{l0}/{l1}/{l2}/{l3}";
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
@@ -475,7 +475,7 @@ namespace MQTTnet.Tests.Server
void CheckTopicHash(string topic, ulong expectedHash, ulong expectedHashMask)
{
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out var hashMask, out var hasWildcard);
Console.WriteLine();
Console.WriteLine("Topic: " + topic);
@@ -501,7 +501,7 @@ namespace MQTTnet.Tests.Server
{
var topicToFind = topicsToFind[countDown];
MqttSubscription.CalculateTopicHash(topicToFind, out var topicHash, out _, out _);
MqttTopicHash.Calculate(topicToFind, out var topicHash, out _, out _);
clientSession.TryCheckSubscriptions(topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient", out var result);
if (result.IsSubscribed)
Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs
+27
View File
@@ -2,10 +2,14 @@
// 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.Linq;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Server
{
@@ -28,5 +32,28 @@ namespace MQTTnet.Tests.Server
await client.UnsubscribeAsync("#");
}
}
[TestMethod]
public async Task Intercept_Unsubscribe_With_User_Properties()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
var server = await testEnvironment.StartServer();
InterceptingUnsubscriptionEventArgs eventArgs = null;
server.InterceptingUnsubscriptionAsync += e =>
{
eventArgs = e;
return CompletedTask.Instance;
};
var client = await testEnvironment.ConnectClient();
var unsubscribeOptions = testEnvironment.Factory.CreateUnsubscribeOptionsBuilder().WithTopicFilter("X").WithUserProperty("A", "1").Build();
await client.UnsubscribeAsync(unsubscribeOptions);
CollectionAssert.AreEqual(unsubscribeOptions.UserProperties.ToList(), eventArgs.UserProperties);
}
}
}
}
Source/MQTTnet.Tests/TopicFilterComparer_Tests.cs
+2 -12
View File
@@ -150,18 +150,8 @@ namespace MQTTnet.Tests
{
Assert.AreEqual(expectedResult, MqttTopicFilterComparer.Compare(topic, filter));
ulong topicHash;
ulong topicHashMask;
bool topicHasWildcard;
MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
ulong filterTopicHash;
ulong filterTopicHashMask;
bool filterTopicHasWildcard;
MqttSubscription.CalculateTopicHash(filter, out filterTopicHash, out filterTopicHashMask, out filterTopicHasWildcard);
MqttTopicHash.Calculate(topic, out var topicHash, out _, out _);
MqttTopicHash.Calculate(filter, out var filterTopicHash, out var filterTopicHashMask, out _);
if (expectedResult == MqttTopicFilterCompareResult.IsMatch)
{
Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+23 -19
View File
@@ -27,18 +27,15 @@ namespace MQTTnet.Adapter
readonly byte[] _fixedHeaderBuffer = new byte[2];
readonly MqttNetSourceLogger _logger;
readonly MqttPacketInspector _packetInspector;
readonly byte[] _singleByteBuffer = new byte[1];
readonly AsyncLock _syncRoot = new AsyncLock();
Statistics _statistics; // mutable struct, don't make readonly!
Statistics _statistics; // mutable struct, don't make readonly!
public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, MqttPacketInspector packetInspector, IMqttNetLogger logger)
public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetLogger logger)
{
_channel = channel ?? throw new ArgumentNullException(nameof(channel));
_packetInspector = packetInspector;
PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
@@ -64,6 +61,10 @@ namespace MQTTnet.Adapter
public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
public MqttPacketInspector PacketInspector { get; set; }
public bool AllowPacketFragmentation { get; set; } = true;
public async Task ConnectAsync(CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
@@ -126,7 +127,7 @@ namespace MQTTnet.Adapter
try
{
_packetInspector?.BeginReceivePacket();
PacketInspector?.BeginReceivePacket();
ReceivedMqttPacket receivedPacket;
var receivedPacketTask = ReceiveAsync(cancellationToken);
@@ -144,7 +145,7 @@ namespace MQTTnet.Adapter
return null;
}
_packetInspector?.EndReceivePacket();
PacketInspector?.EndReceivePacket();
Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength);
@@ -180,7 +181,10 @@ namespace MQTTnet.Adapter
return null;
}
public void ResetStatistics() => _statistics.Reset();
public void ResetStatistics()
{
_statistics.Reset();
}
public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
{
@@ -197,18 +201,18 @@ namespace MQTTnet.Adapter
try
{
var packetBuffer = PacketFormatterAdapter.Encode(packet);
_packetInspector?.BeginSendPacket(packetBuffer);
PacketInspector?.BeginSendPacket(packetBuffer);
_logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
if (packetBuffer.Payload.Count > 0)
if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation)
{
await _channel.WriteAsync(packetBuffer.Packet, false, cancellationToken).ConfigureAwait(false);
await _channel.WriteAsync(packetBuffer.Payload, true, cancellationToken).ConfigureAwait(false);
await _channel.WriteAsync(packetBuffer.Join(), true, cancellationToken).ConfigureAwait(false);
}
else
{
await _channel.WriteAsync(packetBuffer.Packet, true, cancellationToken).ConfigureAwait(false);
await _channel.WriteAsync(packetBuffer.Packet, false, cancellationToken).ConfigureAwait(false);
await _channel.WriteAsync(packetBuffer.Payload, true, cancellationToken).ConfigureAwait(false);
}
Interlocked.Add(ref _statistics._bytesReceived, packetBuffer.Length);
@@ -270,7 +274,7 @@ namespace MQTTnet.Adapter
return 0;
}
_packetInspector?.FillReceiveBuffer(_singleByteBuffer);
PacketInspector?.FillReceiveBuffer(_singleByteBuffer);
encodedByte = _singleByteBuffer[0];
@@ -309,7 +313,7 @@ namespace MQTTnet.Adapter
{
return ReadFixedHeaderResult.ConnectionClosed;
}
if (cancellationToken.IsCancellationRequested)
{
return ReadFixedHeaderResult.Canceled;
@@ -323,7 +327,7 @@ namespace MQTTnet.Adapter
totalBytesRead += bytesRead;
}
_packetInspector?.FillReceiveBuffer(buffer);
PacketInspector?.FillReceiveBuffer(buffer);
var hasRemainingLength = buffer[1] != 0;
if (!hasRemainingLength)
@@ -407,7 +411,7 @@ namespace MQTTnet.Adapter
bodyOffset += readBytes;
} while (bodyOffset < bodyLength);
_packetInspector?.FillReceiveBuffer(body);
PacketInspector?.FillReceiveBuffer(body);
var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
@@ -455,7 +459,7 @@ namespace MQTTnet.Adapter
throw new MqttCommunicationException(exception);
}
private struct Statistics
struct Statistics
{
public long _bytesReceived;
public long _bytesSent;
Source/MQTTnet/Client/IMqttClient.cs
+1 -1
View File
@@ -15,7 +15,7 @@ namespace MQTTnet.Client
event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync;
event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync;
event Func<InspectMqttPacketEventArgs, Task> InspectPacketAsync;
bool IsConnected { get; }
Source/MQTTnet/Client/MqttClient.cs
+1 -1
View File
@@ -98,7 +98,7 @@ namespace MQTTnet.Client
remove => _authenticationHandler.ExtendedAuthenticationExchangeEvent.RemoveHandler(value);
}
public event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync
public event Func<InspectMqttPacketEventArgs, Task> InspectPacketAsync
{
add => _inspectPacketEvent.AddHandler(value);
remove => _inspectPacketEvent.RemoveHandler(value);
Source/MQTTnet/Client/Options/MqttClientOptions.cs
+9 -1
View File
@@ -24,6 +24,14 @@ namespace MQTTnet.Client
/// </summary>
public string AuthenticationMethod { get; set; }
/// <summary>
/// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
/// or WebSocket frames etc. Unfortunately not all brokers (like Amazon Web Services (AWS)) do support this feature and
/// will close the connection when receiving such packets. If such a service is used this flag must
/// be set to _false_.
/// </summary>
public bool AllowPacketFragmentation { get; set; } = true;
public IMqttClientChannelOptions ChannelOptions { get; set; }
/// <summary>
@@ -214,4 +222,4 @@ namespace MQTTnet.Client
/// </summary>
public int WriterBufferSizeMax { get; set; } = 65535;
}
}
}
Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs
+53 -29
View File
@@ -75,7 +75,7 @@ namespace MQTTnet.Client
_options.ChannelOptions = (IMqttClientChannelOptions)_tcpOptions ?? _webSocketOptions;
MqttClientOptionsValidator.ThrowIfNotSupported(_options);
return _options;
}
@@ -86,12 +86,24 @@ namespace MQTTnet.Client
return this;
}
/// <summary>
/// Clean session is used in MQTT versions below 5.0.0. It is the same as setting "CleanStart".
/// </summary>
public MqttClientOptionsBuilder WithCleanSession(bool value = true)
{
_options.CleanSession = value;
return this;
}
/// <summary>
/// Clean start is used in MQTT versions 5.0.0 and higher. It is the same as setting "CleanSession".
/// </summary>
public MqttClientOptionsBuilder WithCleanStart(bool value = true)
{
_options.CleanSession = value;
return this;
}
public MqttClientOptionsBuilder WithClientId(string value)
{
_options.ClientId = value;
@@ -188,6 +200,18 @@ namespace MQTTnet.Client
return WithKeepAlivePeriod(TimeSpan.Zero);
}
/// <summary>
/// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
/// or WebSocket frames etc. Unfortunately not all brokers (like Amazon Web Services (AWS)) do support this feature and
/// will close the connection when receiving such packets. If such a service is used this flag must
/// be set to _true_.
/// </summary>
public MqttClientOptionsBuilder WithoutPacketFragmentation()
{
_options.AllowPacketFragmentation = false;
return this;
}
public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
{
if (value == MqttProtocolVersion.Unknown)
@@ -266,7 +290,7 @@ namespace MQTTnet.Client
return this;
}
public MqttClientOptionsBuilder WithTcpServer(Action<MqttClientTcpOptions> optionsBuilder)
{
if (optionsBuilder == null)
@@ -373,6 +397,18 @@ namespace MQTTnet.Client
return this;
}
public MqttClientOptionsBuilder WithWillContentType(string willContentType)
{
_options.WillContentType = willContentType;
return this;
}
public MqttClientOptionsBuilder WithWillCorrelationData(byte[] willCorrelationData)
{
_options.WillCorrelationData = willCorrelationData;
return this;
}
public MqttClientOptionsBuilder WithWillDelayInterval(uint willDelayInterval)
{
_options.WillDelayInterval = willDelayInterval;
@@ -384,27 +420,33 @@ namespace MQTTnet.Client
_options.WillPayload = willPayload;
return this;
}
public MqttClientOptionsBuilder WithWillPayload(string willPayload)
{
if (string.IsNullOrEmpty(willPayload))
{
return WithWillPayload((byte[])null);
}
_options.WillPayload = Encoding.UTF8.GetBytes(willPayload);
return this;
}
public MqttClientOptionsBuilder WithWillPayloadFormatIndicator(MqttPayloadFormatIndicator willPayloadFormatIndicator)
{
_options.WillPayloadFormatIndicator = willPayloadFormatIndicator;
return this;
}
public MqttClientOptionsBuilder WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel willQualityOfServiceLevel)
{
_options.WillQualityOfServiceLevel = willQualityOfServiceLevel;
return this;
}
public MqttClientOptionsBuilder WithWillTopic(string willTopic)
public MqttClientOptionsBuilder WithWillResponseTopic(string willResponseTopic)
{
_options.WillTopic = willTopic;
_options.WillResponseTopic = willResponseTopic;
return this;
}
@@ -413,38 +455,20 @@ namespace MQTTnet.Client
_options.WillRetain = willRetain;
return this;
}
public MqttClientOptionsBuilder WithWillContentType(string willContentType)
public MqttClientOptionsBuilder WithWillTopic(string willTopic)
{
_options.WillContentType = willContentType;
_options.WillTopic = willTopic;
return this;
}
public MqttClientOptionsBuilder WithWillCorrelationData(byte[] willCorrelationData)
{
_options.WillCorrelationData = willCorrelationData;
return this;
}
public MqttClientOptionsBuilder WithWillResponseTopic(string willResponseTopic)
{
_options.WillResponseTopic = willResponseTopic;
return this;
}
public MqttClientOptionsBuilder WithWillPayloadFormatIndicator(MqttPayloadFormatIndicator willPayloadFormatIndicator)
{
_options.WillPayloadFormatIndicator = willPayloadFormatIndicator;
return this;
}
public MqttClientOptionsBuilder WithWillUserProperty(string name, string value)
{
if (_options.WillUserProperties == null)
{
_options.WillUserProperties = new List<MqttUserProperty>();
}
_options.WillUserProperties.Add(new MqttUserProperty(name, value));
return this;
}
Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs
+1 -1
View File
@@ -42,6 +42,6 @@ namespace MQTTnet.Client
/// Gets the user properties which were part of the UNSUBACK packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; }
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; set; }
}
}
Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs
+6 -1
View File
@@ -40,7 +40,12 @@ namespace MQTTnet.Implementations
var bufferWriter = new MqttBufferWriter(options.WriterBufferSize, options.WriterBufferSizeMax);
var packetFormatterAdapter = new MqttPacketFormatterAdapter(options.ProtocolVersion, bufferWriter);
return new MqttChannelAdapter(channel, packetFormatterAdapter, packetInspector, logger);
return new MqttChannelAdapter(channel, packetFormatterAdapter, logger)
{
AllowPacketFragmentation = options.AllowPacketFragmentation,
PacketInspector = packetInspector
};
}
}
}
Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs
+1 -1
View File
@@ -95,7 +95,7 @@ namespace MQTTnet.Implementations
var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter);
var tcpChannel = new MqttTcpChannel(args.Socket, clientCertificate, _options);
using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, null, _rootLogger))
using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, _rootLogger))
{
await clientHandler(clientAdapter).ConfigureAwait(false);
}
Source/MQTTnet/Implementations/MqttTcpServerListener.cs
+1 -1
View File
@@ -229,7 +229,7 @@ namespace MQTTnet.Implementations
var bufferWriter = new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax);
var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter);
using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, null, _rootLogger))
using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, _rootLogger))
{
await clientHandler(clientAdapter).ConfigureAwait(false);
}
Source/MQTTnet/Implementations/MqttWebSocketChannel.cs
+2 -2
View File
@@ -63,7 +63,7 @@ namespace MQTTnet.Implementations
await clientWebSocket.ConnectAsync(new Uri(uri), cancellationToken).ConfigureAwait(false);
}
catch (Exception)
catch
{
// Prevent a memory leak when always creating new instance which will fail while connecting.
clientWebSocket.Dispose();
@@ -103,7 +103,7 @@ namespace MQTTnet.Implementations
public async Task WriteAsync(ArraySegment<byte> buffer, bool isEndOfPacket, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
#if NET5_0_OR_GREATER
// MQTT Control Packets MUST be sent in WebSocket binary data frames. If any other type of data frame is received the recipient MUST close the Network Connection [MQTT-6.0.0-1].
// A single WebSocket data frame can contain multiple or partial MQTT Control Packets. The receiver MUST NOT assume that MQTT Control Packets are aligned on WebSocket frame boundaries [MQTT-6.0.0-2].
Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs
+1 -1
View File
@@ -9,7 +9,7 @@ namespace MQTTnet.LowLevelClient
{
public interface ILowLevelMqttClient : IDisposable
{
event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync;
event Func<InspectMqttPacketEventArgs, Task> InspectPacketAsync;
bool IsConnected { get; }
Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs
+10 -4
View File
@@ -32,7 +32,7 @@ namespace MQTTnet.LowLevelClient
_logger = logger.WithSource(nameof(LowLevelMqttClient));
}
public event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync
public event Func<InspectMqttPacketEventArgs, Task> InspectPacketAsync
{
add => _inspectPacketEvent.AddHandler(value);
remove => _inspectPacketEvent.RemoveHandler(value);
@@ -52,13 +52,19 @@ namespace MQTTnet.LowLevelClient
throw new InvalidOperationException("Low level MQTT client is already connected. Disconnect first before connecting again.");
}
var newAdapter = _clientAdapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_inspectPacketEvent, _rootLogger), _rootLogger);
MqttPacketInspector packetInspector = null;
if (_inspectPacketEvent.HasHandlers)
{
packetInspector = new MqttPacketInspector(_inspectPacketEvent, _rootLogger);
}
var newAdapter = _clientAdapterFactory.CreateClientAdapter(options, packetInspector, _rootLogger);
try
{
_logger.Verbose("Trying to connect with server '{0}'.", options.ChannelOptions);
_logger.Verbose("Trying to connect with server '{0}'", options.ChannelOptions);
await newAdapter.ConnectAsync(cancellationToken).ConfigureAwait(false);
_logger.Verbose("Connection with server established.");
_logger.Verbose("Connection with server established");
}
catch (Exception)
{
Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs
+4 -1
View File
@@ -16,12 +16,14 @@ namespace MQTTnet.Server
CancellationToken cancellationToken,
string clientId,
MqttSessionStatus session,
MqttTopicFilter topicFilter)
MqttTopicFilter topicFilter,
List<MqttUserProperty> userProperties)
{
CancellationToken = cancellationToken;
ClientId = clientId;
Session = session;
TopicFilter = topicFilter;
UserProperties = userProperties;
}
/// <summary>
@@ -75,6 +77,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets or sets the user properties.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> UserProperties { get; set; }
}
Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs
+10 -1
View File
@@ -4,18 +4,21 @@
using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class InterceptingUnsubscriptionEventArgs : EventArgs
{
public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, IDictionary sessionItems, string topic)
public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, IDictionary sessionItems, string topic, List<MqttUserProperty> userProperties)
{
CancellationToken = cancellationToken;
ClientId = clientId;
SessionItems = sessionItems;
Topic = topic;
UserProperties = userProperties;
}
/// <summary>
@@ -59,5 +62,11 @@ namespace MQTTnet.Server
/// separator).
/// </summary>
public string Topic { get; }
/// <summary>
/// Gets or sets the user properties.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> UserProperties { get; set; }
}
}
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+3 -4
View File
@@ -179,7 +179,7 @@ namespace MQTTnet.Server
}
// Calculate application message topic hash once for subscription checks
MqttSubscription.CalculateTopicHash(applicationMessage.Topic, out var topicHash, out _, out _);
MqttTopicHash.Calculate(applicationMessage.Topic, out var topicHash, out _, out _);
foreach (var session in subscriberSessions)
{
@@ -207,9 +207,8 @@ namespace MQTTnet.Server
if (!eventArgs.AcceptEnqueue)
{
// There will be no reason string and use properties because in this case the clients will
// not receive a packet at all.
return new DispatchApplicationMessageResult(reasonCode, eventArgs.CloseSenderConnection, null, null);
// Continue checking the other subscriptions
continue;
}
}
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+26 -18
View File
@@ -177,15 +177,15 @@ namespace MQTTnet.Server
// lower one.
foreach (var topicFilterItem in subscribePacket.TopicFilters.OrderByDescending(f => f.QualityOfServiceLevel))
{
var subscriptionEventArgs = await InterceptSubscribe(topicFilterItem, cancellationToken).ConfigureAwait(false);
var topicFilter = subscriptionEventArgs.TopicFilter;
var processSubscription = subscriptionEventArgs.ProcessSubscription && subscriptionEventArgs.Response.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
var interceptorEventArgs = await InterceptSubscribe(topicFilterItem, subscribePacket.UserProperties, cancellationToken).ConfigureAwait(false);
var topicFilter = interceptorEventArgs.TopicFilter;
var processSubscription = interceptorEventArgs.ProcessSubscription && interceptorEventArgs.Response.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
result.UserProperties = subscriptionEventArgs.UserProperties;
result.ReasonString = subscriptionEventArgs.ReasonString;
result.ReasonCodes.Add(subscriptionEventArgs.Response.ReasonCode);
result.UserProperties = interceptorEventArgs.UserProperties;
result.ReasonString = interceptorEventArgs.ReasonString;
result.ReasonCodes.Add(interceptorEventArgs.Response.ReasonCode);
if (subscriptionEventArgs.CloseConnection)
if (interceptorEventArgs.CloseConnection)
{
// When any of the interceptor calls leads to a connection close the connection
// must be closed. So do not revert to false!
@@ -197,7 +197,7 @@ namespace MQTTnet.Server
continue;
}
var createSubscriptionResult = CreateSubscription(topicFilter, subscribePacket.SubscriptionIdentifier, subscriptionEventArgs.Response.ReasonCode);
var createSubscriptionResult = CreateSubscription(topicFilter, subscribePacket.SubscriptionIdentifier, interceptorEventArgs.Response.ReasonCode);
addedSubscriptions.Add(topicFilter.Topic);
finalTopicFilters.Add(topicFilter);
@@ -238,12 +238,13 @@ namespace MQTTnet.Server
{
_subscriptions.TryGetValue(topicFilter, out var existingSubscription);
var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription, cancellationToken).ConfigureAwait(false);
var acceptUnsubscription = interceptorContext.Response.ReasonCode == MqttUnsubscribeReasonCode.Success;
var interceptorEventArgs = await InterceptUnsubscribe(topicFilter, existingSubscription, unsubscribePacket.UserProperties, cancellationToken).ConfigureAwait(false);
var acceptUnsubscription = interceptorEventArgs.Response.ReasonCode == MqttUnsubscribeReasonCode.Success;
result.ReasonCodes.Add(interceptorContext.Response.ReasonCode);
result.UserProperties = interceptorEventArgs.UserProperties;
result.ReasonCodes.Add(interceptorEventArgs.Response.ReasonCode);
if (interceptorContext.CloseConnection)
if (interceptorEventArgs.CloseConnection)
{
// When any of the interceptor calls leads to a connection close the connection
// must be closed. So do not revert to false!
@@ -255,7 +256,7 @@ namespace MQTTnet.Server
continue;
}
if (interceptorContext.ProcessUnsubscription)
if (interceptorEventArgs.ProcessUnsubscription)
{
_subscriptions.Remove(topicFilter);
@@ -342,7 +343,7 @@ namespace MQTTnet.Server
using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult())
{
MqttSubscription.CalculateTopicHash(topicFilter.Topic, out var topicHash, out _, out var hasWildcard);
MqttTopicHash.Calculate(topicFilter.Topic, out var topicHash, out _, out var hasWildcard);
if (_subscriptions.TryGetValue(topicFilter.Topic, out var existingSubscription))
{
@@ -452,9 +453,12 @@ namespace MQTTnet.Server
}
}
async Task<InterceptingSubscriptionEventArgs> InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken)
async Task<InterceptingSubscriptionEventArgs> InterceptSubscribe(
MqttTopicFilter topicFilter,
List<MqttUserProperty> userProperties,
CancellationToken cancellationToken)
{
var eventArgs = new InterceptingSubscriptionEventArgs(cancellationToken, _session.Id, new MqttSessionStatus(_session), topicFilter);
var eventArgs = new InterceptingSubscriptionEventArgs(cancellationToken, _session.Id, new MqttSessionStatus(_session), topicFilter, userProperties);
if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{
@@ -481,9 +485,13 @@ namespace MQTTnet.Server
return eventArgs;
}
async Task<InterceptingUnsubscriptionEventArgs> InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
async Task<InterceptingUnsubscriptionEventArgs> InterceptUnsubscribe(
string topicFilter,
MqttSubscription mqttSubscription,
List<MqttUserProperty> userProperties,
CancellationToken cancellationToken)
{
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, _session.Id, _session.Items, topicFilter)
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, _session.Id, _session.Items, topicFilter, userProperties)
{
Response =
{
Source/MQTTnet/Server/Internal/MqttSubscription.cs
+1 -260
View File
@@ -6,163 +6,6 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
/*
* The MqttSubscription object stores subscription parameters and calculates
* topic hashes.
*
* Use of Topic Hashes to improve message processing performance
* =============================================================
*
* Motivation
* -----------
* In a typical use case for MQTT there may be many publishers (sensors or
* other devices in the field) and few subscribers (monitoring all or many topics).
* Each publisher may have one or more topic(s) to publish and therefore both, the
* number of publishers and the number of topics may be large.
*
* Maintaining subscribers in a separate container
* -----------------------------------------------
* Instead of placing all sessions into a single _sessions container, subscribers
* are added into another _subscriberSessions container (if a client is a
* subscriber and a publisher then the client is present in both containers). The
* cost is some additional bookkeeping work upon subscription where each client
* session needs to maintain a list of subscribed topics.
*
* When the client subscribes to the first topic, then the session manager adds
* the client to the _subscriberSessions container, and when the client
* unsubscribes from the last topic then the session manager removes the client
* from the container. Now, when an application message arrives, only the list of
* subscribers need processing instead of looping through potentially thousands of
* publishers.
*
* Improving subscriber topic lookup
* ---------------------------------
* For each subscriber, it needs to be determined whether an application message
* matches any topic the subscriber has subscribed to. There may only be few
* subscribers but there may be many subscribed topics, including wildcard topics.
*
* The implemented approach uses a topic hash and a hash mask calculated on the
* subscribed topic and the published topic (the application message topic) to
* find candidates for a match, with the existing match logic evaluating a reduced
* number of candidates.
*
* For each subscription, the topic hash and a hash mask is stored with the
* subscription, and for each application message received, the hash is calculated
* for the published topic before attempting to find matching subscriptions. The
* hash calculation itself is simple and does not have a large performance impact.
*
* We'll first explain how topic hashes and hash masks are constructed and then how
* they are used.
*
* Topic hash
* ----------
* Topic hashes are stored as 64-bit numbers. Each byte within the 64-bit number
* relates to one MQTT topic level. A checksum is calculated for each topic level
* by iterating over the characters within the topic level (cast to byte) and the
* result is stored into the corresponding byte of the 64-bit number. If a topic
* level contains a wildcard character, then 0x00 is stored instead of the
* checksum.
*
* If there are less than 8 levels then the rest of the 64-bit number is filled
* with 0xff. If there are more than 8 levels then topics where the first 8 MQTT
* topic levels are identical will have the same hash value.
*
* This is the topic hash for the MQTT topic below: 0x655D4AF1FFFFFF
*
* client1/building1/level1/sensor1 (empty) (empty) (empty) (empty)
* \_____/ \_______/ \____/ \_____/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0x65 0x5D 0x4A 0xF1 0xFF 0xFF 0xFF 0xFF
*
* This is the topic hash for an MQTT topic containing a wildcard: 0x655D00F1FFFFFF
*
* client1/building1/ + /sensor1 (empty) (empty) (empty) (empty)
* \_____/ \_______/ \_/ \_____/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0x65 0x5D 0 0xF1 0xFF 0xFF 0xFF 0xFF
*
* For topics that contain the multi level wildcard # at the end, the topic hash
* is filled with 0x00: 0x65004A00000000
*
* client1/ + /level1/ # (empty) (empty) (empty) (empty)
* \_____/ \_/ \____/ \_/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0x65 0 0x4A 0 0 0 0 0
*
*
* Topic hash mask
* ---------------
* The hash mask simply contains 0xFF for non-wildcard topic levels and 0x00 for
* wildcard topic levels. Here are the topic hash masks for the examples above.
*
* client1/building1/level1/sensor1 (empty) (empty) (empty) (empty)
* \_____/ \_______/ \____/ \_____/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF
*
* client1/building1/ + /sensor1 (empty) (empty) (empty) (empty)
* \_____/ \_______/ \_/ \_____/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0xFF 0xFF 0 0xFF 0xFF 0xFF 0xFF 0xFF
*
* client1/ + /level1/ # (empty) (empty) (empty) (empty)
* \_____/ \_/ \____/ \_/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0xFF 0 0xFF 0 0 0 0 0
*
*
* Topic hash and hash mask properties
* -----------------------------------
* The following properties of topic hashes and hash masks can be exploited to
* find potentially matching subscribed topics for a given published topic.
*
* (1) If a subscribed topic does not contain wildcards then the topic hash of the
* subscribed topic must be equal to the topic hash of the published topic,
* otherwise the subscribed topic cannot be a candidate for a match.
*
* (2) If a subscribed topic contains wildcards then the hash of the published
* topic masked with the subscribed topic's hash mask must be equal to the hash of
* the subscribed topic. I.e. a subscribed topic is a candidate for a match if:
* (publishedTopicHash & subscribedTopicHashMask) == subscribedTopicHash
*
* (3) If a subscribed topic contains wildcards then any potentially matching
* published topic must have a hash value that is greater than or equal to the
* hash value of the subscribed topic (because the subscribed topic contains
* zeroes in wildcard positions).
*
* Match finding
* -------------
* The subscription manager maintains two separate dictionaries to assist finding
* matches using topic hashes: a _noWildcardSubscriptionsByTopicHash dictionary
* containing all subscriptions that do not have wildcards, and a
* _wildcardSubscriptionsByTopicHash dictionary containing subscriptions with
* wildcards.
*
* For subscriptions without wildcards, all potential candidates for a match are
* obtained by a single look-up (exploiting point 1 above).
*
* For subscriptions with wildcards, the subscription manager loops through the
* wildcard subscriptions and selects candidates that satisfy condition
* (publishedTopicHash & subscribedTopicMask) == subscribedTopicHash (point 2).
* The loop could exit early if wildcard subscriptions were stored into a sorted
* dictionary (utilizing point 3), but, after testing, there does not seem to be
* any real benefit doing so.
*
* Other considerations
* --------------------
* Characters in the topic string are cast to byte and any additional bytes in a
* multi-byte character are disregarded. Best guess is that this does not impact
* performance in practice.
*
* Instead of one-byte checksums per topic level, one-word checksums per topic
* level could be used. If most topics contained four levels or less then hash
* buckets would be shallower.
*
* For very large numbers of topics, performing a parallel search may help further.
*
* To also handle a larger number of subscribers, it may be beneficial to maintain
* a subscribers-by-subscription-topic dictionary.
*/
public sealed class MqttSubscription
{
public MqttSubscription(
@@ -180,7 +23,7 @@ namespace MQTTnet.Server
GrantedQualityOfServiceLevel = qualityOfServiceLevel;
Identifier = identifier;
CalculateTopicHash(Topic, out var hash, out var hashMask, out var hasWildcard);
MqttTopicHash.Calculate(Topic, out var hash, out var hashMask, out var hasWildcard);
TopicHash = hash;
TopicHashMask = hashMask;
TopicHasWildcard = hasWildcard;
@@ -203,107 +46,5 @@ namespace MQTTnet.Server
public ulong TopicHashMask { get; }
public bool TopicHasWildcard { get; }
public static void CalculateTopicHash(string topic, out ulong resultHash, out ulong resultHashMask, out bool resultHasWildcard)
{
// calculate topic hash
ulong hash = 0;
ulong hashMaskInverted = 0;
ulong levelBitMask = 0;
ulong fillLevelBitMask = 0;
var hasWildcard = false;
byte checkSum = 0;
var level = 0;
var i = 0;
while (i < topic.Length)
{
var c = topic[i];
if (c == MqttTopicFilterComparer.LevelSeparator)
{
// done with this level
hash <<= 8;
hash |= checkSum;
hashMaskInverted <<= 8;
hashMaskInverted |= levelBitMask;
checkSum = 0;
levelBitMask = 0;
++level;
if (level >= 8)
{
break;
}
}
else if (c == MqttTopicFilterComparer.SingleLevelWildcard)
{
levelBitMask = 0xff;
hasWildcard = true;
}
else if (c == MqttTopicFilterComparer.MultiLevelWildcard)
{
// checksum is zero for a valid topic
levelBitMask = 0xff;
// fill rest with this fillLevelBitMask
fillLevelBitMask = 0xff;
hasWildcard = true;
break;
}
else
{
// The checksum should be designed to reduce the hash bucket depth for the expected
// fairly regularly named MQTT topics that don't differ much,
// i.e. "room1/sensor1"
// "room1/sensor2"
// "room1/sensor3"
// etc.
if ((c & 1) == 0)
{
checkSum += (byte)c;
}
else
{
checkSum ^= (byte)(c >> 1);
}
}
++i;
}
// Shift hash left and leave zeroes to fill ulong
if (level < 8)
{
hash <<= 8;
hash |= checkSum;
hashMaskInverted <<= 8;
hashMaskInverted |= levelBitMask;
++level;
while (level < 8)
{
hash <<= 8;
hashMaskInverted <<= 8;
hashMaskInverted |= fillLevelBitMask;
++level;
}
}
if (!hasWildcard)
{
while (i < topic.Length)
{
var c = topic[i];
if (c == MqttTopicFilterComparer.SingleLevelWildcard || c == MqttTopicFilterComparer.MultiLevelWildcard)
{
hasWildcard = true;
break;
}
++i;
}
}
resultHash = hash;
resultHashMask = ~hashMaskInverted;
resultHasWildcard = hasWildcard;
}
}
}
Source/MQTTnet/Server/Internal/MqttTopicHash.cs
+268
View File
@@ -0,0 +1,268 @@
// 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.
namespace MQTTnet.Server
{
/*
* The MqttSubscription object stores subscription parameters and calculates
* topic hashes.
*
* Use of Topic Hashes to improve message processing performance
* =============================================================
*
* Motivation
* -----------
* In a typical use case for MQTT there may be many publishers (sensors or
* other devices in the field) and few subscribers (monitoring all or many topics).
* Each publisher may have one or more topic(s) to publish and therefore both, the
* number of publishers and the number of topics may be large.
*
* Maintaining subscribers in a separate container
* -----------------------------------------------
* Instead of placing all sessions into a single _sessions container, subscribers
* are added into another _subscriberSessions container (if a client is a
* subscriber and a publisher then the client is present in both containers). The
* cost is some additional bookkeeping work upon subscription where each client
* session needs to maintain a list of subscribed topics.
*
* When the client subscribes to the first topic, then the session manager adds
* the client to the _subscriberSessions container, and when the client
* unsubscribes from the last topic then the session manager removes the client
* from the container. Now, when an application message arrives, only the list of
* subscribers need processing instead of looping through potentially thousands of
* publishers.
*
* Improving subscriber topic lookup
* ---------------------------------
* For each subscriber, it needs to be determined whether an application message
* matches any topic the subscriber has subscribed to. There may only be few
* subscribers but there may be many subscribed topics, including wildcard topics.
*
* The implemented approach uses a topic hash and a hash mask calculated on the
* subscribed topic and the published topic (the application message topic) to
* find candidates for a match, with the existing match logic evaluating a reduced
* number of candidates.
*
* For each subscription, the topic hash and a hash mask is stored with the
* subscription, and for each application message received, the hash is calculated
* for the published topic before attempting to find matching subscriptions. The
* hash calculation itself is simple and does not have a large performance impact.
*
* We'll first explain how topic hashes and hash masks are constructed and then how
* they are used.
*
* Topic hash
* ----------
* Topic hashes are stored as 64-bit numbers. Each byte within the 64-bit number
* relates to one MQTT topic level. A checksum is calculated for each topic level
* by iterating over the characters within the topic level (cast to byte) and the
* result is stored into the corresponding byte of the 64-bit number. If a topic
* level contains a wildcard character, then 0x00 is stored instead of the
* checksum.
*
* If there are less than 8 levels then the rest of the 64-bit number is filled
* with 0xff. If there are more than 8 levels then topics where the first 8 MQTT
* topic levels are identical will have the same hash value.
*
* This is the topic hash for the MQTT topic below: 0x655D4AF1FFFFFF
*
* client1/building1/level1/sensor1 (empty) (empty) (empty) (empty)
* \_____/ \_______/ \____/ \_____/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0x65 0x5D 0x4A 0xF1 0xFF 0xFF 0xFF 0xFF
*
* This is the topic hash for an MQTT topic containing a wildcard: 0x655D00F1FFFFFF
*
* client1/building1/ + /sensor1 (empty) (empty) (empty) (empty)
* \_____/ \_______/ \_/ \_____/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0x65 0x5D 0 0xF1 0xFF 0xFF 0xFF 0xFF
*
* For topics that contain the multi level wildcard # at the end, the topic hash
* is filled with 0x00: 0x65004A00000000
*
* client1/ + /level1/ # (empty) (empty) (empty) (empty)
* \_____/ \_/ \____/ \_/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0x65 0 0x4A 0 0 0 0 0
*
*
* Topic hash mask
* ---------------
* The hash mask simply contains 0xFF for non-wildcard topic levels and 0x00 for
* wildcard topic levels. Here are the topic hash masks for the examples above.
*
* client1/building1/level1/sensor1 (empty) (empty) (empty) (empty)
* \_____/ \_______/ \____/ \_____/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF
*
* client1/building1/ + /sensor1 (empty) (empty) (empty) (empty)
* \_____/ \_______/ \_/ \_____/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0xFF 0xFF 0 0xFF 0xFF 0xFF 0xFF 0xFF
*
* client1/ + /level1/ # (empty) (empty) (empty) (empty)
* \_____/ \_/ \____/ \_/ \_____/ \_____/ \_____/ \_____/
* | | | | | | | |
* 0xFF 0 0xFF 0 0 0 0 0
*
*
* Topic hash and hash mask properties
* -----------------------------------
* The following properties of topic hashes and hash masks can be exploited to
* find potentially matching subscribed topics for a given published topic.
*
* (1) If a subscribed topic does not contain wildcards then the topic hash of the
* subscribed topic must be equal to the topic hash of the published topic,
* otherwise the subscribed topic cannot be a candidate for a match.
*
* (2) If a subscribed topic contains wildcards then the hash of the published
* topic masked with the subscribed topic's hash mask must be equal to the hash of
* the subscribed topic. I.e. a subscribed topic is a candidate for a match if:
* (publishedTopicHash & subscribedTopicHashMask) == subscribedTopicHash
*
* (3) If a subscribed topic contains wildcards then any potentially matching
* published topic must have a hash value that is greater than or equal to the
* hash value of the subscribed topic (because the subscribed topic contains
* zeroes in wildcard positions).
*
* Match finding
* -------------
* The subscription manager maintains two separate dictionaries to assist finding
* matches using topic hashes: a _noWildcardSubscriptionsByTopicHash dictionary
* containing all subscriptions that do not have wildcards, and a
* _wildcardSubscriptionsByTopicHash dictionary containing subscriptions with
* wildcards.
*
* For subscriptions without wildcards, all potential candidates for a match are
* obtained by a single look-up (exploiting point 1 above).
*
* For subscriptions with wildcards, the subscription manager loops through the
* wildcard subscriptions and selects candidates that satisfy condition
* (publishedTopicHash & subscribedTopicMask) == subscribedTopicHash (point 2).
* The loop could exit early if wildcard subscriptions were stored into a sorted
* dictionary (utilizing point 3), but, after testing, there does not seem to be
* any real benefit doing so.
*
* Other considerations
* --------------------
* Characters in the topic string are cast to byte and any additional bytes in a
* multi-byte character are disregarded. Best guess is that this does not impact
* performance in practice.
*
* Instead of one-byte checksums per topic level, one-word checksums per topic
* level could be used. If most topics contained four levels or less then hash
* buckets would be shallower.
*
* For very large numbers of topics, performing a parallel search may help further.
*
* To also handle a larger number of subscribers, it may be beneficial to maintain
* a subscribers-by-subscription-topic dictionary.
*/
public static class MqttTopicHash
{
public static void Calculate(string topic, out ulong resultHash, out ulong resultHashMask, out bool resultHasWildcard)
{
// calculate topic hash
ulong hash = 0;
ulong hashMaskInverted = 0;
ulong levelBitMask = 0;
ulong fillLevelBitMask = 0;
var hasWildcard = false;
byte checkSum = 0;
var level = 0;
var i = 0;
while (i < topic.Length)
{
var c = topic[i];
if (c == MqttTopicFilterComparer.LevelSeparator)
{
// done with this level
hash <<= 8;
hash |= checkSum;
hashMaskInverted <<= 8;
hashMaskInverted |= levelBitMask;
checkSum = 0;
levelBitMask = 0;
++level;
if (level >= 8)
{
break;
}
}
else if (c == MqttTopicFilterComparer.SingleLevelWildcard)
{
levelBitMask = 0xff;
hasWildcard = true;
}
else if (c == MqttTopicFilterComparer.MultiLevelWildcard)
{
// checksum is zero for a valid topic
levelBitMask = 0xff;
// fill rest with this fillLevelBitMask
fillLevelBitMask = 0xff;
hasWildcard = true;
break;
}
else
{
// The checksum should be designed to reduce the hash bucket depth for the expected
// fairly regularly named MQTT topics that don't differ much,
// i.e. "room1/sensor1"
// "room1/sensor2"
// "room1/sensor3"
// etc.
if ((c & 1) == 0)
{
checkSum += (byte)c;
}
else
{
checkSum ^= (byte)(c >> 1);
}
}
++i;
}
// Shift hash left and leave zeroes to fill ulong
if (level < 8)
{
hash <<= 8;
hash |= checkSum;
hashMaskInverted <<= 8;
hashMaskInverted |= levelBitMask;
++level;
while (level < 8)
{
hash <<= 8;
hashMaskInverted <<= 8;
hashMaskInverted |= fillLevelBitMask;
++level;
}
}
if (!hasWildcard)
{
while (i < topic.Length)
{
var c = topic[i];
if (c == MqttTopicFilterComparer.SingleLevelWildcard || c == MqttTopicFilterComparer.MultiLevelWildcard)
{
hasWildcard = true;
break;
}
++i;
}
}
resultHash = hash;
resultHashMask = ~hashMaskInverted;
resultHasWildcard = hasWildcard;
}
}
}
Source/MQTTnet/Server/Internal/UnsubscribeResult.cs
+4 -1
View File
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server
@@ -10,7 +11,9 @@ namespace MQTTnet.Server
public sealed class UnsubscribeResult
{
public List<MqttUnsubscribeReasonCode> ReasonCodes { get; } = new List<MqttUnsubscribeReasonCode>(128);
public bool CloseConnection { get; set; }
public List<MqttUserProperty> UserProperties { get; set; }
}
}