Added support for persisted retained messages.

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";
+    }
+}

Source/MQTTnet.Server/Configuration/SettingsModel.cs

@@ -5,10 +5,6 @@
     /// </summary>
     public class SettingsModel
     {
-        public SettingsModel()
-        {
-        }
-
         /// <summary>
         /// Set default connection timeout in seconds
         /// </summary>
@@ -43,5 +39,10 @@
         /// Set limit for max pending messages per client
         /// </summary>
         public int MaxPendingMessagesPerClient { get; set; } = 250;
+
+        /// <summary>
+        /// The settings for retained messages.
+        /// </summary>
+        public RetainedApplicationMessagesModel RetainedApplicationMessages { get; set; } = new RetainedApplicationMessagesModel();
     }
 }

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<MqttServerService> _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<MqttServerService> 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<PythonDictionary>(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)

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<MqttApplicationMessage> _messages = new List<MqttApplicationMessage>();
 
-        public MqttServerStorage(SettingsModel settings)
+        private readonly SettingsModel _settings;
+        private readonly ILogger<MqttServerStorage> _logger;
+
+        private string _filename;
+        private bool _messagesHaveChanged;
+
+        public MqttServerStorage(SettingsModel settings, ILogger<MqttServerStorage> 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<MqttApplicationMessage> messages)
         {
+            lock (_messages)
+            {
+                _messages.Clear();
+                _messages.AddRange(messages);
+
+                _messagesHaveChanged = true;
+            }
+
             return Task.CompletedTask;
         }
 
-        public Task<IList<MqttApplicationMessage>> LoadRetainedMessagesAsync()
+        private async Task SaveRetainedMessagesInternalAsync()
         {
-            return Task.FromResult<IList<MqttApplicationMessage>>(null);
+            while (true)
+            {
+                try
+                {
+                    await Task.Delay(TimeSpan.FromSeconds(_settings.RetainedApplicationMessages.WriteInterval)).ConfigureAwait(false);
+
+                    List<MqttApplicationMessage> messages;
+                    lock (_messages)
+                    {
+                        if (!_messagesHaveChanged)
+                        {
+                            continue;
+                        }
+
+                        messages = new List<MqttApplicationMessage>(_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<IList<MqttApplicationMessage>> 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<List<MqttApplicationMessage>>(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;
         }
     }
 }

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}]

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<MqttNetLoggerWrapper>();
             services.AddSingleton<CustomMqttFactory>();
             services.AddSingleton<MqttServerService>();
+            services.AddSingleton<MqttServerStorage>();
 
             services.AddSingleton<MqttClientConnectedHandler>();
             services.AddSingleton<MqttClientDisconnectedHandler>();

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. 
         }