From 1311535eed520a1eb98ad3e3011c27f452e34225 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 11 Mar 2019 20:15:12 +0100 Subject: [PATCH] Added support for persisted retained messages. --- .../RetainedApplicationMessagesModel.cs | 11 ++ .../Configuration/SettingsModel.cs | 9 +- .../MQTTnet.Server/Mqtt/MqttServerService.cs | 16 ++- .../MQTTnet.Server/Mqtt/MqttServerStorage.cs | 112 +++++++++++++++++- .../RetainedApplicationMessages.json | 1 + Source/MQTTnet.Server/Startup.cs | 2 + Source/MQTTnet.Server/appsettings.json | 2 +- 7 files changed, 139 insertions(+), 14 deletions(-) create mode 100644 Source/MQTTnet.Server/Configuration/RetainedApplicationMessagesModel.cs create mode 100644 Source/MQTTnet.Server/RetainedApplicationMessages.json diff --git a/Source/MQTTnet.Server/Configuration/RetainedApplicationMessagesModel.cs b/Source/MQTTnet.Server/Configuration/RetainedApplicationMessagesModel.cs new file mode 100644 index 00000000..7abae83f --- /dev/null +++ b/Source/MQTTnet.Server/Configuration/RetainedApplicationMessagesModel.cs @@ -0,0 +1,11 @@ +namespace MQTTnet.Server.Configuration +{ + public class RetainedApplicationMessagesModel + { + public bool Persist { get; set; } = false; + + public int WriteInterval { get; set; } = 10; + + public string Filename { get; set; } = "RetainedApplicationMessages.json"; + } +} diff --git a/Source/MQTTnet.Server/Configuration/SettingsModel.cs b/Source/MQTTnet.Server/Configuration/SettingsModel.cs index 42c660e2..228ae1dc 100644 --- a/Source/MQTTnet.Server/Configuration/SettingsModel.cs +++ b/Source/MQTTnet.Server/Configuration/SettingsModel.cs @@ -5,10 +5,6 @@ /// public class SettingsModel { - public SettingsModel() - { - } - /// /// Set default connection timeout in seconds /// @@ -43,5 +39,10 @@ /// Set limit for max pending messages per client /// public int MaxPendingMessagesPerClient { get; set; } = 250; + + /// + /// The settings for retained messages. + /// + public RetainedApplicationMessagesModel RetainedApplicationMessages { get; set; } = new RetainedApplicationMessagesModel(); } } \ No newline at end of file diff --git a/Source/MQTTnet.Server/Mqtt/MqttServerService.cs b/Source/MQTTnet.Server/Mqtt/MqttServerService.cs index 44a93702..f4b7a034 100644 --- a/Source/MQTTnet.Server/Mqtt/MqttServerService.cs +++ b/Source/MQTTnet.Server/Mqtt/MqttServerService.cs @@ -11,6 +11,7 @@ using MQTTnet.Adapter; using MQTTnet.AspNetCore; using MQTTnet.Implementations; using MQTTnet.Protocol; +using MQTTnet.Server.Configuration; using MQTTnet.Server.Scripting; namespace MQTTnet.Server.Mqtt @@ -19,8 +20,9 @@ namespace MQTTnet.Server.Mqtt { private readonly ILogger _logger; - private readonly Configuration.SettingsModel _settings; + private readonly SettingsModel _settings; private readonly MqttApplicationMessageInterceptor _mqttApplicationMessageInterceptor; + private readonly MqttServerStorage _mqttServerStorage; private readonly MqttClientConnectedHandler _mqttClientConnectedHandler; private readonly MqttClientDisconnectedHandler _mqttClientDisconnectedHandler; private readonly MqttClientSubscribedTopicHandler _mqttClientSubscribedTopicHandler; @@ -32,7 +34,7 @@ namespace MQTTnet.Server.Mqtt private readonly MqttWebSocketServerAdapter _webSocketServerAdapter; public MqttServerService( - Configuration.SettingsModel settings, + SettingsModel settings, CustomMqttFactory mqttFactory, MqttClientConnectedHandler mqttClientConnectedHandler, MqttClientDisconnectedHandler mqttClientDisconnectedHandler, @@ -41,10 +43,11 @@ namespace MQTTnet.Server.Mqtt MqttConnectionValidator mqttConnectionValidator, MqttSubscriptionInterceptor mqttSubscriptionInterceptor, MqttApplicationMessageInterceptor mqttApplicationMessageInterceptor, - PythonScriptHostService pythonScriptHostService, + MqttServerStorage mqttServerStorage, + PythonScriptHostService pythonScriptHostService, ILogger logger) { - _settings = settings; + _settings = settings ?? throw new ArgumentNullException(nameof(settings)); _mqttClientConnectedHandler = mqttClientConnectedHandler ?? throw new ArgumentNullException(nameof(mqttClientConnectedHandler)); _mqttClientDisconnectedHandler = mqttClientDisconnectedHandler ?? throw new ArgumentNullException(nameof(mqttClientDisconnectedHandler)); _mqttClientSubscribedTopicHandler = mqttClientSubscribedTopicHandler ?? throw new ArgumentNullException(nameof(mqttClientSubscribedTopicHandler)); @@ -52,6 +55,7 @@ namespace MQTTnet.Server.Mqtt _mqttConnectionValidator = mqttConnectionValidator ?? throw new ArgumentNullException(nameof(mqttConnectionValidator)); _mqttSubscriptionInterceptor = mqttSubscriptionInterceptor ?? throw new ArgumentNullException(nameof(mqttSubscriptionInterceptor)); _mqttApplicationMessageInterceptor = mqttApplicationMessageInterceptor ?? throw new ArgumentNullException(nameof(mqttApplicationMessageInterceptor)); + _mqttServerStorage = mqttServerStorage ?? throw new ArgumentNullException(nameof(mqttServerStorage)); _pythonScriptHostService = pythonScriptHostService ?? throw new ArgumentNullException(nameof(pythonScriptHostService)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); @@ -70,6 +74,8 @@ namespace MQTTnet.Server.Mqtt { _pythonScriptHostService.RegisterProxyObject("publish", new Action(Publish)); + _mqttServerStorage.Configure(); + _mqttServer.ClientConnectedHandler = _mqttClientConnectedHandler; _mqttServer.ClientDisconnectedHandler = _mqttClientDisconnectedHandler; _mqttServer.ClientSubscribedTopicHandler = _mqttClientSubscribedTopicHandler; @@ -139,7 +145,7 @@ namespace MQTTnet.Server.Mqtt .WithConnectionValidator(_mqttConnectionValidator) .WithApplicationMessageInterceptor(_mqttApplicationMessageInterceptor) .WithSubscriptionInterceptor(_mqttSubscriptionInterceptor) - .WithStorage(new MqttServerStorage(_settings)); + .WithStorage(_mqttServerStorage); // Configure unencrypted connections if (_settings.TcpEndPoint.Enabled) diff --git a/Source/MQTTnet.Server/Mqtt/MqttServerStorage.cs b/Source/MQTTnet.Server/Mqtt/MqttServerStorage.cs index 562d362d..39b65d51 100644 --- a/Source/MQTTnet.Server/Mqtt/MqttServerStorage.cs +++ b/Source/MQTTnet.Server/Mqtt/MqttServerStorage.cs @@ -1,27 +1,131 @@ using System; using System.Collections.Generic; +using System.IO; +using System.Text; +using System.Threading; using System.Threading.Tasks; +using Microsoft.Extensions.Logging; using MQTTnet.Server.Configuration; +using Newtonsoft.Json; namespace MQTTnet.Server.Mqtt { public class MqttServerStorage : IMqttServerStorage { - private readonly SettingsModel _settings; + private readonly List _messages = new List(); - public MqttServerStorage(SettingsModel settings) + private readonly SettingsModel _settings; + private readonly ILogger _logger; + + private string _filename; + private bool _messagesHaveChanged; + + public MqttServerStorage(SettingsModel settings, ILogger logger) { _settings = settings ?? throw new ArgumentNullException(nameof(settings)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + } + + public void Configure() + { + if (_settings.RetainedApplicationMessages?.Persist != true) + { + _logger.LogInformation("Persisting of retained application messages is disabled."); + return; + } + + _filename = ExpandFilename(); + + // The retained application messages are stored in a separate thread. + // This is mandatory because writing them to a slow storage (like RaspberryPi SD card) + // will slow down the whole message processing speed. + Task.Run(SaveRetainedMessagesInternalAsync, CancellationToken.None); } public Task SaveRetainedMessagesAsync(IList messages) { + lock (_messages) + { + _messages.Clear(); + _messages.AddRange(messages); + + _messagesHaveChanged = true; + } + return Task.CompletedTask; } - public Task> LoadRetainedMessagesAsync() + private async Task SaveRetainedMessagesInternalAsync() { - return Task.FromResult>(null); + while (true) + { + try + { + await Task.Delay(TimeSpan.FromSeconds(_settings.RetainedApplicationMessages.WriteInterval)).ConfigureAwait(false); + + List messages; + lock (_messages) + { + if (!_messagesHaveChanged) + { + continue; + } + + messages = new List(_messages); + _messagesHaveChanged = false; + } + + var json = JsonConvert.SerializeObject(messages); + await File.WriteAllTextAsync(_filename, json, Encoding.UTF8).ConfigureAwait(false); + + _logger.LogInformation($"{messages.Count} retained MQTT messages written."); + } + catch (Exception exception) + { + _logger.LogError(exception, "Error while writing retained MQTT messages."); + } + } + } + + public async Task> LoadRetainedMessagesAsync() + { + if (_settings.RetainedApplicationMessages?.Persist != true) + { + return null; + } + + if (!File.Exists(_filename)) + { + return null; + } + + try + { + var json = await File.ReadAllTextAsync(_filename).ConfigureAwait(false); + var applicationMessages = JsonConvert.DeserializeObject>(json); + + _logger.LogInformation($"{applicationMessages.Count} retained MQTT messages loaded."); + + return applicationMessages; + } + catch (Exception exception) + { + _logger.LogWarning(exception, "Error while loading persisted retained application messages."); + return null; + } + } + + private string ExpandFilename() + { + var filename = _settings.RetainedApplicationMessages.Filename; + + var uri = new Uri(filename, UriKind.RelativeOrAbsolute); + if (!uri.IsAbsoluteUri) + { + filename = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, filename); + } + + return filename; } } } diff --git a/Source/MQTTnet.Server/RetainedApplicationMessages.json b/Source/MQTTnet.Server/RetainedApplicationMessages.json new file mode 100644 index 00000000..04c16db3 --- /dev/null +++ b/Source/MQTTnet.Server/RetainedApplicationMessages.json @@ -0,0 +1 @@ +[{"Topic":"a","Payload":"YWFzYXNhc2E=","QualityOfServiceLevel":0,"Retain":true,"UserProperties":[],"ContentType":null,"ResponseTopic":null,"PayloadFormatIndicator":null,"MessageExpiryInterval":null,"TopicAlias":null,"CorrelationData":null,"SubscriptionIdentifier":null},{"Topic":"fgdfgd","Payload":"YWFzYXNhc2E=","QualityOfServiceLevel":0,"Retain":true,"UserProperties":[],"ContentType":null,"ResponseTopic":null,"PayloadFormatIndicator":null,"MessageExpiryInterval":null,"TopicAlias":null,"CorrelationData":null,"SubscriptionIdentifier":null}] \ No newline at end of file diff --git a/Source/MQTTnet.Server/Startup.cs b/Source/MQTTnet.Server/Startup.cs index ebdf5cc5..b50e9409 100644 --- a/Source/MQTTnet.Server/Startup.cs +++ b/Source/MQTTnet.Server/Startup.cs @@ -53,6 +53,7 @@ namespace MQTTnet.Server dataSharingService.Configure(); pythonScriptHostService.Configure(); + mqttServerService.Configure(); } @@ -69,6 +70,7 @@ namespace MQTTnet.Server services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/Source/MQTTnet.Server/appsettings.json b/Source/MQTTnet.Server/appsettings.json index c7419e7b..ce93bddc 100644 --- a/Source/MQTTnet.Server/appsettings.json +++ b/Source/MQTTnet.Server/appsettings.json @@ -41,7 +41,7 @@ "EnablePersistentSessions": false, "MaxPendingMessagesPerClient": 250, "RetainedApplicationMessages": { - "Persist": false, + "Persist": true, "Filename": "RetainedApplicationMessages.json", "WriteInterval": 10 // In seconds. }