diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 8188da82..a41be4f1 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; @@ -12,10 +13,14 @@ using MQTTnet.Protocol; namespace MQTTnet.Server { - public class MqttClientSessionsManager + public class MqttClientSessionsManager : IDisposable { private readonly BlockingCollection _messageQueue = new BlockingCollection(); - private readonly ConcurrentDictionary _sessions = new ConcurrentDictionary(); + + /// + /// manual locking dictionaries is faster than using concurrent dictionary + /// + private readonly Dictionary _sessions = new Dictionary(); private readonly CancellationToken _cancellationToken; @@ -44,12 +49,15 @@ namespace MQTTnet.Server public void Stop() { - foreach (var session in _sessions) + lock (_sessions) { - session.Value.Stop(MqttClientDisconnectType.NotClean); - } + foreach (var session in _sessions) + { + session.Value.Stop(MqttClientDisconnectType.NotClean); + } - _sessions.Clear(); + _sessions.Clear(); + } } public Task StartSession(IMqttChannelAdapter clientAdapter) @@ -60,10 +68,11 @@ namespace MQTTnet.Server public Task> GetClientStatusAsync() { var result = new List(); - foreach (var session in _sessions) + + foreach (var session in GetSessions()) { - var status = new MqttClientSessionStatus(this, session.Value); - session.Value.FillStatus(status); + var status = new MqttClientSessionStatus(this, session); + session.FillStatus(status); result.Add(status); } @@ -83,12 +92,15 @@ namespace MQTTnet.Server if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - if (!_sessions.TryGetValue(clientId, out var session)) + lock (_sessions) { - throw new InvalidOperationException($"Client session '{clientId}' is unknown."); - } + if (!_sessions.TryGetValue(clientId, out var session)) + { + throw new InvalidOperationException($"Client session '{clientId}' is unknown."); + } - return session.SubscribeAsync(topicFilters); + return session.SubscribeAsync(topicFilters); + } } public Task UnsubscribeAsync(string clientId, IList topicFilters) @@ -96,20 +108,31 @@ namespace MQTTnet.Server if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - if (!_sessions.TryGetValue(clientId, out var session)) + lock (_sessions) { - throw new InvalidOperationException($"Client session '{clientId}' is unknown."); - } + if (!_sessions.TryGetValue(clientId, out var session)) + { + throw new InvalidOperationException($"Client session '{clientId}' is unknown."); + } - return session.UnsubscribeAsync(topicFilters); + return session.UnsubscribeAsync(topicFilters); + } } public void DeleteSession(string clientId) { - _sessions.TryRemove(clientId, out _); + lock (_sessions) + { + _sessions.Remove(clientId); + } _logger.Verbose("Session for client '{0}' deleted.", clientId); } + public void Dispose() + { + _messageQueue?.Dispose(); + } + private void ProcessQueuedApplicationMessages(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) @@ -143,7 +166,7 @@ namespace MQTTnet.Server _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult(); } - foreach (var clientSession in _sessions.Values) + foreach (var clientSession in GetSessions()) { clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket()); } @@ -158,6 +181,14 @@ namespace MQTTnet.Server } } + private List GetSessions() + { + lock (_sessions) + { + return _sessions.Values.ToList(); + } + } + private async Task RunSession(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken) { var clientId = string.Empty; @@ -263,7 +294,7 @@ namespace MQTTnet.Server { if (connectPacket.CleanSession) { - _sessions.TryRemove(connectPacket.ClientId, out _); + _sessions.Remove(connectPacket.ClientId); clientSession.Stop(MqttClientDisconnectType.Clean); clientSession.Dispose(); diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 69850d59..dd732f1b 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -123,6 +123,8 @@ namespace MQTTnet.Server _cancellationTokenSource = null; _retainedMessagesManager = null; + + _clientSessionsManager?.Dispose(); _clientSessionsManager = null; } }