From fbf1cbc76e1ceaa09d8d4483ccb35203a0129b21 Mon Sep 17 00:00:00 2001 From: Christian Date: Wed, 16 May 2018 00:08:42 +0200 Subject: [PATCH] Refactor ConnectedClient statistics API --- .../ApplicationBuilderExtensions.cs | 3 +- .../MqttWebSocketServerAdapter.cs | 4 +- .../Adapter/IMqttChannelAdapter.cs | 4 +- .../Adapter/MqttChannelAdapter.cs | 2 + .../Channel/IMqttChannel.cs | 2 + .../Implementations/MqttTcpChannel.Uwp.cs | 2 + .../Implementations/MqttTcpChannel.cs | 2 + .../Implementations/MqttWebSocketChannel.cs | 7 +- .../Internal/TestMqttChannel.cs | 2 + .../Server/ConnectedMqttClient.cs | 22 --- .../Server/IMqttClientSessionStatus.cs | 29 +++ .../MQTTnet.NetStandard/Server/IMqttServer.cs | 2 +- .../Server/MqttClientConnectedEventArgs.cs | 6 +- .../Server/MqttClientDisconnectedEventArgs.cs | 6 +- .../Server/MqttClientKeepAliveMonitor.cs | 26 ++- .../Server/MqttClientPendingMessagesQueue.cs | 28 ++- .../Server/MqttClientSession.cs | 171 ++++++++---------- .../Server/MqttClientSessionStatus.cs | 53 ++++++ .../Server/MqttClientSessionsManager.cs | 152 +++++----------- .../Server/MqttClientSubscriptionsManager.cs | 44 ++--- .../Server/MqttRetainedMessagesManager.cs | 56 ++---- .../MQTTnet.NetStandard/Server/MqttServer.cs | 19 +- .../MqttSubscriptionsManagerTests.cs | 12 +- .../TestMqttCommunicationAdapter.cs | 2 + 24 files changed, 306 insertions(+), 350 deletions(-) delete mode 100644 Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs create mode 100644 Frameworks/MQTTnet.NetStandard/Server/IMqttClientSessionStatus.cs create mode 100644 Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionStatus.cs diff --git a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs index 513712e5..22b9e937 100644 --- a/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs @@ -32,7 +32,8 @@ namespace MQTTnet.AspNetCore var adapter = app.ApplicationServices.GetRequiredService(); using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subProtocol)) { - await adapter.AcceptWebSocketAsync(webSocket); + var endpoint = $"{context.Connection.RemoteIpAddress}:{context.Connection.RemotePort}"; + await adapter.RunWebSocketConnectionAsync(webSocket, endpoint); } }); diff --git a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index a10417ff..dbd93604 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -23,11 +23,11 @@ namespace MQTTnet.AspNetCore return Task.CompletedTask; } - public async Task AcceptWebSocketAsync(WebSocket webSocket) + public async Task RunWebSocketConnectionAsync(WebSocket webSocket, string endpoint) { if (webSocket == null) throw new ArgumentNullException(nameof(webSocket)); - var clientAdapter = new MqttChannelAdapter(new MqttWebSocketChannel(webSocket), new MqttPacketSerializer(), new MqttNetLogger().CreateChildLogger(nameof(MqttWebSocketServerAdapter))); + var clientAdapter = new MqttChannelAdapter(new MqttWebSocketChannel(webSocket, endpoint), new MqttPacketSerializer(), new MqttNetLogger().CreateChildLogger(nameof(MqttWebSocketServerAdapter))); var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(clientAdapter); ClientAccepted?.Invoke(this, eventArgs); diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/IMqttChannelAdapter.cs b/Frameworks/MQTTnet.NetStandard/Adapter/IMqttChannelAdapter.cs index 0970affc..a08d9f67 100644 --- a/Frameworks/MQTTnet.NetStandard/Adapter/IMqttChannelAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Adapter/IMqttChannelAdapter.cs @@ -9,8 +9,10 @@ namespace MQTTnet.Adapter { public interface IMqttChannelAdapter : IDisposable { - IMqttPacketSerializer PacketSerializer { get; } + string Endpoint { get; } + IMqttPacketSerializer PacketSerializer { get; } + Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken); Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken); diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs index 9af60315..ee2f586c 100644 --- a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs @@ -33,6 +33,8 @@ namespace MQTTnet.Adapter _logger = logger.CreateChildLogger(nameof(MqttChannelAdapter)); } + public string Endpoint => _channel.Endpoint; + public IMqttPacketSerializer PacketSerializer { get; } public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) diff --git a/Frameworks/MQTTnet.NetStandard/Channel/IMqttChannel.cs b/Frameworks/MQTTnet.NetStandard/Channel/IMqttChannel.cs index dd9d1f27..6050bcb1 100644 --- a/Frameworks/MQTTnet.NetStandard/Channel/IMqttChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Channel/IMqttChannel.cs @@ -6,6 +6,8 @@ namespace MQTTnet.Channel { public interface IMqttChannel : IDisposable { + string Endpoint { get; } + Task ConnectAsync(CancellationToken cancellationToken); Task DisconnectAsync(); diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs index 19d58240..15f8566e 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs @@ -43,6 +43,8 @@ namespace MQTTnet.Implementations public static Func> CustomIgnorableServerCertificateErrorsResolver { get; set; } + public string Endpoint => _socket?.Information?.RemoteAddress?.ToString(); // TODO: Check if contains also the port. + public async Task ConnectAsync(CancellationToken cancellationToken) { if (_socket == null) diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index fe017810..40edf367 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -41,6 +41,8 @@ namespace MQTTnet.Implementations public static Func CustomCertificateValidationCallback { get; set; } + public string Endpoint => _socket?.RemoteEndPoint?.ToString(); + public async Task ConnectAsync(CancellationToken cancellationToken) { if (_socket == null) diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs index 5ceecbba..733bcdd1 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs @@ -14,17 +14,20 @@ namespace MQTTnet.Implementations private readonly MqttClientWebSocketOptions _options; private WebSocket _webSocket; - + public MqttWebSocketChannel(MqttClientWebSocketOptions options) { _options = options ?? throw new ArgumentNullException(nameof(options)); } - public MqttWebSocketChannel(WebSocket webSocket) + public MqttWebSocketChannel(WebSocket webSocket, string endpoint) { _webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket)); + Endpoint = endpoint; } + public string Endpoint { get; } + public async Task ConnectAsync(CancellationToken cancellationToken) { var uri = _options.Uri; diff --git a/Frameworks/MQTTnet.NetStandard/Internal/TestMqttChannel.cs b/Frameworks/MQTTnet.NetStandard/Internal/TestMqttChannel.cs index b380b085..21d0584b 100644 --- a/Frameworks/MQTTnet.NetStandard/Internal/TestMqttChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Internal/TestMqttChannel.cs @@ -18,6 +18,8 @@ namespace MQTTnet.Core.Internal { } + public string Endpoint { get; } + public Task ConnectAsync(CancellationToken cancellationToken) { return Task.FromResult(0); diff --git a/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs b/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs deleted file mode 100644 index c12759a0..00000000 --- a/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs +++ /dev/null @@ -1,22 +0,0 @@ -using System; -using MQTTnet.Serializer; - -namespace MQTTnet.Server -{ - // TODO: Rename to "RegisteredClient" - // TODO: Add IsConnected - // TODO: Add interface - - public class ConnectedMqttClient - { - public string ClientId { get; set; } - - public MqttProtocolVersion? ProtocolVersion { get; set; } - - public TimeSpan LastPacketReceived { get; set; } - - public TimeSpan LastNonKeepAlivePacketReceived { get; set; } - - public int PendingApplicationMessages { get; set; } - } -} diff --git a/Frameworks/MQTTnet.NetStandard/Server/IMqttClientSessionStatus.cs b/Frameworks/MQTTnet.NetStandard/Server/IMqttClientSessionStatus.cs new file mode 100644 index 00000000..d9c55938 --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/Server/IMqttClientSessionStatus.cs @@ -0,0 +1,29 @@ +using System; +using System.Threading.Tasks; +using MQTTnet.Serializer; + +namespace MQTTnet.Server +{ + public interface IMqttClientSessionStatus + { + string ClientId { get; } + + string Endpoint { get; } + + bool IsConnected { get; } + + MqttProtocolVersion? ProtocolVersion { get; } + + TimeSpan LastPacketReceived { get; } + + TimeSpan LastNonKeepAlivePacketReceived { get; } + + int PendingApplicationMessagesCount { get; } + + Task DisconnectAsync(); + + Task DeleteSessionAsync(); + + Task ClearPendingApplicationMessagesAsync(); + } +} diff --git a/Frameworks/MQTTnet.NetStandard/Server/IMqttServer.cs b/Frameworks/MQTTnet.NetStandard/Server/IMqttServer.cs index 229e79c5..b0fce015 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/IMqttServer.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/IMqttServer.cs @@ -15,7 +15,7 @@ namespace MQTTnet.Server IMqttServerOptions Options { get; } - Task> GetConnectedClientsAsync(); + Task> GetClientSessionsStatusAsync(); Task SubscribeAsync(string clientId, IList topicFilters); Task UnsubscribeAsync(string clientId, IList topicFilters); diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientConnectedEventArgs.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientConnectedEventArgs.cs index 1ca3f406..6f043b44 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientConnectedEventArgs.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientConnectedEventArgs.cs @@ -4,11 +4,11 @@ namespace MQTTnet.Server { public class MqttClientConnectedEventArgs : EventArgs { - public MqttClientConnectedEventArgs(ConnectedMqttClient client) + public MqttClientConnectedEventArgs(string clientId) { - Client = client ?? throw new ArgumentNullException(nameof(client)); + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); } - public ConnectedMqttClient Client { get; } + public string ClientId { get; } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectedEventArgs.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectedEventArgs.cs index 6dd505e5..7e88548e 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectedEventArgs.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectedEventArgs.cs @@ -4,13 +4,13 @@ namespace MQTTnet.Server { public class MqttClientDisconnectedEventArgs : EventArgs { - public MqttClientDisconnectedEventArgs(ConnectedMqttClient client, bool wasCleanDisconnect) + public MqttClientDisconnectedEventArgs(string clientId, bool wasCleanDisconnect) { - Client = client ?? throw new ArgumentNullException(nameof(client)); + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); WasCleanDisconnect = wasCleanDisconnect; } - public ConnectedMqttClient Client { get; } + public string ClientId { get; } public bool WasCleanDisconnect { get; } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs index 46196008..d73b183c 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs @@ -12,18 +12,18 @@ namespace MQTTnet.Server private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch(); private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch(); - private readonly string _clientId; - private readonly Action _timeoutCallback; private readonly IMqttNetChildLogger _logger; + private readonly string _clientId; + private readonly Action _callback; private Task _workerTask; - public MqttClientKeepAliveMonitor(string clientId, Action timeoutCallback, IMqttNetChildLogger logger) + public MqttClientKeepAliveMonitor(string clientId, Action callback, IMqttNetChildLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); _clientId = clientId; - _timeoutCallback = timeoutCallback; + _callback = callback; _logger = logger.CreateChildLogger(nameof(MqttClientKeepAliveMonitor)); } @@ -38,16 +38,13 @@ namespace MQTTnet.Server return; } - _workerTask = Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken).ConfigureAwait(false), cancellationToken); + _workerTask = Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken); } - public void WaitForCompletion() - { - if (_workerTask != null) - { - Task.WaitAll(_workerTask); - } - } + ////public void WaitForCompletion() + ////{ + //// SpinWait.SpinUntil(() => _workerTask == null || _workerTask.IsCanceled || _workerTask.IsCompleted || _workerTask.IsFaulted); + ////} private async Task RunAsync(int keepAlivePeriod, CancellationToken cancellationToken) { @@ -62,9 +59,8 @@ namespace MQTTnet.Server if (_lastPacketReceivedTracker.Elapsed.TotalSeconds > keepAlivePeriod * 1.5D) { _logger.Warning(null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientId); - - _timeoutCallback?.Invoke(); - + _callback(); + return; } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs index fc74d4c5..619ff703 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs @@ -13,12 +13,12 @@ namespace MQTTnet.Server { public sealed class MqttClientPendingMessagesQueue : IDisposable { - private readonly ConcurrentQueue _queue = new ConcurrentQueue(); private readonly AsyncAutoResetEvent _queueAutoResetEvent = new AsyncAutoResetEvent(); private readonly IMqttServerOptions _options; private readonly MqttClientSession _clientSession; private readonly IMqttNetChildLogger _logger; + private ConcurrentQueue _queue = new ConcurrentQueue(); private Task _workerTask; public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetChildLogger logger) @@ -43,15 +43,7 @@ namespace MQTTnet.Server _workerTask = Task.Run(() => SendQueuedPacketsAsync(adapter, cancellationToken), cancellationToken); } - - public void WaitForCompletion() - { - if (_workerTask != null) - { - Task.WaitAll(_workerTask); - } - } - + public void Enqueue(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); @@ -75,6 +67,17 @@ namespace MQTTnet.Server _logger.Verbose("Enqueued packet (ClientId: {0}).", _clientSession.ClientId); } + public void Clear() + { + var newQueue = new ConcurrentQueue(); + Interlocked.Exchange(ref _queue, newQueue); + } + + public void Dispose() + { + _queueAutoResetEvent?.Dispose(); + } + private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken) { try @@ -151,10 +154,5 @@ namespace MQTTnet.Server } } } - - public void Dispose() - { - _queueAutoResetEvent?.Dispose(); - } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs index 1d0b77d4..c42a3007 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs @@ -9,21 +9,26 @@ using MQTTnet.Exceptions; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; -using MQTTnet.Serializer; namespace MQTTnet.Server { public sealed class MqttClientSession : IDisposable { private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); + private readonly MqttRetainedMessagesManager _retainedMessagesManager; + private readonly MqttClientKeepAliveMonitor _keepAliveMonitor; + private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue; + private readonly MqttClientSubscriptionsManager _subscriptionsManager; + private readonly MqttClientSessionsManager _sessionsManager; + private readonly IMqttNetChildLogger _logger; private readonly IMqttServerOptions _options; - private readonly MqttClientSessionsManager _sessionsManager; private CancellationTokenSource _cancellationTokenSource; private MqttApplicationMessage _willMessage; private bool _wasCleanDisconnect; + private IMqttChannelAdapter _adapter; public MqttClientSession( string clientId, @@ -37,27 +42,28 @@ namespace MQTTnet.Server _options = options ?? throw new ArgumentNullException(nameof(options)); _sessionsManager = sessionsManager; _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); - + ClientId = clientId; _logger = logger.CreateChildLogger(nameof(MqttClientSession)); - KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeout, _logger); - SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server); - PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger); + _keepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, () => Stop(MqttClientDisconnectType.NotClean), _logger); + _subscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server); + _pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger); } - public MqttClientSubscriptionsManager SubscriptionsManager { get; } - - public MqttClientPendingMessagesQueue PendingMessagesQueue { get; } - - public MqttClientKeepAliveMonitor KeepAliveMonitor { get; } - public string ClientId { get; } - public MqttProtocolVersion? ProtocolVersion { get; private set; } - - public bool IsConnected { get; private set; } + public void FillStatus(MqttClientSessionStatus status) + { + status.ClientId = ClientId; + status.IsConnected = _adapter != null; + status.Endpoint = _adapter?.Endpoint; + status.ProtocolVersion = _adapter?.PacketSerializer?.ProtocolVersion; + status.PendingApplicationMessagesCount = _pendingMessagesQueue.Count; + status.LastPacketReceived = _keepAliveMonitor.LastPacketReceived; + status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; + } public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) { @@ -66,34 +72,41 @@ namespace MQTTnet.Server try { - _cancellationTokenSource = new CancellationTokenSource(); + _adapter = adapter; + _cancellationTokenSource = new CancellationTokenSource(); _wasCleanDisconnect = false; _willMessage = connectPacket.WillMessage; - IsConnected = true; - ProtocolVersion = adapter.PacketSerializer.ProtocolVersion; + _pendingMessagesQueue.Start(adapter, _cancellationTokenSource.Token); + _keepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token); - PendingMessagesQueue.Start(adapter, _cancellationTokenSource.Token); - KeepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token); - - await ReceivePacketsAsync(adapter, _cancellationTokenSource.Token).ConfigureAwait(false); + while (!_cancellationTokenSource.IsCancellationRequested) + { + var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false); + _keepAliveMonitor.PacketReceived(packet); + await ProcessReceivedPacketAsync(adapter, packet, _cancellationTokenSource.Token).ConfigureAwait(false); + } } catch (OperationCanceledException) { } - catch (MqttCommunicationException exception) - { - _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); - } catch (Exception exception) { - _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); + if (exception is MqttCommunicationException) + { + _logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", ClientId); + } + else + { + _logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId); + } + + Stop(MqttClientDisconnectType.NotClean); } finally { - ProtocolVersion = null; - IsConnected = false; + _adapter = null; _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; @@ -106,7 +119,8 @@ namespace MQTTnet.Server { try { - if (_cancellationTokenSource == null) + var cts = _cancellationTokenSource; + if (cts == null || cts.IsCancellationRequested) { return; } @@ -114,16 +128,16 @@ namespace MQTTnet.Server _wasCleanDisconnect = type == MqttClientDisconnectType.Clean; _cancellationTokenSource?.Cancel(false); - PendingMessagesQueue.WaitForCompletion(); - KeepAliveMonitor.WaitForCompletion(); - var willMessage = _willMessage; - _willMessage = null; // clear willmessage so it is send just once - - if (willMessage != null && !_wasCleanDisconnect) + if (_willMessage != null && !_wasCleanDisconnect) { - _sessionsManager.StartDispatchApplicationMessage(this, willMessage); + _sessionsManager.StartDispatchApplicationMessage(this, _willMessage); } + + _willMessage = null; + + ////_pendingMessagesQueue.WaitForCompletion(); + ////_keepAliveMonitor.WaitForCompletion(); } finally { @@ -131,11 +145,11 @@ namespace MQTTnet.Server } } - public async Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage) + public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage) { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - var result = await SubscriptionsManager.CheckSubscriptionsAsync(applicationMessage).ConfigureAwait(false); + var result = _subscriptionsManager.CheckSubscriptions(applicationMessage); if (!result.IsSubscribed) { return; @@ -149,26 +163,27 @@ namespace MQTTnet.Server publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); } - PendingMessagesQueue.Enqueue(publishPacket); + _pendingMessagesQueue.Enqueue(publishPacket); } - public async Task SubscribeAsync(IList topicFilters) + public Task SubscribeAsync(IList topicFilters) { if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - SubscriptionsManager.Subscribe(new MqttSubscribePacket + _subscriptionsManager.Subscribe(new MqttSubscribePacket { TopicFilters = topicFilters }); - await EnqueueSubscribedRetainedMessagesAsync(topicFilters).ConfigureAwait(false); + EnqueueSubscribedRetainedMessages(topicFilters); + return Task.FromResult(0); } public Task UnsubscribeAsync(IList topicFilters) { if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - SubscriptionsManager.Unsubscribe(new MqttUnsubscribePacket + _subscriptionsManager.Unsubscribe(new MqttUnsubscribePacket { TopicFilters = topicFilters }); @@ -176,49 +191,18 @@ namespace MQTTnet.Server return Task.FromResult(0); } + public void ClearPendingApplicationMessages() + { + _pendingMessagesQueue.Clear(); + } + public void Dispose() { - SubscriptionsManager?.Dispose(); - PendingMessagesQueue?.Dispose(); + _pendingMessagesQueue?.Dispose(); _cancellationTokenSource?.Dispose(); } - private void StopDueToKeepAliveTimeout() - { - _logger.Info("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId); - Stop(MqttClientDisconnectType.NotClean); - } - - private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken) - { - try - { - while (!cancellationToken.IsCancellationRequested) - { - var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false); - KeepAliveMonitor.PacketReceived(packet); - await ProcessReceivedPacketAsync(adapter, packet, cancellationToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) - { - } - catch (Exception exception) - { - if (exception is MqttCommunicationException) - { - _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); - } - else - { - _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); - } - - Stop(MqttClientDisconnectType.NotClean); - } - } - private Task ProcessReceivedPacketAsync(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken) { if (packet is MqttPublishPacket publishPacket) @@ -276,22 +260,21 @@ namespace MQTTnet.Server _logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); Stop(MqttClientDisconnectType.NotClean); - return Task.FromResult(0); } - private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection topicFilters) + private void EnqueueSubscribedRetainedMessages(ICollection topicFilters) { - var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters); + var retainedMessages = _retainedMessagesManager.GetSubscribedMessages(topicFilters); foreach (var applicationMessage in retainedMessages) { - await EnqueueApplicationMessageAsync(applicationMessage).ConfigureAwait(false); + EnqueueApplicationMessage(applicationMessage); } } private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { - var subscribeResult = SubscriptionsManager.Subscribe(subscribePacket); + var subscribeResult = _subscriptionsManager.Subscribe(subscribePacket); await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { subscribeResult.ResponsePacket }, cancellationToken).ConfigureAwait(false); if (subscribeResult.CloseConnection) @@ -300,13 +283,13 @@ namespace MQTTnet.Server return; } - await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false); + EnqueueSubscribedRetainedMessages(subscribePacket.TopicFilters); } - private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) + private Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { - var unsubscribeResult = SubscriptionsManager.Unsubscribe(unsubscribePacket); - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { unsubscribeResult }, cancellationToken); + var unsubscribeResult = _subscriptionsManager.Unsubscribe(unsubscribePacket); + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { unsubscribeResult }, cancellationToken); } private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -335,21 +318,21 @@ namespace MQTTnet.Server } } - private async Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) + private Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) { _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage); var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }; - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken).ConfigureAwait(false); + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken); } - private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) + private Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) { // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage); var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }; - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken).ConfigureAwait(false); + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken); } private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionStatus.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionStatus.cs new file mode 100644 index 00000000..4d83357e --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionStatus.cs @@ -0,0 +1,53 @@ +using System; +using System.Threading.Tasks; +using MQTTnet.Serializer; + +namespace MQTTnet.Server +{ + public class MqttClientSessionStatus : IMqttClientSessionStatus + { + private readonly MqttClientSessionsManager _sessionsManager; + private readonly MqttClientSession _session; + + public MqttClientSessionStatus(MqttClientSessionsManager sessionsManager, MqttClientSession session) + { + _sessionsManager = sessionsManager; + _session = session; + } + + public string ClientId { get; set; } + public string Endpoint { get; set; } + public bool IsConnected { get; set; } + public MqttProtocolVersion? ProtocolVersion { get; set; } + public TimeSpan LastPacketReceived { get; set; } + public TimeSpan LastNonKeepAlivePacketReceived { get; set; } + public int PendingApplicationMessagesCount { get; set; } + + public Task DisconnectAsync() + { + _session.Stop(MqttClientDisconnectType.NotClean); + return Task.FromResult(0); + } + + public Task DeleteSessionAsync() + { + try + { + _session.Stop(MqttClientDisconnectType.NotClean); + } + finally + { + _sessionsManager.DeleteSession(ClientId); + } + + return Task.FromResult(0); + } + + public Task ClearPendingApplicationMessagesAsync() + { + _session.ClearPendingApplicationMessages(); + + return Task.FromResult(0); + } + } +} diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index 267c8e87..5282dc74 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -1,6 +1,6 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; -using System.Linq; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; @@ -14,8 +14,8 @@ namespace MQTTnet.Server { public sealed class MqttClientSessionsManager : IDisposable { - private readonly Dictionary _sessions = new Dictionary(); - private readonly AsyncLock _sessionsLock = new AsyncLock(); + private readonly ConcurrentDictionary _sessions = new ConcurrentDictionary(); + private readonly AsyncLock _sessionPreparationLock = new AsyncLock(); private readonly MqttRetainedMessagesManager _retainedMessagesManager; private readonly IMqttServerOptions _options; @@ -38,7 +38,6 @@ namespace MQTTnet.Server { var clientId = string.Empty; var wasCleanDisconnect = false; - MqttClientSession clientSession = null; try { @@ -69,7 +68,7 @@ namespace MQTTnet.Server } var result = await PrepareClientSessionAsync(connectPacket).ConfigureAwait(false); - clientSession = result.Session; + var clientSession = result.Session; await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { @@ -80,11 +79,7 @@ namespace MQTTnet.Server } }, cancellationToken).ConfigureAwait(false); - Server.OnClientConnected(new ConnectedMqttClient - { - ClientId = clientId, - ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion - }); + Server.OnClientConnected(clientId); wasCleanDisconnect = await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false); } @@ -109,55 +104,36 @@ namespace MQTTnet.Server if (!_options.EnablePersistentSessions) { - await DeleteSessionAsync(clientId).ConfigureAwait(false); + DeleteSession(clientId); } - Server.OnClientDisconnected(new ConnectedMqttClient - { - ClientId = clientId, - ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion, - PendingApplicationMessages = clientSession?.PendingMessagesQueue.Count ?? 0 - }, - wasCleanDisconnect); + Server.OnClientDisconnected(clientId, wasCleanDisconnect); } } - public async Task StopAsync() + public Task StopAsync() { - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + foreach (var session in _sessions) { - foreach (var session in _sessions) - { - session.Value.Stop(MqttClientDisconnectType.NotClean); - } + session.Value.Stop(MqttClientDisconnectType.NotClean); + } - _sessions.Clear(); - } - finally - { - _sessionsLock.Exit(); - } + _sessions.Clear(); + return Task.FromResult(0); } - public async Task> GetConnectedClientsAsync() + public Task> GetClientStatusAsync() { - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + var result = new List(); + foreach (var session in _sessions) { - return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient - { - ClientId = s.Value.ClientId, - ProtocolVersion = s.Value.ProtocolVersion, - LastPacketReceived = s.Value.KeepAliveMonitor.LastPacketReceived, - LastNonKeepAlivePacketReceived = s.Value.KeepAliveMonitor.LastNonKeepAlivePacketReceived, - PendingApplicationMessages = s.Value.PendingMessagesQueue.Count - }).ToList(); - } - finally - { - _sessionsLock.Exit(); + var status = new MqttClientSessionStatus(this, session.Value); + session.Value.FillStatus(status); + + result.Add(status); } + + return Task.FromResult((IList)result); } public void StartDispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) @@ -165,51 +141,41 @@ namespace MQTTnet.Server Task.Run(() => DispatchApplicationMessageAsync(senderClientSession, applicationMessage)); } - public async Task SubscribeAsync(string clientId, IList topicFilters) + public Task SubscribeAsync(string clientId, IList topicFilters) { if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + if (!_sessions.TryGetValue(clientId, out var session)) { - if (!_sessions.TryGetValue(clientId, out var session)) - { - throw new InvalidOperationException($"Client session '{clientId}' is unknown."); - } + throw new InvalidOperationException($"Client session '{clientId}' is unknown."); + } - await session.SubscribeAsync(topicFilters).ConfigureAwait(false); - } - finally - { - _sessionsLock.Exit(); - } + return session.SubscribeAsync(topicFilters); } - public async Task UnsubscribeAsync(string clientId, IList topicFilters) + public Task UnsubscribeAsync(string clientId, IList topicFilters) { if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + if (!_sessions.TryGetValue(clientId, out var session)) { - if (!_sessions.TryGetValue(clientId, out var session)) - { - throw new InvalidOperationException($"Client session '{clientId}' is unknown."); - } + throw new InvalidOperationException($"Client session '{clientId}' is unknown."); + } - await session.UnsubscribeAsync(topicFilters).ConfigureAwait(false); - } - finally - { - _sessionsLock.Exit(); - } + return session.UnsubscribeAsync(topicFilters); + } + + public void DeleteSession(string clientId) + { + _sessions.TryRemove(clientId, out _); + _logger.Verbose("Session for client '{0}' deleted.", clientId); } public void Dispose() { - _sessionsLock?.Dispose(); + _sessionPreparationLock?.Dispose(); } private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket) @@ -229,23 +195,9 @@ namespace MQTTnet.Server return context.ReturnCode; } - private async Task DeleteSessionAsync(string clientId) - { - await _sessionsLock.EnterAsync(CancellationToken.None); - try - { - _sessions.Remove(clientId); - _logger.Verbose("Session for client '{0}' deleted.", clientId); - } - finally - { - _sessionsLock.Exit(); - } - } - private async Task PrepareClientSessionAsync(MqttConnectPacket connectPacket) { - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); + await _sessionPreparationLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); try { var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession); @@ -253,7 +205,7 @@ namespace MQTTnet.Server { if (connectPacket.CleanSession) { - _sessions.Remove(connectPacket.ClientId); + _sessions.TryRemove(connectPacket.ClientId, out _); clientSession.Stop(MqttClientDisconnectType.Clean); clientSession.Dispose(); @@ -282,7 +234,7 @@ namespace MQTTnet.Server } finally { - _sessionsLock.Exit(); + _sessionPreparationLock.Exit(); } } @@ -301,30 +253,22 @@ namespace MQTTnet.Server return; } + Server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage); + if (applicationMessage.Retain) { await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false); } - Server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage); + foreach (var clientSession in _sessions.Values) + { + clientSession.EnqueueApplicationMessage(applicationMessage); + } } catch (Exception exception) { _logger.Error(exception, "Error while processing application message"); } - - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try - { - foreach (var clientSession in _sessions.Values) - { - await clientSession.EnqueueApplicationMessageAsync(applicationMessage).ConfigureAwait(false); - } - } - finally - { - _sessionsLock.Exit(); - } } private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs index dfb94638..6dfa98e3 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs @@ -2,17 +2,14 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; -using System.Threading; -using System.Threading.Tasks; using MQTTnet.Packets; using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class MqttClientSubscriptionsManager : IDisposable + public sealed class MqttClientSubscriptionsManager { private readonly ConcurrentDictionary _subscriptions = new ConcurrentDictionary(); - private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1); private readonly IMqttServerOptions _options; private readonly MqttServer _server; private readonly string _clientId; @@ -81,43 +78,30 @@ namespace MQTTnet.Server }; } - public async Task CheckSubscriptionsAsync(MqttApplicationMessage applicationMessage) + public CheckSubscriptionsResult CheckSubscriptions(MqttApplicationMessage applicationMessage) { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - await _semaphore.WaitAsync().ConfigureAwait(false); - try + var qosLevels = new HashSet(); + foreach (var subscription in _subscriptions) { - var qosLevels = new HashSet(); - foreach (var subscription in _subscriptions) + if (!MqttTopicFilterComparer.IsMatch(applicationMessage.Topic, subscription.Key)) { - if (!MqttTopicFilterComparer.IsMatch(applicationMessage.Topic, subscription.Key)) - { - continue; - } - - qosLevels.Add(subscription.Value); + continue; } - if (qosLevels.Count == 0) - { - return new CheckSubscriptionsResult - { - IsSubscribed = false - }; - } - - return CreateSubscriptionResult(applicationMessage, qosLevels); + qosLevels.Add(subscription.Value); } - finally + + if (qosLevels.Count == 0) { - _semaphore.Release(); + return new CheckSubscriptionsResult + { + IsSubscribed = false + }; } - } - public void Dispose() - { - _semaphore?.Dispose(); + return CreateSubscriptionResult(applicationMessage, qosLevels); } private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel) diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs index be0a5b7b..f2ab9448 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs @@ -1,17 +1,15 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; -using System.Threading; using System.Threading.Tasks; using MQTTnet.Diagnostics; -using MQTTnet.Internal; namespace MQTTnet.Server { - public sealed class MqttRetainedMessagesManager : IDisposable + public sealed class MqttRetainedMessagesManager { - private readonly Dictionary _messages = new Dictionary(); - private readonly AsyncLock _messagesLock = new AsyncLock(); + private readonly ConcurrentDictionary _messages = new ConcurrentDictionary(); private readonly IMqttNetChildLogger _logger; private readonly IMqttServerOptions _options; @@ -29,7 +27,6 @@ namespace MQTTnet.Server return; } - await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); try { var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false); @@ -44,84 +41,61 @@ namespace MQTTnet.Server { _logger.Error(exception, "Unhandled exception while loading retained messages."); } - finally - { - _messagesLock.Exit(); - } } public async Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage) { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); try { - await HandleMessageInternalAsync(clientId, applicationMessage); + await HandleMessageInternalAsync(clientId, applicationMessage).ConfigureAwait(false); } catch (Exception exception) { _logger.Error(exception, "Unhandled exception while handling retained messages."); } - finally - { - _messagesLock.Exit(); - } } - public async Task> GetSubscribedMessagesAsync(ICollection topicFilters) + public IList GetSubscribedMessages(ICollection topicFilters) { var retainedMessages = new List(); - await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + foreach (var retainedMessage in _messages.Values) { - foreach (var retainedMessage in _messages.Values) + foreach (var topicFilter in topicFilters) { - foreach (var topicFilter in topicFilters) + if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic)) { - if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic)) - { - continue; - } - - retainedMessages.Add(retainedMessage); - break; + continue; } + + retainedMessages.Add(retainedMessage); + break; } } - finally - { - _messagesLock.Exit(); - } return retainedMessages; } - public void Dispose() - { - _messagesLock?.Dispose(); - } - private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage) { var saveIsRequired = false; - if (applicationMessage.Payload?.Any() == false) + if (applicationMessage.Payload?.Length > 0) { - saveIsRequired = _messages.Remove(applicationMessage.Topic); + saveIsRequired = _messages.TryRemove(applicationMessage.Topic, out _); _logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); } else { - if (!_messages.ContainsKey(applicationMessage.Topic)) + if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage)) { _messages[applicationMessage.Topic] = applicationMessage; saveIsRequired = true; } else { - var existingMessage = _messages[applicationMessage.Topic]; if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0])) { _messages[applicationMessage.Topic] = applicationMessage; diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs index f7f506dc..8004afa2 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs @@ -37,9 +37,9 @@ namespace MQTTnet.Server public IMqttServerOptions Options { get; private set; } - public Task> GetConnectedClientsAsync() + public Task> GetClientSessionsStatusAsync() { - return _clientSessionsManager.GetConnectedClientsAsync(); + return _clientSessionsManager.GetClientStatusAsync(); } public Task SubscribeAsync(string clientId, IList topicFilters) @@ -120,24 +120,23 @@ namespace MQTTnet.Server finally { _clientSessionsManager?.Dispose(); - _retainedMessagesManager?.Dispose(); - + _cancellationTokenSource = null; _retainedMessagesManager = null; _clientSessionsManager = null; } } - internal void OnClientConnected(ConnectedMqttClient client) + internal void OnClientConnected(string clientId) { - _logger.Info("Client '{0}': Connected.", client.ClientId); - ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client)); + _logger.Info("Client '{0}': Connected.", clientId); + ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientId)); } - internal void OnClientDisconnected(ConnectedMqttClient client, bool wasCleanDisconnect) + internal void OnClientDisconnected(string clientId, bool wasCleanDisconnect) { - _logger.Info("Client '{0}': Disconnected (clean={1}).", client.ClientId, wasCleanDisconnect); - ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client, wasCleanDisconnect)); + _logger.Info("Client '{0}': Disconnected (clean={1}).", clientId, wasCleanDisconnect); + ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientId, wasCleanDisconnect)); } internal void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter) diff --git a/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs b/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs index cf922a15..aa84dd3e 100644 --- a/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs @@ -26,7 +26,7 @@ namespace MQTTnet.Core.Tests QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }; - var result = sm.CheckSubscriptionsAsync(pp).Result; + var result = sm.CheckSubscriptions(pp); Assert.IsTrue(result.IsSubscribed); Assert.AreEqual(result.QualityOfServiceLevel, MqttQualityOfServiceLevel.AtMostOnce); } @@ -47,7 +47,7 @@ namespace MQTTnet.Core.Tests QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce }; - var result = sm.CheckSubscriptionsAsync(pp).Result; + var result = sm.CheckSubscriptions(pp); Assert.IsTrue(result.IsSubscribed); Assert.AreEqual(result.QualityOfServiceLevel, MqttQualityOfServiceLevel.AtMostOnce); } @@ -69,7 +69,7 @@ namespace MQTTnet.Core.Tests QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce }; - var result = sm.CheckSubscriptionsAsync(pp).Result; + var result = sm.CheckSubscriptions(pp); Assert.IsTrue(result.IsSubscribed); Assert.AreEqual(result.QualityOfServiceLevel, MqttQualityOfServiceLevel.AtLeastOnce); } @@ -90,7 +90,7 @@ namespace MQTTnet.Core.Tests QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }; - Assert.IsFalse(sm.CheckSubscriptionsAsync(pp).Result.IsSubscribed); + Assert.IsFalse(sm.CheckSubscriptions(pp).IsSubscribed); } [TestMethod] @@ -109,13 +109,13 @@ namespace MQTTnet.Core.Tests QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }; - Assert.IsTrue(sm.CheckSubscriptionsAsync(pp).Result.IsSubscribed); + Assert.IsTrue(sm.CheckSubscriptions(pp).IsSubscribed); var up = new MqttUnsubscribePacket(); up.TopicFilters.Add("A/B/C"); sm.Unsubscribe(up); - Assert.IsFalse(sm.CheckSubscriptionsAsync(pp).Result.IsSubscribed); + Assert.IsFalse(sm.CheckSubscriptions(pp).IsSubscribed); } } } diff --git a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs index 837221b1..3b2dd1fb 100644 --- a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs +++ b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs @@ -15,6 +15,8 @@ namespace MQTTnet.Core.Tests public TestMqttCommunicationAdapter Partner { get; set; } + public string Endpoint { get; } + public IMqttPacketSerializer PacketSerializer { get; } = new MqttPacketSerializer(); public void Dispose()