From ebbf50b41f095f44238f4f5ee3516842fe3a822d Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 13 Aug 2022 09:28:16 +0200 Subject: [PATCH] Add storage of creation time. --- .../Server_Retained_Messages_Samples.cs | 3 +- Source/MQTTnet.TestApp/ServerTest.cs | 8 +- Source/MQTTnet.Tests/Server/General.cs | 2 +- .../LoadingRetainedMessagesEventArgs.cs | 2 +- .../Events/RetainedMessageChangeType.cs | 15 ++ .../Events/RetainedMessageChangedEventArgs.cs | 17 +- .../Internal/MqttClientSessionsManager.cs | 4 +- .../Server/Internal/MqttRetainedMessage.cs | 18 ++ .../Internal/MqttRetainedMessagesManager.cs | 243 +++++++++++++----- Source/MQTTnet/Server/MqttServer.cs | 7 +- 10 files changed, 231 insertions(+), 88 deletions(-) create mode 100644 Source/MQTTnet/Server/Events/RetainedMessageChangeType.cs create mode 100644 Source/MQTTnet/Server/Internal/MqttRetainedMessage.cs diff --git a/Samples/Server/Server_Retained_Messages_Samples.cs b/Samples/Server/Server_Retained_Messages_Samples.cs index e392460f..c2a0201c 100644 --- a/Samples/Server/Server_Retained_Messages_Samples.cs +++ b/Samples/Server/Server_Retained_Messages_Samples.cs @@ -7,6 +7,7 @@ // ReSharper disable InconsistentNaming using System.Text.Json; +using MQTTnet.Server; namespace MQTTnet.Samples.Server; @@ -32,7 +33,7 @@ public static class Server_Retained_Messages_Samples { try { - eventArgs.LoadedRetainedMessages = await JsonSerializer.DeserializeAsync>(File.OpenRead(storePath)); + eventArgs.LoadedRetainedMessages = await JsonSerializer.DeserializeAsync>(File.OpenRead(storePath)); Console.WriteLine("Retained messages loaded."); } catch (FileNotFoundException) diff --git a/Source/MQTTnet.TestApp/ServerTest.cs b/Source/MQTTnet.TestApp/ServerTest.cs index e71f9761..e622c388 100644 --- a/Source/MQTTnet.TestApp/ServerTest.cs +++ b/Source/MQTTnet.TestApp/ServerTest.cs @@ -4,12 +4,10 @@ using System; using System.Collections.Generic; -using System.Diagnostics; using System.IO; using System.Text; using System.Threading.Tasks; using MQTTnet.Diagnostics; -using MQTTnet.Extensions.ManagedClient; using MQTTnet.Implementations; using MQTTnet.Protocol; using MQTTnet.Server; @@ -80,15 +78,15 @@ namespace MQTTnet.TestApp mqttServer.LoadingRetainedMessageAsync += e => { - List retainedMessages; + List retainedMessages; if (File.Exists(Filename)) { var json = File.ReadAllText(Filename); - retainedMessages = JsonConvert.DeserializeObject>(json); + retainedMessages = JsonConvert.DeserializeObject>(json); } else { - retainedMessages = new List(); + retainedMessages = new List(); } e.LoadedRetainedMessages = retainedMessages; diff --git a/Source/MQTTnet.Tests/Server/General.cs b/Source/MQTTnet.Tests/Server/General.cs index 2d9cf418..c6d3981d 100644 --- a/Source/MQTTnet.Tests/Server/General.cs +++ b/Source/MQTTnet.Tests/Server/General.cs @@ -394,7 +394,7 @@ namespace MQTTnet.Tests.Server { using (var testEnvironment = CreateTestEnvironment()) { - List savedRetainedMessages = null; + ICollection savedRetainedMessages = null; var s = await testEnvironment.StartServer(); s.RetainedMessageChangedAsync += e => diff --git a/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs b/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs index 12aa0469..a7167fe3 100644 --- a/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs +++ b/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs @@ -9,6 +9,6 @@ namespace MQTTnet.Server { public sealed class LoadingRetainedMessagesEventArgs : EventArgs { - public List LoadedRetainedMessages { get; set; } = new List(); + public ICollection LoadedRetainedMessages { get; set; } = new List(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/RetainedMessageChangeType.cs b/Source/MQTTnet/Server/Events/RetainedMessageChangeType.cs new file mode 100644 index 00000000..e503a38f --- /dev/null +++ b/Source/MQTTnet/Server/Events/RetainedMessageChangeType.cs @@ -0,0 +1,15 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +namespace MQTTnet.Server +{ + public enum RetainedMessageChangeType + { + Created, + + Updated, + + Deleted + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs b/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs index c7beb807..22b9c610 100644 --- a/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs @@ -9,17 +9,24 @@ namespace MQTTnet.Server { public sealed class RetainedMessageChangedEventArgs : EventArgs { - public RetainedMessageChangedEventArgs(string clientId, MqttApplicationMessage changedRetainedMessage, List storedRetainedMessages) + public RetainedMessageChangedEventArgs( + string clientId, + MqttRetainedMessage changedMessage, + RetainedMessageChangeType changeType, + List allMessages) { ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); - ChangedRetainedMessage = changedRetainedMessage ?? throw new ArgumentNullException(nameof(changedRetainedMessage)); - StoredRetainedMessages = storedRetainedMessages ?? throw new ArgumentNullException(nameof(storedRetainedMessages)); + ChangedRetainedMessage = changedMessage ?? throw new ArgumentNullException(nameof(changedMessage)); + ChangeType = changeType; + StoredRetainedMessages = allMessages ?? throw new ArgumentNullException(nameof(allMessages)); } - public MqttApplicationMessage ChangedRetainedMessage { get; } + public RetainedMessageChangeType ChangeType { get; } + + public MqttRetainedMessage ChangedRetainedMessage { get; } public string ClientId { get; } - public List StoredRetainedMessages { get; } + public List StoredRetainedMessages { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index eb2359c7..82539d22 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -58,7 +58,7 @@ namespace MQTTnet.Server _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); } - public async Task CloseAllConnectionsAsync() + public async Task Stop() { List connections; lock (_clients) @@ -131,7 +131,7 @@ namespace MQTTnet.Server { if (applicationMessage.Retain) { - await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false); + await _retainedMessagesManager.HandleMessage(senderId, applicationMessage).ConfigureAwait(false); } var deliveryCount = 0; diff --git a/Source/MQTTnet/Server/Internal/MqttRetainedMessage.cs b/Source/MQTTnet/Server/Internal/MqttRetainedMessage.cs new file mode 100644 index 00000000..fea77f21 --- /dev/null +++ b/Source/MQTTnet/Server/Internal/MqttRetainedMessage.cs @@ -0,0 +1,18 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; + +namespace MQTTnet.Server +{ + public sealed class MqttRetainedMessage + { + public MqttApplicationMessage ApplicationMessage { get; set; } + + /// + /// Gets the timestamp in UTC when this retained message eas published. + /// + public DateTime PublishedTimestamp { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs index acc4672d..c4da295d 100644 --- a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs @@ -3,6 +3,7 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -15,47 +16,42 @@ namespace MQTTnet.Server { public sealed class MqttRetainedMessagesManager { - readonly Dictionary _messages = new Dictionary(4096); - readonly AsyncLock _storageAccessLock = new AsyncLock(); - readonly MqttServerEventContainer _eventContainer; readonly MqttNetSourceLogger _logger; + readonly ConcurrentDictionary _messages = new ConcurrentDictionary(); + readonly AsyncLock _storageAccessLock = new AsyncLock(); + + CancellationTokenSource _cancellationToken; public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger) { _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); - - if (logger == null) throw new ArgumentNullException(nameof(logger)); + + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } + _logger = logger.WithSource(nameof(MqttRetainedMessagesManager)); } - - public async Task Start() - { - try - { - var eventArgs = new LoadingRetainedMessagesEventArgs(); - await _eventContainer.LoadingRetainedMessagesEvent.InvokeAsync(eventArgs).ConfigureAwait(false); - - lock (_messages) - { - _messages.Clear(); - if (eventArgs.LoadedRetainedMessages != null) - { - foreach (var retainedMessage in eventArgs.LoadedRetainedMessages) - { - _messages[retainedMessage.Topic] = retainedMessage; - } - } - } - } - catch (Exception exception) + public async Task ClearMessages() + { + _messages.Clear(); + + using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) { - _logger.Error(exception, "Unhandled exception while loading retained messages."); + await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); } } - public async Task UpdateMessage(string clientId, MqttApplicationMessage applicationMessage) + public Task> GetMessages() + { + var result = new List(_messages.Select(m => m.Value.ApplicationMessage)); + return Task.FromResult((IList)result); + } + + public async Task HandleMessage(string clientId, MqttApplicationMessage applicationMessage) { if (applicationMessage == null) { @@ -64,78 +60,185 @@ namespace MQTTnet.Server try { - List messagesForSave = null; + List allMessages = null; var saveIsRequired = false; - - lock (_messages) + + var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0; + var changeType = RetainedMessageChangeType.Created; + + var retainedMessage = new MqttRetainedMessage { - var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0; + ApplicationMessage = applicationMessage, + PublishedTimestamp = DateTime.UtcNow + }; - if (!hasPayload) - { - saveIsRequired = _messages.Remove(applicationMessage.Topic); - _logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); - } - else - { - if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage)) + MqttRetainedMessage changedMessage; + + if (!hasPayload) + { + // Messages without a payload will remove the message. + changeType = RetainedMessageChangeType.Deleted; + saveIsRequired = _messages.TryRemove(applicationMessage.Topic, out changedMessage); + _logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); + } + else + { + var messageChanged = false; + + _messages.AddOrUpdate( + applicationMessage.Topic, + _ => { - _messages[applicationMessage.Topic] = applicationMessage; + changeType = RetainedMessageChangeType.Created; + changedMessage = retainedMessage; saveIsRequired = true; - } - else + + return retainedMessage; + }, + (_, existingMessage) => { - if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray)) + var hasSamePayload = existingMessage.ApplicationMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray); + var hasSameQoSLevel = existingMessage.ApplicationMessage.QualityOfServiceLevel == applicationMessage.QualityOfServiceLevel; + + if (!hasSameQoSLevel || !hasSamePayload) { - _messages[applicationMessage.Topic] = applicationMessage; + changeType = RetainedMessageChangeType.Updated; saveIsRequired = true; + messageChanged = true; + return retainedMessage; } - } - _logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic); - } + // Keep the existing message. + return existingMessage; + }); - if (saveIsRequired) + if (messageChanged) { - messagesForSave = new List(_messages.Values); + _logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic); } } if (saveIsRequired) { - using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) - { - var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessage, messagesForSave); - await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); - } + allMessages = new List(_messages.Values); + } + + if (saveIsRequired) + { + await SaveMessages(clientId, retainedMessage, changeType, allMessages).ConfigureAwait(false); } } catch (Exception exception) { - _logger.Error(exception, "Unhandled exception while handling retained messages."); - } - } - - public Task> GetMessages() - { - lock (_messages) - { - var result = new List(_messages.Values); - return Task.FromResult((IList)result); + _logger.Error(exception, "Error while handling retained messages."); } } - public async Task ClearMessages() + public async Task Start() { - lock (_messages) + _cancellationToken = new CancellationTokenSource(); + + await LoadMessages().ConfigureAwait(false); + + _ = Task.Run(() => RemoveExpiredMessagesLoop(_cancellationToken.Token), _cancellationToken.Token); + } + + public Task Stop() + { + _cancellationToken?.Cancel(); + _cancellationToken?.Dispose(); + _cancellationToken = null; + + return PlatformAbstractionLayer.CompletedTask; + } + + async Task LoadMessages() + { + if (!_eventContainer.LoadingRetainedMessagesEvent.HasHandlers) { - _messages.Clear(); + return; } + var eventArgs = new LoadingRetainedMessagesEventArgs(); + await _eventContainer.LoadingRetainedMessagesEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + + _messages.Clear(); + + if (eventArgs.LoadedRetainedMessages != null) + { + foreach (var retainedMessage in eventArgs.LoadedRetainedMessages) + { + _messages[retainedMessage.ApplicationMessage.Topic] = retainedMessage; + } + } + } + + async Task RemoveExpiredMessages() + { + var expiredMessages = new List(); + + var now = DateTime.UtcNow; + + foreach (var retainedMessage in _messages) + { + if (retainedMessage.Value.ApplicationMessage.MessageExpiryInterval == 0) + { + // The feature is not used. + continue; + } + + var messageAge = (int)(now - retainedMessage.Value.PublishedTimestamp).TotalSeconds; + if (messageAge < retainedMessage.Value.ApplicationMessage.MessageExpiryInterval) + { + // The message is not expired yet. + continue; + } + + expiredMessages.Add(retainedMessage.Value); + } + + foreach (var expiredMessage in expiredMessages) + { + if (_messages.TryRemove(expiredMessage.ApplicationMessage.Topic, out _)) + { + _logger.Verbose("Server cleared retained message for topic '{0}' due to expiration.", expiredMessage.ApplicationMessage.Topic); + + await SaveMessages(null, expiredMessage, RetainedMessageChangeType.Deleted, new List(_messages.Values)); + } + } + } + + async Task RemoveExpiredMessagesLoop(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + await RemoveExpiredMessages().ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, "Error while removing expired retained messages."); + } + finally + { + // The expiry interval is specified in seconds so that a message may gets expired + // every second. + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false); + } + } + } + + async Task SaveMessages(string clientId, MqttRetainedMessage changedMessage, RetainedMessageChangeType changeType, List allMessages) + { using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) { - await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); + if (_eventContainer.RetainedMessageChangedEvent.HasHandlers) + { + var eventArgs = new RetainedMessageChangedEventArgs(clientId, changedMessage, changeType, allMessages); + await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + } } } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 85b73354..353a8297 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -287,8 +287,9 @@ namespace MQTTnet.Server _cancellationTokenSource.Cancel(false); - await _clientSessionsManager.CloseAllConnectionsAsync().ConfigureAwait(false); - + await _clientSessionsManager.Stop().ConfigureAwait(false); + await _retainedMessagesManager.Stop().ConfigureAwait(false); + foreach (var adapter in _adapters) { adapter.ClientHandler = null; @@ -357,7 +358,7 @@ namespace MQTTnet.Server ThrowIfDisposed(); ThrowIfNotStarted(); - return _retainedMessagesManager?.UpdateMessage(null, retainedMessage); + return _retainedMessagesManager?.HandleMessage(null, retainedMessage); } protected override void Dispose(bool disposing)