diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index e69de29b..0aafba3b 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -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). diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0597dcdc..83f7c057 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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: diff --git a/Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs b/Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs index a0c474a1..0efc7ffa 100644 --- a/Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs @@ -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(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); } diff --git a/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs b/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs index 3349ca5d..f5b9a733 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs @@ -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(); + } } } \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs index f16d55fa..66551983 100644 --- a/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs @@ -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() { diff --git a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs index 1cdcd760..4a2e1f06 100644 --- a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs @@ -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); } diff --git a/Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs b/Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs index e2748f03..7a16ff30 100644 --- a/Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs +++ b/Source/MQTTnet.Tests/Formatter/MqttPacketWriter_Tests.cs @@ -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(writer.GetBuffer(), 0, writer.Length).ToArray(); diff --git a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs index 9995ee82..b3543706 100644 --- a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs +++ b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs @@ -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() diff --git a/Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs b/Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs index 02eeb40a..0a3d3f5e 100644 --- a/Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs +++ b/Source/MQTTnet.Tests/MqttPacketWriter_Tests.cs @@ -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); } } diff --git a/Source/MQTTnet/Client/Internal/MqttClientEvents.cs b/Source/MQTTnet/Client/Internal/MqttClientEvents.cs new file mode 100644 index 00000000..fe786e82 --- /dev/null +++ b/Source/MQTTnet/Client/Internal/MqttClientEvents.cs @@ -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 ApplicationMessageReceivedEvent { get; } = new AsyncEvent(); + public AsyncEvent ConnectedEvent { get; } = new AsyncEvent(); + public AsyncEvent ConnectingEvent { get; } = new AsyncEvent(); + public AsyncEvent DisconnectedEvent { get; } = new AsyncEvent(); + public AsyncEvent InspectPacketEvent { get; } = new AsyncEvent(); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs b/Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs new file mode 100644 index 00000000..ef2800fe --- /dev/null +++ b/Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs @@ -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(); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index a85105b1..b4bb8cd0 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -21,23 +21,13 @@ namespace MQTTnet.Client public sealed class MqttClient : Disposable, IMqttClient { readonly IMqttClientAdapterFactory _adapterFactory; - readonly AsyncEvent _applicationMessageReceivedEvent = new AsyncEvent(); - - readonly MqttClientAuthenticationHandler _authenticationHandler; - - readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory(); - readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory(); - readonly MqttClientUnsubscribeResultFactory _clientUnsubscribeResultFactory = new MqttClientUnsubscribeResultFactory(); - - readonly AsyncEvent _connectedEvent = new AsyncEvent(); - readonly AsyncEvent _connectingEvent = new AsyncEvent(); - readonly AsyncEvent _disconnectedEvent = new AsyncEvent(); readonly object _disconnectLock = new object(); - readonly AsyncEvent _inspectPacketEvent = new AsyncEvent(); - 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 ApplicationMessageReceivedAsync { - add => _applicationMessageReceivedEvent.AddHandler(value); - remove => _applicationMessageReceivedEvent.RemoveHandler(value); + add => _events.ApplicationMessageReceivedEvent.AddHandler(value); + remove => _events.ApplicationMessageReceivedEvent.RemoveHandler(value); } public event Func ConnectedAsync { - add => _connectedEvent.AddHandler(value); - remove => _connectedEvent.RemoveHandler(value); + add => _events.ConnectedEvent.AddHandler(value); + remove => _events.ConnectedEvent.RemoveHandler(value); } public event Func ConnectingAsync { - add => _connectingEvent.AddHandler(value); - remove => _connectingEvent.RemoveHandler(value); + add => _events.ConnectingEvent.AddHandler(value); + remove => _events.ConnectingEvent.RemoveHandler(value); } public event Func DisconnectedAsync { - add => _disconnectedEvent.AddHandler(value); - remove => _disconnectedEvent.RemoveHandler(value); + add => _events.DisconnectedEvent.AddHandler(value); + remove => _events.DisconnectedEvent.RemoveHandler(value); } public event Func ExtendedAuthenticationExchangeAsync @@ -100,8 +90,8 @@ namespace MQTTnet.Client public event Func 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 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 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 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 HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) + async Task 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 PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + async Task PublishAtLeastOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); var pubAckPacket = await Request(publishPacket, cancellationToken).ConfigureAwait(false); - return _clientPublishResultFactory.Create(pubAckPacket); + return MqttClientResultFactory.PublishResult.Create(pubAckPacket); } async Task 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 PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + async Task PublishExactlyOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) { publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier(); @@ -723,7 +766,24 @@ namespace MQTTnet.Client var pubCompPacket = await Request(pubRelPacket, cancellationToken).ConfigureAwait(false); - return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket); + return MqttClientResultFactory.PublishResult.Create(pubRecPacket, pubCompPacket); + } + + async Task 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 Request(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 { diff --git a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs index a7a0099f..3f9f3adb 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs @@ -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 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)) diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs index c9859d1b..e36ce75e 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs +++ b/Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs @@ -34,7 +34,7 @@ namespace MQTTnet.Client /// Gets or sets the reason code. /// MQTT 5.0.0+ feature. /// - public MqttClientPublishReasonCode ReasonCode { get; } = MqttClientPublishReasonCode.Success; + public MqttClientPublishReasonCode ReasonCode { get; } /// /// Gets or sets the reason string. diff --git a/Source/MQTTnet/Formatter/MqttBufferWriter.cs b/Source/MQTTnet/Formatter/MqttBufferWriter.cs index 0849f5b3..14ad707b 100644 --- a/Source/MQTTnet/Formatter/MqttBufferWriter.cs +++ b/Source/MQTTnet/Formatter/MqttBufferWriter.cs @@ -19,7 +19,8 @@ namespace MQTTnet.Formatter /// 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; } } diff --git a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs index 580c59b3..af4e4976 100644 --- a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs @@ -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) diff --git a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs index 92fbb836..808140b4 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV3PacketFormatter.cs @@ -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); diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs index eaddaaad..8023c9c0 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PacketEncoder.cs @@ -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); diff --git a/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs b/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs index 30da6de8..29c06566 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV5PropertiesWriter.cs @@ -335,7 +335,7 @@ namespace MQTTnet.Formatter.V5 } _bufferWriter.WriteByte((byte)id); - _bufferWriter.WriteBinaryData(value); + _bufferWriter.WriteBinary(value); } void WriteAsFourByteInteger(MqttPropertyId id, uint value) diff --git a/Source/MQTTnet/MqttApplicationMessageBuilder.cs b/Source/MQTTnet/MqttApplicationMessageBuilder.cs index a8a740a0..8d65bd31 100644 --- a/Source/MQTTnet/MqttApplicationMessageBuilder.cs +++ b/Source/MQTTnet/MqttApplicationMessageBuilder.cs @@ -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 _payloadSegment; MqttPayloadFormatIndicator _payloadFormatIndicator; + ArraySegment _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(payload); + _payloadSegment = payload == null || payload.Length == 0 ? EmptyBuffer.ArraySegment : new ArraySegment(payload); return this; } - public MqttApplicationMessageBuilder WithPayloadSegment(ArraySegment payloadSegment) + public MqttApplicationMessageBuilder WithPayload(ArraySegment 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 payloadSegment) - { - return MemoryMarshal.TryGetArray(payloadSegment, out var segment) - ? WithPayloadSegment(segment) - : WithPayload(payloadSegment.ToArray()); - } -#endif - /// /// Adds the payload format indicator to the message. /// MQTT 5.0.0+ feature. @@ -182,6 +168,20 @@ namespace MQTTnet return this; } + public MqttApplicationMessageBuilder WithPayloadSegment(ArraySegment payloadSegment) + { + _payloadSegment = payloadSegment; + return this; + } + + +#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1 + public MqttApplicationMessageBuilder WithPayloadSegment(ReadOnlyMemory payloadSegment) + { + return MemoryMarshal.TryGetArray(payloadSegment, out var segment) ? WithPayloadSegment(segment) : WithPayload(payloadSegment.ToArray()); + } +#endif + /// /// 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 diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 627ffe6d..4053e438 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -36,7 +36,7 @@ namespace MQTTnet.Server // See the MqttSubscription object for a detailed explanation. readonly Dictionary _sessions = new Dictionary(4096); - readonly object _sessionsManagementLock = new object(); + readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim(); readonly HashSet _subscriberSessions = new HashSet(); 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 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(); - 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)result); } @@ -415,7 +440,8 @@ namespace MQTTnet.Server public void OnSubscriptionsAdded(MqttSession clientSession, List 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 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 ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) @@ -695,4 +740,4 @@ namespace MQTTnet.Server return eventArgs; } } -} \ No newline at end of file +} diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index a54a0045..037d6489 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -31,7 +31,7 @@ namespace MQTTnet.Server readonly Dictionary _subscriptions = new Dictionary(); // Use subscription lock to maintain consistency across subscriptions and topic hash dictionaries - readonly AsyncLock _subscriptionsLock = new AsyncLock(); + readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim(); readonly Dictionary _wildcardSubscriptionsByTopicHash = new Dictionary(); public MqttClientSubscriptionsManager( @@ -51,7 +51,8 @@ namespace MQTTnet.Server var possibleSubscriptions = new List(); // 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 Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) @@ -232,7 +240,8 @@ namespace MQTTnet.Server var removedSubscriptions = new List(); - 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; } } } -} \ No newline at end of file +}