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)