From f7db00c08e49216ca1706ca140d4a359c4b596b8 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sun, 11 Oct 2020 16:41:13 +0200 Subject: [PATCH 1/8] improved behavior when multiple connections fight over a session --- Source/MQTTnet/Client/MqttClient.cs | 8 +++- Source/MQTTnet/Server/MqttClientConnection.cs | 9 ++++ .../Server/MqttClientSessionsManager.cs | 45 ++++++++++--------- Tests/MQTTnet.Core.Tests/Session_Tests.cs | 36 +++++++++++++++ 4 files changed, 76 insertions(+), 22 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 8a045bed..51a3612f 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -561,12 +561,16 @@ namespace MQTTnet.Client { await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false); } - else if (packet is MqttDisconnectPacket) + else if (packet is MqttDisconnectPacket disc) { // Also dispatch disconnect to waiting threads to generate a proper exception. _packetDispatcher.Dispatch(packet); - await DisconnectAsync(null, cancellationToken).ConfigureAwait(false); + await DisconnectAsync(new MqttClientDisconnectOptions() + { + // todo conversion + ReasonCode = disc.ReasonCode + }, cancellationToken).ConfigureAwait(false); } else if (packet is MqttAuthPacket authPacket) { diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index bf51269c..57fb63eb 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -257,6 +257,15 @@ namespace MQTTnet.Server Session.WillMessage = null; } + if (_isTakeover) + { + // dont use SendAsync here _cancellationToken is already cancelled + await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket() + { + ReasonCode = MqttDisconnectReasonCode.SessionTakenOver + }, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false); + } + _packetDispatcher.Reset(); _channelAdapter.ReadingPacketStartedCallback = null; diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 48f69809..6935bcbe 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -316,7 +316,11 @@ namespace MQTTnet.Server { if (clientId != null) { - _connections.TryRemove(clientId, out _); + // in case it is a takeover _connections already contains the new connection + if (disconnectType != MqttClientDisconnectType.Takeover) + { + _connections.TryRemove(clientId, out _); + } if (!_options.EnablePersistentSessions) { @@ -364,38 +368,39 @@ namespace MQTTnet.Server { using (await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false)) { - var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var session); - - var isConnectionPresent = _connections.TryGetValue(connectPacket.ClientId, out var existingConnection); - if (isConnectionPresent) + var session = _sessions.AddOrUpdate(connectPacket.ClientId, key => { - await existingConnection.StopAsync(true).ConfigureAwait(false); - } - - if (isSessionPresent) + _logger.Verbose("Created a new session for client '{0}'.", key); + return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger); + }, (key, existingSession) => { if (connectPacket.CleanSession) { - session = null; - _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId); + return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger); } else { _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId); + return existingSession; } - } - - if (session == null) - { - session = new MqttClientSession(connectPacket.ClientId, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger); - _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId); - } + }); var connection = new MqttClientConnection(connectPacket, channelAdapter, session, _options, this, _retainedMessagesManager, onStart, onStop, _rootLogger); + MqttClientConnection existingConnection = null; + _connections.AddOrUpdate(connectPacket.ClientId, key => + { + return connection; + }, (key, tempexistingConnection) => + { + existingConnection = tempexistingConnection; + return connection; + }); - _connections[connection.ClientId] = connection; - _sessions[session.ClientId] = session; + if (existingConnection != null) + { + await existingConnection.StopAsync(true).ConfigureAwait(false); + } return connection; } diff --git a/Tests/MQTTnet.Core.Tests/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Session_Tests.cs index 973fbf3e..1f4c418d 100644 --- a/Tests/MQTTnet.Core.Tests/Session_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Session_Tests.cs @@ -1,5 +1,6 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; +using MQTTnet.Client.Options; using MQTTnet.Client.Subscribing; using MQTTnet.Server; using MQTTnet.Tests.Mockups; @@ -84,5 +85,40 @@ namespace MQTTnet.Tests Assert.AreEqual(true, session.Items["can_subscribe_x"]); } } + + + [TestMethod] + public async Task Manage_Session_MaxParallel() + { + using (var testEnvironment = new TestEnvironment(TestContext)) + { + testEnvironment.IgnoreClientLogErrors = true; + var serverOptions = new MqttServerOptionsBuilder(); + await testEnvironment.StartServerAsync(serverOptions); + + var options = new MqttClientOptionsBuilder() + .WithClientId("1") + ; + + var clients = await Task.WhenAll(Enumerable.Range(0, 10) + .Select(i => TryConnect(testEnvironment, options))); + + var connectedClients = clients.Where(c => c?.IsConnected ?? false).ToList(); + + Assert.AreEqual(1, connectedClients.Count); + } + } + + private async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) + { + try + { + return await testEnvironment.ConnectClientAsync(options); + } + catch (System.Exception) + { + return null; + } + } } } From d7b552fce23ecc41502639e70bdf17b36f32374c Mon Sep 17 00:00:00 2001 From: JanEggers Date: Tue, 13 Oct 2020 15:57:21 +0200 Subject: [PATCH 2/8] renaming and added cast --- Build/MQTTnet.nuspec | 4 +++- Source/MQTTnet/Client/MqttClient.cs | 4 ++-- Source/MQTTnet/Server/MqttClientSessionsManager.cs | 4 ++-- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index d8558191..fd4bc2e0 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -14,8 +14,10 @@ * [Client] Fixed wrong value for "ClientWasConnected" in "MqttClientDisconnectedEventArgs" #976 (thanks to @dbeinder). * [Client] Added direct support for Amazon AWS connections (requires .NET Core 3.1) (thanks to @henning-krause). +* [Client] handle disconnect package and propagate disconnect reason code (thanks to @JanEggers) * [ManagedClient] Exposed the internal MQTT client. -* [Server] Added client message queue interceptor for QoS level (tahnks to @msallin). +* [Server] Added client message queue interceptor for QoS level (thanks to @msallin). +* [Server] improved behavior when multiple connections fight over a session (thanks to @JanEggers) Copyright Christian Kratky 2016-2020 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 51a3612f..44cd85cd 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -561,7 +561,7 @@ namespace MQTTnet.Client { await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false); } - else if (packet is MqttDisconnectPacket disc) + else if (packet is MqttDisconnectPacket disconnectPacket) { // Also dispatch disconnect to waiting threads to generate a proper exception. _packetDispatcher.Dispatch(packet); @@ -569,7 +569,7 @@ namespace MQTTnet.Client await DisconnectAsync(new MqttClientDisconnectOptions() { // todo conversion - ReasonCode = disc.ReasonCode + ReasonCode = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.UnspecifiedError) }, cancellationToken).ConfigureAwait(false); } else if (packet is MqttAuthPacket authPacket) diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 6935bcbe..84fa7251 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -391,9 +391,9 @@ namespace MQTTnet.Server _connections.AddOrUpdate(connectPacket.ClientId, key => { return connection; - }, (key, tempexistingConnection) => + }, (key, tempExistingConnection) => { - existingConnection = tempexistingConnection; + existingConnection = tempExistingConnection; return connection; }); From 2eaebe0e6a152d3d493512d992911e26f3dc262c Mon Sep 17 00:00:00 2001 From: JanEggers Date: Tue, 13 Oct 2020 15:59:39 +0200 Subject: [PATCH 3/8] removed todo --- Source/MQTTnet/Client/MqttClient.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 44cd85cd..1923a2e1 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -568,7 +568,6 @@ namespace MQTTnet.Client await DisconnectAsync(new MqttClientDisconnectOptions() { - // todo conversion ReasonCode = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.UnspecifiedError) }, cancellationToken).ConfigureAwait(false); } From 1a6be47144eab9ec6b153b96793aa34b24878b04 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 14 Oct 2020 23:10:13 +0200 Subject: [PATCH 4/8] Update docs. --- Build/MQTTnet.nuspec | 1 + 1 file changed, 1 insertion(+) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index d8558191..83fba903 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -16,6 +16,7 @@ * [Client] Added direct support for Amazon AWS connections (requires .NET Core 3.1) (thanks to @henning-krause). * [ManagedClient] Exposed the internal MQTT client. * [Server] Added client message queue interceptor for QoS level (tahnks to @msallin). +* [Server] Exposed _ClientDisconnectedInterceptor_ (thanks to @SeppPenner). Copyright Christian Kratky 2016-2020 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor From a26b641284cbd437e0ce159748d631fc3279857a Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Thu, 15 Oct 2020 01:01:35 +0200 Subject: [PATCH 5/8] Fix broken connection takeover. --- Source/MQTTnet/Client/MqttClient.cs | 27 +++--- Source/MQTTnet/Internal/AsyncQueue.cs | 21 +++-- Source/MQTTnet/Server/MqttClientConnection.cs | 56 +++++------ .../Server/MqttClientSessionsManager.cs | 92 +++++++++---------- 4 files changed, 101 insertions(+), 95 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 1923a2e1..cb87120a 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -245,16 +245,22 @@ namespace MQTTnet.Client void Cleanup() { - _backgroundCancellationTokenSource?.Cancel(false); - _backgroundCancellationTokenSource?.Dispose(); - _backgroundCancellationTokenSource = null; + try + { + _backgroundCancellationTokenSource?.Cancel(false); + } + finally + { + _backgroundCancellationTokenSource?.Dispose(); + _backgroundCancellationTokenSource = null; - _publishPacketReceiverQueue?.Dispose(); - _publishPacketReceiverQueue = null; + _publishPacketReceiverQueue?.Dispose(); + _publishPacketReceiverQueue = null; - _adapter?.Dispose(); + _adapter?.Dispose(); + } } - + protected override void Dispose(bool disposing) { if (disposing) @@ -387,7 +393,7 @@ namespace MQTTnet.Client async Task SendAndReceiveAsync(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket { cancellationToken.ThrowIfCancellationRequested(); - + ushort identifier = 0; if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier.HasValue) { @@ -436,7 +442,7 @@ namespace MQTTnet.Client { // Values described here: [MQTT-3.1.2-24]. var waitTime = keepAlivePeriod - _sendTracker.Elapsed; - + if (waitTime <= TimeSpan.Zero) { await SendAndReceiveAsync(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false); @@ -566,7 +572,7 @@ namespace MQTTnet.Client // Also dispatch disconnect to waiting threads to generate a proper exception. _packetDispatcher.Dispatch(packet); - await DisconnectAsync(new MqttClientDisconnectOptions() + await DisconnectAsync(new MqttClientDisconnectOptions() { ReasonCode = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.UnspecifiedError) }, cancellationToken).ConfigureAwait(false); @@ -631,7 +637,6 @@ namespace MQTTnet.Client try { var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken); - if (!publishPacketDequeueResult.IsSuccess) { return; diff --git a/Source/MQTTnet/Internal/AsyncQueue.cs b/Source/MQTTnet/Internal/AsyncQueue.cs index 43ad938a..868ed617 100644 --- a/Source/MQTTnet/Internal/AsyncQueue.cs +++ b/Source/MQTTnet/Internal/AsyncQueue.cs @@ -7,16 +7,15 @@ namespace MQTTnet.Internal { public sealed class AsyncQueue : IDisposable { - private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0); - - private ConcurrentQueue _queue = new ConcurrentQueue(); - + SemaphoreSlim _semaphore = new SemaphoreSlim(0); + ConcurrentQueue _queue = new ConcurrentQueue(); + public int Count => _queue.Count; public void Enqueue(TItem item) { _queue.Enqueue(item); - _semaphore.Release(); + _semaphore?.Release(); } public async Task> TryDequeueAsync(CancellationToken cancellationToken) @@ -25,12 +24,17 @@ namespace MQTTnet.Internal { try { + if (_semaphore == null) + { + return new AsyncQueueDequeueResult(false, default); + } + await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); } catch (OperationCanceledException) { - return new AsyncQueueDequeueResult(false, default(TItem)); + return new AsyncQueueDequeueResult(false, default); } if (_queue.TryDequeue(out var item)) @@ -39,7 +43,7 @@ namespace MQTTnet.Internal } } - return new AsyncQueueDequeueResult(false, default(TItem)); + return new AsyncQueueDequeueResult(false, default); } public AsyncQueueDequeueResult TryDequeue() @@ -49,7 +53,7 @@ namespace MQTTnet.Internal return new AsyncQueueDequeueResult(true, item); } - return new AsyncQueueDequeueResult(false, default(TItem)); + return new AsyncQueueDequeueResult(false, default); } public void Clear() @@ -60,6 +64,7 @@ namespace MQTTnet.Internal public void Dispose() { _semaphore?.Dispose(); + _semaphore = null; } } } diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 57fb63eb..d501c8eb 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -23,8 +23,6 @@ namespace MQTTnet.Server readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource(); readonly IMqttRetainedMessagesManager _retainedMessagesManager; - readonly Func _onStart; - readonly Func _onStop; readonly MqttClientKeepAliveMonitor _keepAliveMonitor; readonly MqttClientSessionsManager _sessionsManager; @@ -32,6 +30,7 @@ namespace MQTTnet.Server readonly IMqttServerOptions _serverOptions; readonly IMqttChannelAdapter _channelAdapter; + readonly MqttConnectionValidatorContext _connectionValidatorContext; readonly IMqttDataConverter _dataConverter; readonly string _endpoint; readonly DateTime _connectedTimestamp; @@ -47,25 +46,22 @@ namespace MQTTnet.Server volatile bool _isTakeover; - public MqttClientConnection( - MqttConnectPacket connectPacket, + public MqttClientConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttClientSession session, + MqttConnectionValidatorContext connectionValidatorContext, IMqttServerOptions serverOptions, MqttClientSessionsManager sessionsManager, IMqttRetainedMessagesManager retainedMessagesManager, - Func onStart, - Func onStop, IMqttNetLogger logger) { Session = session ?? throw new ArgumentNullException(nameof(session)); _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager)); _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); - _onStart = onStart ?? throw new ArgumentNullException(nameof(onStart)); - _onStop = onStop ?? throw new ArgumentNullException(nameof(onStop)); _channelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter)); + _connectionValidatorContext = connectionValidatorContext ?? throw new ArgumentNullException(nameof(connectionValidatorContext)); _dataConverter = _channelAdapter.PacketFormatterAdapter.DataConverter; _endpoint = _channelAdapter.Endpoint; ConnectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket)); @@ -131,18 +127,17 @@ namespace MQTTnet.Server _cancellationToken.Dispose(); } - public Task RunAsync(MqttConnectionValidatorContext connectionValidatorContext) + public Task RunAsync() { - _packageReceiverTask = RunInternalAsync(connectionValidatorContext); + _packageReceiverTask = RunInternalAsync(); return _packageReceiverTask; } - async Task RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext) + async Task RunInternalAsync() { var disconnectType = MqttClientDisconnectType.NotClean; try { - await _onStart(); _logger.Info("Client '{0}': Session started.", ClientId); _channelAdapter.ReadingPacketStartedCallback = OnAdapterReadingPacketStarted; @@ -155,9 +150,7 @@ namespace MQTTnet.Server // TODO: Change to single thread in SessionManager. Or use SessionManager and stats from KeepAliveMonitor. _keepAliveMonitor.Start(ConnectPacket.KeepAlivePeriod, _cancellationToken.Token); - await SendAsync( - _channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext) - ).ConfigureAwait(false); + await SendAsync(_channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(_connectionValidatorContext)).ConfigureAwait(false); Session.IsCleanSession = false; @@ -257,31 +250,38 @@ namespace MQTTnet.Server Session.WillMessage = null; } - if (_isTakeover) - { - // dont use SendAsync here _cancellationToken is already cancelled - await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket() - { - ReasonCode = MqttDisconnectReasonCode.SessionTakenOver - }, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false); - } - _packetDispatcher.Reset(); _channelAdapter.ReadingPacketStartedCallback = null; _channelAdapter.ReadingPacketCompletedCallback = null; - _logger.Info("Client '{0}': Connection stopped.", ClientId); - _packageReceiverTask = null; + if (_isTakeover) + { + try + { + // Don't use SendAsync here _cancellationToken is already cancelled. + await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket + { + ReasonCode = MqttDisconnectReasonCode.SessionTakenOver + }, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, "Client '{0}': Error while sending DISCONNECT packet after takeover.", ClientId); + } + } + + _logger.Info("Client '{0}': Connection stopped.", ClientId); + try { - await _onStop(disconnectType); + await _sessionsManager.CleanUpClient(ClientId, _channelAdapter, disconnectType); } catch (Exception e) { - _logger.Error(e, "client '{0}': Error while cleaning up", ClientId); + _logger.Error(e, "Client '{0}': Error while cleaning up", ClientId); } } } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 84fa7251..19c8478d 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -21,6 +21,7 @@ namespace MQTTnet.Server readonly AsyncLock _createConnectionGate = new AsyncLock(); readonly ConcurrentDictionary _connections = new ConcurrentDictionary(); readonly ConcurrentDictionary _sessions = new ConcurrentDictionary(); + readonly IDictionary _serverSessionItems = new ConcurrentDictionary(); readonly CancellationToken _cancellationToken; @@ -183,7 +184,6 @@ namespace MQTTnet.Server } var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false); - if (!dequeueResult.IsSuccess) { return; @@ -258,8 +258,6 @@ namespace MQTTnet.Server async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) { - string clientId = null; - try { MqttConnectPacket connectPacket; @@ -291,17 +289,9 @@ namespace MQTTnet.Server return; } - clientId = connectPacket.ClientId; - - var connection = await CreateClientConnectionAsync( - connectPacket, - connectionValidatorContext, - channelAdapter, - async () => await _eventDispatcher.SafeNotifyClientConnectedAsync(clientId).ConfigureAwait(false), - async disconnectType => await CleanUpClient(clientId, channelAdapter, disconnectType) - ).ConfigureAwait(false); - - await connection.RunAsync(connectionValidatorContext).ConfigureAwait(false); + var connection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket.ClientId).ConfigureAwait(false); + await connection.RunAsync().ConfigureAwait(false); } catch (OperationCanceledException) { @@ -312,7 +302,7 @@ namespace MQTTnet.Server } } - private async Task CleanUpClient(string clientId, IMqttChannelAdapter channelAdapter, MqttClientDisconnectType disconnectType) + public async Task CleanUpClient(string clientId, IMqttChannelAdapter channelAdapter, MqttClientDisconnectType disconnectType) { if (clientId != null) { @@ -320,11 +310,11 @@ namespace MQTTnet.Server if (disconnectType != MqttClientDisconnectType.Takeover) { _connections.TryRemove(clientId, out _); - } - if (!_options.EnablePersistentSessions) - { - await DeleteSessionAsync(clientId).ConfigureAwait(false); + if (!_options.EnablePersistentSessions) + { + await DeleteSessionAsync(clientId).ConfigureAwait(false); + } } } @@ -364,46 +354,52 @@ namespace MQTTnet.Server return context; } - async Task CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter, Func onStart, Func onStop) + async Task CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter) { + MqttClientConnection connection; + MqttClientConnection existingConnection = null; + using (await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false)) { - var session = _sessions.AddOrUpdate(connectPacket.ClientId, key => - { - _logger.Verbose("Created a new session for client '{0}'.", key); - return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger); - }, (key, existingSession) => - { - if (connectPacket.CleanSession) + var session = _sessions.AddOrUpdate( + connectPacket.ClientId, + key => { - _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId); + _logger.Verbose("Created a new session for client '{0}'.", key); return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger); - } - else + }, + (key, existingSession) => { + if (connectPacket.CleanSession) + { + _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId); + return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger); + } + _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId); return existingSession; - } - }); + }); - var connection = new MqttClientConnection(connectPacket, channelAdapter, session, _options, this, _retainedMessagesManager, onStart, onStop, _rootLogger); - MqttClientConnection existingConnection = null; - _connections.AddOrUpdate(connectPacket.ClientId, key => - { - return connection; - }, (key, tempExistingConnection) => - { - existingConnection = tempExistingConnection; - return connection; - }); + connection = new MqttClientConnection(connectPacket, channelAdapter, session, connectionValidatorContext, _options, this, _retainedMessagesManager, _rootLogger); - if (existingConnection != null) - { - await existingConnection.StopAsync(true).ConfigureAwait(false); - } - - return connection; + _connections.AddOrUpdate( + connectPacket.ClientId, + key => connection, + (key, tempExistingConnection) => + { + existingConnection = tempExistingConnection; + return connection; + }); } + + // Disconnect the client outside of the lock so that new clients can still connect while a single + // one is being disconnected. + if (existingConnection != null) + { + await existingConnection.StopAsync(true).ConfigureAwait(false); + } + + return connection; } async Task InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage) From adce5662ebc052f746c53725707ec1048a06d42d Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 24 Oct 2020 12:22:03 +0200 Subject: [PATCH 6/8] Prepare docs for next release. --- Build/MQTTnet.nuspec | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 16028584..c15d5321 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -12,13 +12,7 @@ true MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol. -* [Client] Fixed wrong value for "ClientWasConnected" in "MqttClientDisconnectedEventArgs" #976 (thanks to @dbeinder). -* [Client] Added direct support for Amazon AWS connections (requires .NET Core 3.1) (thanks to @henning-krause). -* [Client] handle disconnect package and propagate disconnect reason code (thanks to @JanEggers) -* [ManagedClient] Exposed the internal MQTT client. -* [Server] Added client message queue interceptor for QoS level (thanks to @msallin). -* [Server] improved behavior when multiple connections fight over a session (thanks to @JanEggers) -* [Server] Exposed _ClientDisconnectedInterceptor_ (thanks to @SeppPenner). +* [Server] .. Copyright Christian Kratky 2016-2020 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor From c778cd7a0f8844fe71ebe3388231131b9e3ff2fa Mon Sep 17 00:00:00 2001 From: Jay424 Date: Wed, 28 Oct 2020 17:14:09 +0100 Subject: [PATCH 7/8] Fixed evaluation of IgnoreCertificateRevocationErrors to correctly set SslClientAuthenticationOptions.CertificateRevocationCheckMode The if was just flipped. --- Source/MQTTnet/Implementations/MqttTcpChannel.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index 5d96740d..8a349ecc 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -87,7 +87,7 @@ namespace MQTTnet.Implementations ApplicationProtocols = _options.TlsOptions.ApplicationProtocols, ClientCertificates = LoadCertificates(), EnabledSslProtocols = _options.TlsOptions.SslProtocol, - CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.Online : X509RevocationMode.NoCheck, + CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : X509RevocationMode.Online, TargetHost = _options.Server }; From 3ea281ff849c0f79461a678aa2f7fe210f95664d Mon Sep 17 00:00:00 2001 From: Jay424 Date: Wed, 28 Oct 2020 17:58:48 +0100 Subject: [PATCH 8/8] Updated release notes --- Build/MQTTnet.nuspec | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 16028584..2bf6cb71 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -14,7 +14,8 @@ * [Client] Fixed wrong value for "ClientWasConnected" in "MqttClientDisconnectedEventArgs" #976 (thanks to @dbeinder). * [Client] Added direct support for Amazon AWS connections (requires .NET Core 3.1) (thanks to @henning-krause). -* [Client] handle disconnect package and propagate disconnect reason code (thanks to @JanEggers) +* [Client] handle disconnect package and propagate disconnect reason code (thanks to @JanEggers) +* [Client] Fixed an issue when checking for revoked SSL certificates for target platform NETCOREAPP3_1. * [ManagedClient] Exposed the internal MQTT client. * [Server] Added client message queue interceptor for QoS level (thanks to @msallin). * [Server] improved behavior when multiple connections fight over a session (thanks to @JanEggers)