YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-16 09:13:02 +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-05-06 13:03:13 +02:00
parent 9d9076ba77 449624fd72
commit 3aaf101b3b
22 changed files with 424 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ReleaseNotes.md
+4
View File
@@ -0,0 +1,4 @@
* [Core] Add validation of maximum string lengths (#1718).
* [Client] Added overloads for setting packet payload and will payload (#1720).
* [Client] The proper connect result is now exposed in the _Disconnected_ event when authentication fails (#1139).
* [Server] Improved performance by changing internal locking strategy for subscriptions (#1716, thanks to @zeheng).
.github/workflows/ci.yml
+1 -1
View File
@@ -3,7 +3,7 @@ name: CI
on: [push, pull_request]
env:
VERSION: "4.2.0.${{github.run_number}}"
VERSION: "4.2.1.${{github.run_number}}"
jobs:
build:
Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs
+5 -5
View File
@@ -30,17 +30,17 @@ namespace MQTTnet.Benchmarks
var writer = new MqttBufferWriter(4096, 65535);
writer.WriteString("A relative short string.");
writer.WriteBinaryData(_demoPayload);
writer.WriteBinary(_demoPayload);
writer.WriteByte(0x01);
writer.WriteByte(0x02);
writer.WriteVariableByteInteger(5647382);
writer.WriteString("A relative short string.");
writer.WriteVariableByteInteger(8574489);
writer.WriteBinaryData(_demoPayload);
writer.WriteBinary(_demoPayload);
writer.WriteByte(2);
writer.WriteByte(0x02);
writer.WriteString("fjgffiogfhgfhoihgoireghreghreguhreguireoghreouighreouighreughreguiorehreuiohruiorehreuioghreug");
writer.WriteBinaryData(_demoPayload);
writer.WriteBinary(_demoPayload);
_readPayload = new ArraySegment<byte>(writer.GetBuffer(), 0, writer.Length).ToArray();
}
@@ -83,11 +83,11 @@ namespace MQTTnet.Benchmarks
writer.WriteVariableByteInteger(5647382);
writer.WriteString("A relative short string.");
writer.WriteVariableByteInteger(8574589);
writer.WriteBinaryData(_demoPayload);
writer.WriteBinary(_demoPayload);
writer.WriteByte(2);
writer.WriteByte(0x02);
writer.WriteString("fjgffiogfhgfhoihgoireghreghreguhreguireoghreouighreouighreughreguiorehreuiohruiorehreuioghreug");
writer.WriteBinaryData(_demoPayload);
writer.WriteBinary(_demoPayload);
writer.Reset(0);
}
Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs
+21 -6
View File
@@ -12,22 +12,37 @@ namespace MQTTnet.Extensions.ManagedClient
{
public static IManagedMqttClient CreateManagedMqttClient(this MqttFactory factory, IMqttClient mqttClient = null)
{
if (factory == null) throw new ArgumentNullException(nameof(factory));
if (factory == null)
{
throw new ArgumentNullException(nameof(factory));
}
if (mqttClient == null)
{
return new ManagedMqttClient(factory.CreateMqttClient(), factory.DefaultLogger);
return new ManagedMqttClient(factory.CreateMqttClient(), factory.DefaultLogger);
}
return new ManagedMqttClient(mqttClient, factory.DefaultLogger);
}
public static IManagedMqttClient CreateManagedMqttClient(this MqttFactory factory, IMqttNetLogger logger)
{
if (factory == null) throw new ArgumentNullException(nameof(factory));
if (logger == null) throw new ArgumentNullException(nameof(logger));
if (factory == null)
{
throw new ArgumentNullException(nameof(factory));
}
if (logger == null)
{
throw new ArgumentNullException(nameof(logger));
}
return new ManagedMqttClient(factory.CreateMqttClient(logger), logger);
}
public static ManagedMqttClientOptionsBuilder CreateManagedMqttClientOptionsBuilder(this MqttFactory _)
{
return new ManagedMqttClientOptionsBuilder();
}
}
}
Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs
+32
View File
@@ -22,6 +22,38 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
[TestClass]
public sealed class ManagedMqttClient_Tests : BaseTestClass
{
[TestMethod]
public async Task Expose_Custom_Connection_Error()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
server.ValidatingConnectionAsync += args =>
{
args.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
return CompletedTask.Instance;
};
var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
MqttClientDisconnectedEventArgs disconnectedArgs = null;
managedClient.DisconnectedAsync += args =>
{
disconnectedArgs = args;
return CompletedTask.Instance;
};
var clientOptions = testEnvironment.Factory.CreateManagedMqttClientOptionsBuilder().WithClientOptions(testEnvironment.CreateDefaultClientOptions()).Build();
await managedClient.StartAsync(clientOptions);
await LongTestDelay();
Assert.IsNotNull(disconnectedArgs);
Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, disconnectedArgs.ConnectResult.ResultCode);
}
}
[TestMethod]
public async Task Receive_While_Not_Cleanly_Disconnected()
{
Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs
+2
View File
@@ -127,6 +127,8 @@ namespace MQTTnet.Tests.Clients.MqttClient
// Perform a clean disconnect.
await client.DisconnectAsync(disconnectOptions);
await LongTestDelay();
Assert.IsNotNull(eventArgs);
Assert.AreEqual(MqttDisconnectReasonCode.MessageRateTooHigh, eventArgs.ReasonCode);
}
Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs
+13 -3
View File
@@ -1,6 +1,7 @@
using System;
using System.Linq;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
namespace MQTTnet.Tests.Formatter
@@ -47,23 +48,32 @@ namespace MQTTnet.Tests.Formatter
Assert.AreEqual(1234U, reader.ReadVariableByteInteger());
Assert.AreEqual(9876U, reader.ReadVariableByteInteger());
}
[TestMethod]
[ExpectedException(typeof(MqttProtocolViolationException))]
public void Throw_If_String_Too_Long()
{
var writer = new MqttBufferWriter(4096, 65535);
writer.WriteString(string.Empty.PadLeft(65536));
}
[TestMethod]
public void Write_And_Read_Multiple_Times()
{
var writer = new MqttBufferWriter(4096, 65535);
writer.WriteString("A relative short string.");
writer.WriteBinaryData(new byte[1234]);
writer.WriteBinary(new byte[1234]);
writer.WriteByte(0x01);
writer.WriteByte(0x02);
writer.WriteVariableByteInteger(5647382);
writer.WriteString("A relative short string.");
writer.WriteVariableByteInteger(8574489);
writer.WriteBinaryData(new byte[48]);
writer.WriteBinary(new byte[48]);
writer.WriteByte(2);
writer.WriteByte(0x02);
writer.WriteString("fjgffiogfhgfhoihgoireghreghreguhreguireoghreouighreouighreughreguiorehreuiohruiorehreuioghreug");
writer.WriteBinaryData(new byte[3]);
writer.WriteBinary(new byte[3]);
var readPayload = new ArraySegment<byte>(writer.GetBuffer(), 0, writer.Length).ToArray();
Source/MQTTnet.Tests/Mockups/TestEnvironment.cs
+1 -1
View File
@@ -245,7 +245,7 @@ namespace MQTTnet.Tests.Mockups
public MqttClientOptionsBuilder CreateDefaultClientOptionsBuilder()
{
return Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion).WithTcpServer("127.0.0.1", ServerPort);
return Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion).WithTcpServer("127.0.0.1", ServerPort).WithClientId(TestContext.TestName + "_" + Guid.NewGuid());
}
public ILowLevelMqttClient CreateLowLevelClient()
Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs
+2 -2
View File
@@ -24,10 +24,10 @@ namespace MQTTnet.Tests
writer.WriteString("1234567890");
Assert.AreEqual(10 + 2, writer.Length);
writer.WriteBinaryData(new byte[300]);
writer.WriteBinary(new byte[300]);
Assert.AreEqual(300 + 2 + 12, writer.Length);
writer.WriteBinaryData(new byte[5000]);
writer.WriteBinary(new byte[5000]);
Assert.AreEqual(5000 + 2 + 300 + 2 + 12, writer.Length);
}
}
Source/MQTTnet/Client/Internal/MqttClientEvents.cs
+18
View File
@@ -0,0 +1,18 @@
// 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 MQTTnet.Diagnostics;
using MQTTnet.Internal;
namespace MQTTnet.Client.Internal
{
public sealed class MqttClientEvents
{
public AsyncEvent<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceivedEvent { get; } = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
public AsyncEvent<MqttClientConnectedEventArgs> ConnectedEvent { get; } = new AsyncEvent<MqttClientConnectedEventArgs>();
public AsyncEvent<MqttClientConnectingEventArgs> ConnectingEvent { get; } = new AsyncEvent<MqttClientConnectingEventArgs>();
public AsyncEvent<MqttClientDisconnectedEventArgs> DisconnectedEvent { get; } = new AsyncEvent<MqttClientDisconnectedEventArgs>();
public AsyncEvent<InspectMqttPacketEventArgs> InspectPacketEvent { get; } = new AsyncEvent<InspectMqttPacketEventArgs>();
}
}
Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs
+13
View File
@@ -0,0 +1,13 @@
// 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.Client.Internal
{
public static class MqttClientResultFactory
{
public static readonly MqttClientPublishResultFactory PublishResult = new MqttClientPublishResultFactory();
public static readonly MqttClientSubscribeResultFactory SubscribeResult = new MqttClientSubscribeResultFactory();
public static readonly MqttClientUnsubscribeResultFactory UnsubscribeResult = new MqttClientUnsubscribeResultFactory();
}
}
Source/MQTTnet/Client/MqttClient.cs
+152 -93
View File
@@ -21,23 +21,13 @@ namespace MQTTnet.Client
public sealed class MqttClient : Disposable, IMqttClient
{
readonly IMqttClientAdapterFactory _adapterFactory;
readonly AsyncEvent<MqttApplicationMessageReceivedEventArgs> _applicationMessageReceivedEvent = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
readonly MqttClientAuthenticationHandler _authenticationHandler;
readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory();
readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory();
readonly MqttClientUnsubscribeResultFactory _clientUnsubscribeResultFactory = new MqttClientUnsubscribeResultFactory();
readonly AsyncEvent<MqttClientConnectedEventArgs> _connectedEvent = new AsyncEvent<MqttClientConnectedEventArgs>();
readonly AsyncEvent<MqttClientConnectingEventArgs> _connectingEvent = new AsyncEvent<MqttClientConnectingEventArgs>();
readonly AsyncEvent<MqttClientDisconnectedEventArgs> _disconnectedEvent = new AsyncEvent<MqttClientDisconnectedEventArgs>();
readonly object _disconnectLock = new object();
readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
readonly MqttClientKeepAliveHandler _keepAliveHandler;
readonly MqttClientEvents _events = new MqttClientEvents();
readonly MqttNetSourceLogger _logger;
readonly MqttClientAuthenticationHandler _authenticationHandler;
readonly MqttClientKeepAliveHandler _keepAliveHandler;
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
readonly IMqttNetLogger _rootLogger;
@@ -70,26 +60,26 @@ namespace MQTTnet.Client
public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
{
add => _applicationMessageReceivedEvent.AddHandler(value);
remove => _applicationMessageReceivedEvent.RemoveHandler(value);
add => _events.ApplicationMessageReceivedEvent.AddHandler(value);
remove => _events.ApplicationMessageReceivedEvent.RemoveHandler(value);
}
public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync
{
add => _connectedEvent.AddHandler(value);
remove => _connectedEvent.RemoveHandler(value);
add => _events.ConnectedEvent.AddHandler(value);
remove => _events.ConnectedEvent.RemoveHandler(value);
}
public event Func<MqttClientConnectingEventArgs, Task> ConnectingAsync
{
add => _connectingEvent.AddHandler(value);
remove => _connectingEvent.RemoveHandler(value);
add => _events.ConnectingEvent.AddHandler(value);
remove => _events.ConnectingEvent.RemoveHandler(value);
}
public event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync
{
add => _disconnectedEvent.AddHandler(value);
remove => _disconnectedEvent.RemoveHandler(value);
add => _events.DisconnectedEvent.AddHandler(value);
remove => _events.DisconnectedEvent.RemoveHandler(value);
}
public event Func<MqttExtendedAuthenticationExchangeEventArgs, Task> ExtendedAuthenticationExchangeAsync
@@ -100,8 +90,8 @@ namespace MQTTnet.Client
public event Func<InspectMqttPacketEventArgs, Task> InspectPacketAsync
{
add => _inspectPacketEvent.AddHandler(value);
remove => _inspectPacketEvent.RemoveHandler(value);
add => _events.InspectPacketEvent.AddHandler(value);
remove => _events.InspectPacketEvent.RemoveHandler(value);
}
public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
@@ -125,9 +115,9 @@ namespace MQTTnet.Client
{
Options = options;
if (_connectingEvent.HasHandlers)
if (_events.ConnectingEvent.HasHandlers)
{
await _connectingEvent.InvokeAsync(new MqttClientConnectingEventArgs(options));
await _events.ConnectingEvent.InvokeAsync(new MqttClientConnectingEventArgs(options));
}
Cleanup();
@@ -136,7 +126,7 @@ namespace MQTTnet.Client
_packetDispatcher = new MqttPacketDispatcher();
_clientAlive = new CancellationTokenSource();
var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_inspectPacketEvent, _rootLogger), _rootLogger);
var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_events.InspectPacketEvent, _rootLogger), _rootLogger);
_adapter = adapter;
if (cancellationToken.CanBeCanceled)
@@ -155,23 +145,35 @@ namespace MQTTnet.Client
if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
{
await DisconnectInternalAsync(null, null, connectResult).ConfigureAwait(false);
return connectResult;
_logger.Info($"Using keep alive value ({connectResult.ServerKeepAlive}) sent from the server");
keepAliveInterval = TimeSpan.FromSeconds(connectResult.ServerKeepAlive);
}
if (keepAliveInterval != TimeSpan.Zero)
{
_keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessages(mqttClientAliveToken), mqttClientAliveToken);
}
CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting);
_logger.Info("Connected");
await OnConnected(connectResult).ConfigureAwait(false);
return connectResult;
}
catch (Exception exception)
{
if (exception is MqttConnectingFailedException connectingFailedException)
{
connectResult = connectingFailedException.Result;
}
_disconnectReason = (int)MqttClientDisconnectOptionsReason.UnspecifiedError;
_logger.Error(exception, "Error while connecting with server.");
_logger.Error(exception, "Error while connecting with server");
await DisconnectInternalAsync(null, exception, connectResult).ConfigureAwait(false);
await DisconnectInternal(null, exception, connectResult).ConfigureAwait(false);
throw;
}
@@ -214,19 +216,19 @@ namespace MQTTnet.Client
if (cancellationToken.CanBeCanceled)
{
await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false);
await Send(disconnectPacket, cancellationToken).ConfigureAwait(false);
}
else
{
using (var timeout = new CancellationTokenSource(Options.Timeout))
{
await SendAsync(disconnectPacket, timeout.Token).ConfigureAwait(false);
await Send(disconnectPacket, timeout.Token).ConfigureAwait(false);
}
}
}
finally
{
await DisconnectCoreAsync(null, null, null, clientWasConnected).ConfigureAwait(false);
await DisconnectCore(null, null, null, clientWasConnected).ConfigureAwait(false);
}
}
@@ -269,11 +271,11 @@ namespace MQTTnet.Client
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
return PublishAtLeastOnceAsync(publishPacket, cancellationToken);
return PublishAtLeastOnce(publishPacket, cancellationToken);
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
return PublishExactlyOnceAsync(publishPacket, cancellationToken);
return PublishExactlyOnce(publishPacket, cancellationToken);
}
default:
{
@@ -301,7 +303,7 @@ namespace MQTTnet.Client
UserProperties = data.UserProperties
};
return SendAsync(authPacket, cancellationToken);
return Send(authPacket, cancellationToken);
}
public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default)
@@ -340,7 +342,7 @@ namespace MQTTnet.Client
}
}
return _clientSubscribeResultFactory.Create(subscribePacket, subAckPacket);
return MqttClientResultFactory.SubscribeResult.Create(subscribePacket, subAckPacket);
}
public async Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default)
@@ -379,7 +381,7 @@ namespace MQTTnet.Client
}
}
return _clientUnsubscribeResultFactory.Create(unsubscribePacket, unsubAckPacket);
return MqttClientResultFactory.UnsubscribeResult.Create(unsubscribePacket, unsubAckPacket);
}
internal Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult)
@@ -424,7 +426,7 @@ namespace MQTTnet.Client
if (!eventArgs.ProcessingFailed)
{
var pubAckPacket = MqttPacketFactories.PubAck.Create(eventArgs);
return SendAsync(pubAckPacket, cancellationToken);
return Send(pubAckPacket, cancellationToken);
}
}
else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -432,7 +434,7 @@ namespace MQTTnet.Client
if (!eventArgs.ProcessingFailed)
{
var pubRecPacket = MqttPacketFactories.PubRec.Create(eventArgs);
return SendAsync(pubRecPacket, cancellationToken);
return Send(pubRecPacket, cancellationToken);
}
}
else
@@ -443,6 +445,42 @@ namespace MQTTnet.Client
return CompletedTask.Instance;
}
async Task<MqttClientConnectResult> Authenticate(MqttClientOptions options, CancellationToken cancellationToken)
{
MqttClientConnectResult result;
try
{
var connectPacket = MqttPacketFactories.Connect.Create(options);
await Send(connectPacket, cancellationToken).ConfigureAwait(false);
var receivedPacket = await Receive(cancellationToken).ConfigureAwait(false);
if (receivedPacket is MqttConnAckPacket connAckPacket)
{
var clientConnectResultFactory = new MqttClientConnectResultFactory();
result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
}
else
{
throw new NotSupportedException("Extended authentication handler is not yet supported");
}
}
catch (Exception exception)
{
throw new MqttConnectingFailedException($"Error while authenticating. {exception.Message}", exception, null);
}
if (result.ResultCode != MqttClientConnectResultCode.Success)
{
throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
}
_logger.Verbose("Authenticated MQTT connection with server established.");
return result;
}
void Cleanup()
{
try
@@ -499,7 +537,7 @@ namespace MQTTnet.Client
}
}
async Task DisconnectCoreAsync(Task sender, Exception exception, MqttClientConnectResult connectResult, bool clientWasConnected)
async Task DisconnectCore(Task sender, Exception exception, MqttClientConnectResult connectResult, bool clientWasConnected)
{
TryInitiateDisconnect();
@@ -543,10 +581,34 @@ namespace MQTTnet.Client
_cleanDisconnectInitiated = false;
CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnected, MqttClientConnectionStatus.Disconnecting);
OnDisconnected(clientWasConnected, connectResult, exception);
_logger.Info("Disconnected.");
var eventArgs = new MqttClientDisconnectedEventArgs(
clientWasConnected,
connectResult,
(MqttClientDisconnectReason)_disconnectReason,
_disconnectReasonString,
_disconnectUserProperties,
exception);
// This handler must be executed in a new thread because otherwise a dead lock may happen
// when trying to reconnect in that handler etc.
Task.Run(() => _events.DisconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger);
}
}
Task DisconnectInternal(Task sender, Exception exception, MqttClientConnectResult connectResult)
{
var clientWasConnected = IsConnected;
if (!DisconnectIsPendingOrFinished())
{
return DisconnectCore(sender, exception, connectResult, clientWasConnected);
}
return CompletedTask.Instance;
}
bool DisconnectIsPendingOrFinished()
{
var connectionStatus = (MqttClientConnectionStatus)_connectionStatus;
@@ -586,45 +648,26 @@ namespace MQTTnet.Client
}
}
async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessage(MqttPublishPacket publishPacket)
{
var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket);
var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket);
await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
await _events.ApplicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
return eventArgs;
}
Task OnConnected(MqttClientConnectResult connectResult)
{
_logger.Info("Successfully connected");
if (_connectedEvent.HasHandlers)
if (_events.ConnectedEvent.HasHandlers)
{
var eventArgs = new MqttClientConnectedEventArgs(connectResult);
return _connectedEvent.InvokeAsync(eventArgs);
return _events.ConnectedEvent.InvokeAsync(eventArgs);
}
return CompletedTask.Instance;
}
void OnDisconnected(bool clientWasConnected, MqttClientConnectResult connectResult, Exception exception)
{
_logger.Info("Disconnected");
var eventArgs = new MqttClientDisconnectedEventArgs(
clientWasConnected,
connectResult,
(MqttClientDisconnectReason)_disconnectReason,
_disconnectReasonString,
_disconnectUserProperties,
exception);
// This handler must be executed in a new thread because otherwise a dead lock may happen
// when trying to reconnect in that handler etc.
Task.Run(() => _disconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger);
}
Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket)
{
var extendedAuthenticationExchangeHandler = Options.ExtendedAuthenticationExchangeHandler;
@@ -645,7 +688,7 @@ namespace MQTTnet.Client
// Also dispatch disconnect to waiting threads to generate a proper exception.
_packetDispatcher.Dispose(new MqttClientUnexpectedDisconnectReceivedException(disconnectPacket));
return DisconnectInternalAsync(_packetReceiverTask, null, null);
return DisconnectInternal(_packetReceiverTask, null, null);
}
async Task ProcessReceivedPublishPackets(CancellationToken cancellationToken)
@@ -661,7 +704,7 @@ namespace MQTTnet.Client
}
var publishPacket = publishPacketDequeueResult.Item;
var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
var eventArgs = await HandleReceivedApplicationMessage(publishPacket).ConfigureAwait(false);
if (eventArgs.AutoAcknowledge)
{
@@ -685,7 +728,7 @@ namespace MQTTnet.Client
// The packet is unknown. Probably due to a restart of the client.
// So wen send this to the server to trigger a full resend of the message.
var pubRelPacket = MqttPacketFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
return SendAsync(pubRelPacket, cancellationToken);
return Send(pubRelPacket, cancellationToken);
}
return CompletedTask.Instance;
@@ -694,26 +737,26 @@ namespace MQTTnet.Client
Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{
var pubCompPacket = MqttPacketFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
return SendAsync(pubCompPacket, cancellationToken);
return Send(pubCompPacket, cancellationToken);
}
async Task<MqttClientPublishResult> PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
async Task<MqttClientPublishResult> PublishAtLeastOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
var pubAckPacket = await Request<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
return _clientPublishResultFactory.Create(pubAckPacket);
return MqttClientResultFactory.PublishResult.Create(pubAckPacket);
}
async Task<MqttClientPublishResult> PublishAtMostOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false);
await Send(publishPacket, cancellationToken).ConfigureAwait(false);
return _clientPublishResultFactory.Create(null);
return MqttClientResultFactory.PublishResult.Create(null);
}
async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
async Task<MqttClientPublishResult> PublishExactlyOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
@@ -723,7 +766,24 @@ namespace MQTTnet.Client
var pubCompPacket = await Request<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket);
return MqttClientResultFactory.PublishResult.Create(pubRecPacket, pubCompPacket);
}
async Task<MqttPacket> Receive(CancellationToken cancellationToken)
{
var packetTask = _adapter.ReceivePacketAsync(cancellationToken);
MqttPacket packet;
if (packetTask.IsCompleted)
{
packet = packetTask.Result;
}
else
{
packet = await packetTask.ConfigureAwait(false);
}
return packet;
}
async Task<TResponsePacket> Request<TResponsePacket>(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket
@@ -740,7 +800,7 @@ namespace MQTTnet.Client
{
try
{
await SendAsync(requestPacket, cancellationToken).ConfigureAwait(false);
await Send(requestPacket, cancellationToken).ConfigureAwait(false);
}
catch (Exception exception)
{
@@ -764,6 +824,15 @@ namespace MQTTnet.Client
}
}
Task Send(MqttPacket packet, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
_lastPacketSentTimestamp = DateTime.UtcNow;
return _adapter.SendPacketAsync(packet, cancellationToken);
}
void ThrowIfConnected(string message)
{
if (IsConnected)
@@ -818,7 +887,7 @@ namespace MQTTnet.Client
}
}
async Task TryProcessReceivedPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
async Task TryProcessReceivedPacket(MqttPacket packet, CancellationToken cancellationToken)
{
try
{
@@ -879,11 +948,11 @@ namespace MQTTnet.Client
_packetDispatcher.FailAll(exception);
await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
}
}
async Task TryReceivePacketsAsync(CancellationToken cancellationToken)
async Task ReceivePacketsLoop(CancellationToken cancellationToken)
{
try
{
@@ -891,17 +960,7 @@ namespace MQTTnet.Client
while (!cancellationToken.IsCancellationRequested)
{
MqttPacket packet;
var packetTask = _adapter.ReceivePacketAsync(cancellationToken);
if (packetTask.IsCompleted)
{
packet = packetTask.Result;
}
else
{
packet = await packetTask.ConfigureAwait(false);
}
var packet = await Receive(cancellationToken).ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
{
@@ -910,12 +969,12 @@ namespace MQTTnet.Client
if (packet == null)
{
await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
await DisconnectInternal(_packetReceiverTask, null, null).ConfigureAwait(false);
return;
}
await TryProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false);
await TryProcessReceivedPacket(packet, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception exception)
@@ -944,7 +1003,7 @@ namespace MQTTnet.Client
_packetDispatcher.FailAll(exception);
await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
}
finally
{
Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs
+13 -1
View File
@@ -1,4 +1,4 @@
// Licensed to the .NET Foundation under one or more agreements.
// 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.
@@ -421,6 +421,18 @@ namespace MQTTnet.Client
return this;
}
public MqttClientOptionsBuilder WithWillPayload(ArraySegment<byte> willPayload)
{
if (willPayload.Count == 0)
{
_options.WillPayload = null;
return this;
}
_options.WillPayload = willPayload.ToArray();
return this;
}
public MqttClientOptionsBuilder WithWillPayload(string willPayload)
{
if (string.IsNullOrEmpty(willPayload))
Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs
+1 -1
View File
@@ -34,7 +34,7 @@ namespace MQTTnet.Client
/// Gets or sets the reason code.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public MqttClientPublishReasonCode ReasonCode { get; } = MqttClientPublishReasonCode.Success;
public MqttClientPublishReasonCode ReasonCode { get; }
/// <summary>
/// Gets or sets the reason string.
Source/MQTTnet/Formatter/MqttBufferWriter.cs
+31 -22
View File
@@ -19,7 +19,8 @@ namespace MQTTnet.Formatter
/// </summary>
public sealed class MqttBufferWriter
{
public const uint VariableByteIntegerMaxValue = 268435455;
const uint VariableByteIntegerMaxValue = 268435455;
const int EncodedStringMaxLength = 65535;
readonly int _maxBufferSize;
@@ -112,25 +113,7 @@ namespace MQTTnet.Formatter
WriteBinary(propertyWriter._buffer, 0, propertyWriter.Length);
}
public void WriteBinary(byte[] buffer, int offset, int count)
{
if (buffer == null)
{
throw new ArgumentNullException(nameof(buffer));
}
if (count == 0)
{
return;
}
EnsureAdditionalCapacity(count);
MqttMemoryHelper.Copy(buffer, offset, _buffer, _position, count);
IncreasePosition(count);
}
public void WriteBinaryData(byte[] value)
public void WriteBinary(byte[] value)
{
if (value == null || value.Length == 0)
{
@@ -155,6 +138,24 @@ namespace MQTTnet.Formatter
}
}
public void WriteBinary(byte[] buffer, int offset, int count)
{
if (buffer == null)
{
throw new ArgumentNullException(nameof(buffer));
}
if (count == 0)
{
return;
}
EnsureAdditionalCapacity(count);
MqttMemoryHelper.Copy(buffer, offset, _buffer, _position, count);
IncreasePosition(count);
}
public void WriteByte(byte @byte)
{
EnsureAdditionalCapacity(1);
@@ -186,6 +187,13 @@ namespace MQTTnet.Formatter
var writtenBytes = Encoding.UTF8.GetBytes(value, 0, value.Length, _buffer, _position + 2);
// From RFC: 1.5.4 UTF-8 Encoded String
// Unless stated otherwise all UTF-8 encoded strings can have any length in the range 0 to 65,535 bytes.
if (writtenBytes > EncodedStringMaxLength)
{
throw new MqttProtocolViolationException($"The maximum string length is 65535. The current string has a length of {writtenBytes}.");
}
_buffer[_position] = (byte)(writtenBytes >> 8);
_buffer[_position + 1] = (byte)writtenBytes;
@@ -273,8 +281,7 @@ namespace MQTTnet.Formatter
newBufferLength *= 2;
}
// Array.Resize will create a new array and copy the existing
// data to the new one.
// Array.Resize will create a new array and copy the existing data to the new one.
Array.Resize(ref _buffer, newBufferLength);
}
@@ -285,6 +292,8 @@ namespace MQTTnet.Formatter
if (_position > Length)
{
// Also extend the position because we reached the end of the
// pre allocated buffer.
Length = _position;
}
}
Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs
+6 -7
View File
@@ -31,10 +31,14 @@ namespace MQTTnet.Formatter
public MqttProtocolVersion ProtocolVersion { get; private set; } = MqttProtocolVersion.Unknown;
public void Cleanup()
{
_bufferWriter.Cleanup();
}
public MqttPacket Decode(ReceivedMqttPacket receivedMqttPacket)
{
ThrowIfFormatterNotSet();
return _formatter.Decode(receivedMqttPacket);
}
@@ -50,11 +54,6 @@ namespace MQTTnet.Formatter
return _formatter.Encode(packet);
}
public void Cleanup()
{
_bufferWriter.Cleanup();
}
public static IMqttPacketFormatter GetMqttPacketFormatter(MqttProtocolVersion protocolVersion, MqttBufferWriter bufferWriter)
{
if (protocolVersion == MqttProtocolVersion.Unknown)
@@ -98,7 +97,7 @@ namespace MQTTnet.Formatter
// Remove the mosquitto try_private flag (MQTT 3.1.1 Bridge).
// This flag is accepted but not yet used.
protocolLevel &= 0x7F;
if (protocolName == "MQTT")
{
if (protocolLevel == 5)
Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs
+4 -4
View File
@@ -459,7 +459,7 @@ namespace MQTTnet.Formatter.V3
if (packet.WillFlag)
{
bufferWriter.WriteString(packet.WillTopic);
bufferWriter.WriteBinaryData(packet.WillMessage);
bufferWriter.WriteBinary(packet.WillMessage);
}
if (packet.Username != null)
@@ -469,7 +469,7 @@ namespace MQTTnet.Formatter.V3
if (packet.Password != null)
{
bufferWriter.WriteBinaryData(packet.Password);
bufferWriter.WriteBinary(packet.Password);
}
return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Connect);
@@ -521,7 +521,7 @@ namespace MQTTnet.Formatter.V3
if (packet.WillFlag)
{
bufferWriter.WriteString(packet.WillTopic);
bufferWriter.WriteBinaryData(packet.WillMessage);
bufferWriter.WriteBinary(packet.WillMessage);
}
if (packet.Username != null)
@@ -531,7 +531,7 @@ namespace MQTTnet.Formatter.V3
if (packet.Password != null)
{
bufferWriter.WriteBinaryData(packet.Password);
bufferWriter.WriteBinary(packet.Password);
}
return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Connect);
Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs
+2 -2
View File
@@ -186,7 +186,7 @@ namespace MQTTnet.Formatter.V5
_propertiesWriter.Reset();
_bufferWriter.WriteString(packet.WillTopic);
_bufferWriter.WriteBinaryData(packet.WillMessage);
_bufferWriter.WriteBinary(packet.WillMessage);
}
if (packet.Username != null)
@@ -196,7 +196,7 @@ namespace MQTTnet.Formatter.V5
if (packet.Password != null)
{
_bufferWriter.WriteBinaryData(packet.Password);
_bufferWriter.WriteBinary(packet.Password);
}
return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Connect);
Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs
+1 -1
View File
@@ -335,7 +335,7 @@ namespace MQTTnet.Formatter.V5
}
_bufferWriter.WriteByte((byte)id);
_bufferWriter.WriteBinaryData(value);
_bufferWriter.WriteBinary(value);
}
void WriteAsFourByteInteger(MqttPropertyId id, uint value)
Source/MQTTnet/MqttApplicationMessageBuilder.cs
+22 -22
View File
@@ -2,16 +2,16 @@
// 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.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Runtime.InteropServices;
using System.Text;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet
{
@@ -20,9 +20,9 @@ namespace MQTTnet
string _contentType;
byte[] _correlationData;
uint _messageExpiryInterval;
ArraySegment<byte> _payloadSegment;
MqttPayloadFormatIndicator _payloadFormatIndicator;
ArraySegment<byte> _payloadSegment;
MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
string _responseTopic;
bool _retain;
@@ -89,13 +89,11 @@ namespace MQTTnet
public MqttApplicationMessageBuilder WithPayload(byte[] payload)
{
_payloadSegment = payload == null || payload.Length == 0
? EmptyBuffer.ArraySegment
: new ArraySegment<byte>(payload);
_payloadSegment = payload == null || payload.Length == 0 ? EmptyBuffer.ArraySegment : new ArraySegment<byte>(payload);
return this;
}
public MqttApplicationMessageBuilder WithPayloadSegment(ArraySegment<byte> payloadSegment)
public MqttApplicationMessageBuilder WithPayload(ArraySegment<byte> payloadSegment)
{
_payloadSegment = payloadSegment;
return this;
@@ -123,9 +121,7 @@ namespace MQTTnet
public MqttApplicationMessageBuilder WithPayload(Stream payload)
{
return payload == null
? WithPayload(default(byte[]))
: WithPayload(payload, payload.Length - payload.Position);
return payload == null ? WithPayload(default(byte[])) : WithPayload(payload, payload.Length - payload.Position);
}
public MqttApplicationMessageBuilder WithPayload(Stream payload, long length)
@@ -162,16 +158,6 @@ namespace MQTTnet
return WithPayload(payloadBuffer);
}
#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1
public MqttApplicationMessageBuilder WithPayloadSegment(ReadOnlyMemory<byte> payloadSegment)
{
return MemoryMarshal.TryGetArray(payloadSegment, out var segment)
? WithPayloadSegment(segment)
: WithPayload(payloadSegment.ToArray());
}
#endif
/// <summary>
/// Adds the payload format indicator to the message.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
@@ -182,6 +168,20 @@ namespace MQTTnet
return this;
}
public MqttApplicationMessageBuilder WithPayloadSegment(ArraySegment<byte> payloadSegment)
{
_payloadSegment = payloadSegment;
return this;
}
#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1
public MqttApplicationMessageBuilder WithPayloadSegment(ReadOnlyMemory<byte> payloadSegment)
{
return MemoryMarshal.TryGetArray(payloadSegment, out var segment) ? WithPayloadSegment(segment) : WithPayload(payloadSegment.ToArray());
}
#endif
/// <summary>
/// The quality of service level.
/// The Quality of Service (QoS) level is an agreement between the sender of a message and the receiver of a message
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+55 -10
View File
@@ -36,7 +36,7 @@ namespace MQTTnet.Server
// See the MqttSubscription object for a detailed explanation.
readonly Dictionary<string, MqttSession> _sessions = new Dictionary<string, MqttSession>(4096);
readonly object _sessionsManagementLock = new object();
readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>();
public MqttClientSessionsManager(
@@ -86,7 +86,8 @@ namespace MQTTnet.Server
MqttSession session;
lock (_sessionsManagementLock)
_sessionsManagementLock.EnterWriteLock();
try
{
_sessions.TryGetValue(clientId, out session);
_sessions.Remove(clientId);
@@ -96,6 +97,10 @@ namespace MQTTnet.Server
_subscriberSessions.Remove(session);
}
}
finally
{
_sessionsManagementLock.ExitWriteLock();
}
try
{
@@ -173,10 +178,15 @@ namespace MQTTnet.Server
}
List<MqttSession> subscriberSessions;
lock (_sessionsManagementLock)
_sessionsManagementLock.EnterReadLock();
try
{
subscriberSessions = _subscriberSessions.ToList();
}
finally
{
_sessionsManagementLock.ExitReadLock();
}
// Calculate application message topic hash once for subscription checks
MqttTopicHash.Calculate(applicationMessage.Topic, out var topicHash, out _, out _);
@@ -256,13 +266,23 @@ namespace MQTTnet.Server
{
_createConnectionSyncRoot.Dispose();
lock (_sessionsManagementLock)
_sessionsManagementLock.EnterWriteLock();
try
{
foreach (var sessionItem in _sessions)
{
sessionItem.Value.Dispose();
}
}
finally
{
_sessionsManagementLock.ExitWriteLock();
}
if (_sessionsManagementLock != null)
{
_sessionsManagementLock.Dispose();
}
}
public MqttClient GetClient(string id)
@@ -310,7 +330,8 @@ namespace MQTTnet.Server
{
var result = new List<MqttSessionStatus>();
lock (_sessionsManagementLock)
_sessionsManagementLock.EnterReadLock();
try
{
foreach (var sessionItem in _sessions)
{
@@ -318,6 +339,10 @@ namespace MQTTnet.Server
result.Add(sessionStatus);
}
}
finally
{
_sessionsManagementLock.ExitReadLock();
}
return Task.FromResult((IList<MqttSessionStatus>)result);
}
@@ -415,7 +440,8 @@ namespace MQTTnet.Server
public void OnSubscriptionsAdded(MqttSession clientSession, List<string> topics)
{
lock (_sessionsManagementLock)
_sessionsManagementLock.EnterWriteLock();
try
{
if (!clientSession.HasSubscribedTopics)
{
@@ -428,11 +454,16 @@ namespace MQTTnet.Server
clientSession.AddSubscribedTopic(topic);
}
}
finally
{
_sessionsManagementLock.ExitWriteLock();
}
}
public void OnSubscriptionsRemoved(MqttSession clientSession, List<string> subscriptionTopics)
{
lock (_sessionsManagementLock)
_sessionsManagementLock.EnterWriteLock();
try
{
foreach (var subscriptionTopic in subscriptionTopics)
{
@@ -445,6 +476,10 @@ namespace MQTTnet.Server
_subscriberSessions.Remove(clientSession);
}
}
finally
{
_sessionsManagementLock.ExitWriteLock();
}
}
public void Start()
@@ -545,7 +580,8 @@ namespace MQTTnet.Server
MqttSession oldSession;
MqttClient oldClient;
lock (_sessionsManagementLock)
_sessionsManagementLock.EnterWriteLock();
try
{
MqttSession session;
@@ -590,6 +626,10 @@ namespace MQTTnet.Server
_clients[connectPacket.ClientId] = client;
}
finally
{
_sessionsManagementLock.ExitWriteLock();
}
if (!connAckPacket.IsSessionPresent)
{
@@ -636,7 +676,8 @@ namespace MQTTnet.Server
MqttSession GetClientSession(string clientId)
{
lock (_sessionsManagementLock)
_sessionsManagementLock.EnterReadLock();
try
{
if (!_sessions.TryGetValue(clientId, out var session))
{
@@ -645,6 +686,10 @@ namespace MQTTnet.Server
return session;
}
finally
{
_sessionsManagementLock.ExitReadLock();
}
}
async Task<MqttConnectPacket> ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
@@ -695,4 +740,4 @@ namespace MQTTnet.Server
return eventArgs;
}
}
}
}
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+25 -8
View File
@@ -31,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 AsyncLock _subscriptionsLock = new AsyncLock();
readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim();
readonly Dictionary<ulong, TopicHashMaskSubscriptions> _wildcardSubscriptionsByTopicHash = new Dictionary<ulong, TopicHashMaskSubscriptions>();
public MqttClientSubscriptionsManager(
@@ -51,7 +51,8 @@ namespace MQTTnet.Server
var possibleSubscriptions = new List<MqttSubscription>();
// Check for possible subscriptions. They might have collisions but this is fine.
using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult())
_subscriptionsLock.EnterReadLock();
try
{
if (_noWildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var noWildcardSubscriptions))
{
@@ -73,6 +74,10 @@ namespace MQTTnet.Server
}
}
}
finally
{
_subscriptionsLock.ExitReadLock();
}
// The pre check has evaluated that nothing is subscribed.
// If there were some possible candidates they get checked below
@@ -157,7 +162,10 @@ namespace MQTTnet.Server
public void Dispose()
{
_subscriptionsLock.Dispose();
if (_subscriptionsLock != null)
{
_subscriptionsLock.Dispose();
}
}
public async Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
@@ -232,7 +240,8 @@ namespace MQTTnet.Server
var removedSubscriptions = new List<string>();
using (await _subscriptionsLock.EnterAsync(cancellationToken).ConfigureAwait(false))
_subscriptionsLock.EnterWriteLock();
try
{
foreach (var topicFilter in unsubscribePacket.TopicFilters)
{
@@ -293,8 +302,11 @@ namespace MQTTnet.Server
}
}
}
_subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions);
finally
{
_subscriptionsLock.ExitWriteLock();
_subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions);
}
if (_eventContainer.ClientUnsubscribedTopicEvent.HasHandlers)
{
@@ -341,7 +353,8 @@ namespace MQTTnet.Server
// Add to subscriptions and maintain topic hash dictionaries
using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult())
_subscriptionsLock.EnterWriteLock();
try
{
MqttTopicHash.Calculate(topicFilter.Topic, out var topicHash, out _, out var hasWildcard);
@@ -391,6 +404,10 @@ namespace MQTTnet.Server
subscriptions.Add(subscription);
}
}
finally
{
_subscriptionsLock.ExitWriteLock();
}
return new CreateSubscriptionResult
{
@@ -511,4 +528,4 @@ namespace MQTTnet.Server
public MqttSubscription Subscription { get; set; }
}
}
}
}