Added IMqttRetainedMessagesManager

* Added interface IMqttRetainedMessagesManager to allow for different
retained messages manager implementations. The interface copies almost
exactly the current MqttRetainedMessagesManager implementation.

* Added IMqttRetainedMessagesManager.Start() to configure the provided
IMqttRetainedMessagesManager outside the constructor. This method returns
Task because some implementations that use external storage may be slow on
initialization (e.g., connect to a database).

* Modified MqttRetainedMessagesManager to implement new interface.

Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs

@@ -0,0 +1,21 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using MQTTnet.Diagnostics;
+
+namespace MQTTnet.Server
+{
+    public interface IMqttRetainedMessagesManager
+    {
+        Task Start(IMqttServerOptions options, IMqttNetChildLogger logger);
+
+        Task LoadMessagesAsync();
+
+        Task ClearMessagesAsync();
+
+        Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage);
+
+        Task<IList<MqttApplicationMessage>> GetMessagesAsync();
+
+        Task<List<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<TopicFilter> topicFilters);
+    }
+}

Source/MQTTnet/Server/IMqttServerOptions.cs

@@ -24,6 +24,6 @@ namespace MQTTnet.Server
 
         IMqttServerStorage Storage { get; }
 
-
+        IMqttRetainedMessagesManager RetainedMessagesManager { get; }
     }
 }

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -21,7 +21,7 @@ namespace MQTTnet.Server
         private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
         private readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();
 
-        private readonly MqttRetainedMessagesManager _retainedMessagesManager;
+        private readonly IMqttRetainedMessagesManager _retainedMessagesManager;
         private readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
         private readonly MqttClientSessionsManager _sessionsManager;
 
@@ -48,7 +48,7 @@ namespace MQTTnet.Server
             MqttClientSession session,
             IMqttServerOptions serverOptions,
             MqttClientSessionsManager sessionsManager,
-            MqttRetainedMessagesManager retainedMessagesManager,
+            IMqttRetainedMessagesManager retainedMessagesManager,
             IMqttNetChildLogger logger)
         {
             Session = session ?? throw new ArgumentNullException(nameof(session));

Source/MQTTnet/Server/MqttClientSession.cs

@@ -52,7 +52,7 @@ namespace MQTTnet.Server
             ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage);
         }
 
-        public async Task SubscribeAsync(ICollection<TopicFilter> topicFilters, MqttRetainedMessagesManager retainedMessagesManager)
+        public async Task SubscribeAsync(ICollection<TopicFilter> topicFilters, IMqttRetainedMessagesManager retainedMessagesManager)
         {
             await SubscriptionsManager.SubscribeAsync(topicFilters).ConfigureAwait(false);
 

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -25,13 +25,13 @@ namespace MQTTnet.Server
         private readonly CancellationToken _cancellationToken;
         private readonly MqttServerEventDispatcher _eventDispatcher;
 
-        private readonly MqttRetainedMessagesManager _retainedMessagesManager;
+        private readonly IMqttRetainedMessagesManager _retainedMessagesManager;
         private readonly IMqttServerOptions _options;
         private readonly IMqttNetChildLogger _logger;
 
         public MqttClientSessionsManager(
             IMqttServerOptions options,
-            MqttRetainedMessagesManager retainedMessagesManager,
+            IMqttRetainedMessagesManager retainedMessagesManager,
             CancellationToken cancellationToken,
             MqttServerEventDispatcher eventDispatcher,
             IMqttNetChildLogger logger)

Source/MQTTnet/Server/MqttRetainedMessagesManager.cs

@@ -7,20 +7,21 @@ using MQTTnet.Internal;
 
 namespace MQTTnet.Server
 {
-    public class MqttRetainedMessagesManager
+    public class MqttRetainedMessagesManager : IMqttRetainedMessagesManager
     {
         private readonly byte[] _emptyArray = new byte[0];
         private readonly AsyncLock _messagesLock = new AsyncLock();
         private readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
 
-        private readonly IMqttNetChildLogger _logger;
+        private IMqttNetChildLogger _logger;
-        private readonly IMqttServerOptions _options;
+        private IMqttServerOptions _options;
 
-        public MqttRetainedMessagesManager(IMqttServerOptions options, IMqttNetChildLogger logger)
+        public Task Start(IMqttServerOptions options, IMqttNetChildLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateChildLogger(nameof(MqttRetainedMessagesManager));
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            return Task.CompletedTask;
         }
 
         public async Task LoadMessagesAsync()

Source/MQTTnet/Server/MqttServer.cs

@@ -19,7 +19,7 @@ namespace MQTTnet.Server
         private readonly IMqttNetChildLogger _logger;
 
         private MqttClientSessionsManager _clientSessionsManager;
-        private MqttRetainedMessagesManager _retainedMessagesManager;
+        private IMqttRetainedMessagesManager _retainedMessagesManager;
         private CancellationTokenSource _cancellationTokenSource;
 
         public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetChildLogger logger)
@@ -121,7 +121,8 @@ namespace MQTTnet.Server
 
             _cancellationTokenSource = new CancellationTokenSource();
 
-            _retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
+            _retainedMessagesManager = Options.RetainedMessagesManager ?? new MqttRetainedMessagesManager();
+            await _retainedMessagesManager.Start(Options, _logger);
             await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false);
 
             _clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _cancellationTokenSource.Token, _eventDispatcher, _logger);

Source/MQTTnet/Server/MqttServerOptions.cs

@@ -29,5 +29,7 @@ namespace MQTTnet.Server
         public IMqttServerUnsubscriptionInterceptor UnsubscriptionInterceptor { get; set; }
 
         public IMqttServerStorage Storage { get; set; }
+
+        public IMqttRetainedMessagesManager RetainedMessagesManager { get; set; }
     }
 }

Source/MQTTnet/Server/MqttServerOptionsBuilder.cs

@@ -125,6 +125,12 @@ namespace MQTTnet.Server
             return this;
         }
 
+        public MqttServerOptionsBuilder WithRetainedMessagesManager(IMqttRetainedMessagesManager value)
+        {
+            _options.RetainedMessagesManager = value;
+            return this;
+        }
+
         public MqttServerOptionsBuilder WithConnectionValidator(IMqttServerConnectionValidator value)
         {
             _options.ConnectionValidator = value;