Add samples.

Samples/Client/Client_Connection_Samples.cs

@@ -10,7 +10,7 @@ using System.Security.Authentication;
 using MQTTnet.Client;
 using MQTTnet.Extensions.WebSocket4Net;
 using MQTTnet.Formatter;
-using MQTTnet.Samples.Helpers;
+using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Client;
 

Samples/Client/Client_Subscribe_Samples.cs

@@ -7,7 +7,7 @@
 // ReSharper disable InconsistentNaming
 
 using MQTTnet.Client;
-using MQTTnet.Samples.Helpers;
+using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Client;
 

Samples/Diagnostics/Logger_Samples.cs

@@ -8,6 +8,7 @@
 
 using System.Text;
 using MQTTnet.Diagnostics;
+using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Diagnostics;
 
@@ -22,9 +23,7 @@ public static class Logger_Samples
 
         var mqttFactory = new MqttFactory(new MyLogger());
 
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
-            .WithTcpServer("broker.hivemq.com")
-            .Build();
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
@@ -32,13 +31,31 @@ public static class Logger_Samples
 
             Console.WriteLine("MQTT client is connected.");
 
-            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder()
-                .Build();
+            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
 
             await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
         }
     }
 
+    public static async Task Log_To_Console()
+    {
+        /*
+       * This sample covers the creation of a custom logger which is used to log to the console.
+       */
+
+        var mqttFactory = new MqttFactory(new ConsoleLogger());
+
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+            Console.WriteLine("MQTT client is connected. Press any key to exit.");
+            Console.ReadKey(true);
+        }
+    }
+
     public static async Task Use_Event_Logger()
     {
         /*
@@ -64,9 +81,7 @@ public static class Logger_Samples
 
         var mqttFactory = new MqttFactory(mqttEventLogger);
 
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
-            .WithTcpServer("broker.hivemq.com")
-            .Build();
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
@@ -74,8 +89,7 @@ public static class Logger_Samples
 
             Console.WriteLine("MQTT client is connected.");
 
-            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder()
-                .Build();
+            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
 
             await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
         }
@@ -83,7 +97,7 @@ public static class Logger_Samples
 
     sealed class MyLogger : IMqttNetLogger
     {
-        public bool IsEnabled { get; set; } = true;
+        public bool IsEnabled { get; } = true;
 
         public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
         {

Samples/Program.cs

@@ -8,6 +8,7 @@ Console.WriteLine("Welcome to MQTTnet samples!");
 Console.WriteLine();
 
 var sampleClasses = Assembly.GetExecutingAssembly().GetExportedTypes().OrderBy(c => c.Name).ToList();
+sampleClasses.RemoveAll(c => c.Namespace.Contains(".Shared"));
 
 var index = 0;
 foreach (var sampleClass in sampleClasses)

Samples/Server/Server_Retained_Messages_Samples.cs

@@ -7,7 +7,6 @@
 // ReSharper disable InconsistentNaming
 
 using System.Text.Json;
-using MQTTnet.Server;
 
 namespace MQTTnet.Samples.Server;
 
@@ -33,7 +32,7 @@ public static class Server_Retained_Messages_Samples
             {
                 try
                 {
-                    eventArgs.LoadedRetainedMessages = await JsonSerializer.DeserializeAsync<List<MqttRetainedMessage>>(File.OpenRead(storePath));
+                    eventArgs.LoadedRetainedMessages = await JsonSerializer.DeserializeAsync<List<MqttApplicationMessage>>(File.OpenRead(storePath));
                     Console.WriteLine("Retained messages loaded.");
                 }
                 catch (FileNotFoundException)

Samples/Server/Server_Simple_Samples.cs

@@ -6,8 +6,8 @@
 // ReSharper disable UnusedMember.Global
 // ReSharper disable InconsistentNaming
 
-using MQTTnet.Diagnostics;
 using MQTTnet.Protocol;
+using MQTTnet.Samples.Shared;
 using MQTTnet.Server;
 
 namespace MQTTnet.Samples.Server;
@@ -170,50 +170,4 @@ public static class Server_Simple_Samples
         await server.StartAsync();
         return server;
     }
-
-    class ConsoleLogger : IMqttNetLogger
-    {
-        readonly object _consoleSyncRoot = new();
-
-        public bool IsEnabled => true;
-
-        public void Publish(MqttNetLogLevel logLevel, string source, string message, object[]? parameters, Exception? exception)
-        {
-            var foregroundColor = ConsoleColor.White;
-            switch (logLevel)
-            {
-                case MqttNetLogLevel.Verbose:
-                    foregroundColor = ConsoleColor.White;
-                    break;
-
-                case MqttNetLogLevel.Info:
-                    foregroundColor = ConsoleColor.Green;
-                    break;
-
-                case MqttNetLogLevel.Warning:
-                    foregroundColor = ConsoleColor.DarkYellow;
-                    break;
-
-                case MqttNetLogLevel.Error:
-                    foregroundColor = ConsoleColor.Red;
-                    break;
-            }
-
-            if (parameters?.Length > 0)
-            {
-                message = string.Format(message, parameters);
-            }
-
-            lock (_consoleSyncRoot)
-            {
-                Console.ForegroundColor = foregroundColor;
-                Console.WriteLine(message);
-
-                if (exception != null)
-                {
-                    Console.WriteLine(exception);
-                }
-            }
-        }
-    }
 }

Samples/Shared/ConsoleLogger.cs

@@ -0,0 +1,53 @@
+// 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 MQTTnet.Diagnostics;
+
+namespace MQTTnet.Samples.Shared;
+
+public sealed class ConsoleLogger : IMqttNetLogger
+{
+    readonly object _consoleSyncRoot = new();
+
+    public bool IsEnabled => true;
+
+    public void Publish(MqttNetLogLevel logLevel, string source, string message, object[]? parameters, Exception? exception)
+    {
+        var foregroundColor = ConsoleColor.White;
+        switch (logLevel)
+        {
+            case MqttNetLogLevel.Verbose:
+                foregroundColor = ConsoleColor.White;
+                break;
+
+            case MqttNetLogLevel.Info:
+                foregroundColor = ConsoleColor.Green;
+                break;
+
+            case MqttNetLogLevel.Warning:
+                foregroundColor = ConsoleColor.DarkYellow;
+                break;
+
+            case MqttNetLogLevel.Error:
+                foregroundColor = ConsoleColor.Red;
+                break;
+        }
+
+        if (parameters?.Length > 0)
+        {
+            message = string.Format(message, parameters);
+        }
+
+        lock (_consoleSyncRoot)
+        {
+            Console.ForegroundColor = foregroundColor;
+            Console.WriteLine(message);
+
+            if (exception != null)
+            {
+                Console.WriteLine(exception);
+            }
+        }
+    }
+}

Samples/Shared/ObjectExtensions.cs

@@ -4,7 +4,7 @@
 
 using System.Text.Json;
 
-namespace MQTTnet.Samples.Helpers;
+namespace MQTTnet.Samples.Shared;
 
 internal static class ObjectExtensions
 {

Source/MQTTnet.TestApp/ServerTest.cs

@@ -4,10 +4,12 @@
 
 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;
@@ -78,15 +80,15 @@ namespace MQTTnet.TestApp
                 
                 mqttServer.LoadingRetainedMessageAsync += e =>
                 {
-                    List<MqttRetainedMessage> retainedMessages;
+                    List<MqttApplicationMessage> retainedMessages;
                     if (File.Exists(Filename))
                     {
                         var json = File.ReadAllText(Filename);
-                        retainedMessages = JsonConvert.DeserializeObject<List<MqttRetainedMessage>>(json);
+                        retainedMessages = JsonConvert.DeserializeObject<List<MqttApplicationMessage>>(json);
                     }
                     else
                     {
-                        retainedMessages = new List<MqttRetainedMessage>();
+                        retainedMessages = new List<MqttApplicationMessage>();
                     }
 
                     e.LoadedRetainedMessages = retainedMessages;

Source/MQTTnet.Tests/Server/General.cs

@@ -394,7 +394,7 @@ namespace MQTTnet.Tests.Server
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
-                ICollection<MqttRetainedMessage> savedRetainedMessages = null;
+                List<MqttApplicationMessage> savedRetainedMessages = null;
 
                 var s = await testEnvironment.StartServer();
                 s.RetainedMessageChangedAsync += e =>

Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs

@@ -9,6 +9,6 @@ namespace MQTTnet.Server
 {
     public sealed class LoadingRetainedMessagesEventArgs : EventArgs
     {
-        public ICollection<MqttRetainedMessage> LoadedRetainedMessages { get; set; } = new List<MqttRetainedMessage>();
+        public List<MqttApplicationMessage> LoadedRetainedMessages { get; set; } = new List<MqttApplicationMessage>();
     }
 }

Source/MQTTnet/Server/Events/RetainedMessageChangeType.cs

@@ -1,15 +0,0 @@
-// 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
-    }
-}

Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs

@@ -9,24 +9,17 @@ namespace MQTTnet.Server
 {
     public sealed class RetainedMessageChangedEventArgs : EventArgs
     {
-        public RetainedMessageChangedEventArgs(
-            string clientId,
-            MqttRetainedMessage changedMessage,
-            RetainedMessageChangeType changeType,
-            List<MqttRetainedMessage> allMessages)
+        public RetainedMessageChangedEventArgs(string clientId, MqttApplicationMessage changedRetainedMessage, List<MqttApplicationMessage> storedRetainedMessages)
         {
             ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
-            ChangedRetainedMessage = changedMessage ?? throw new ArgumentNullException(nameof(changedMessage));
-            ChangeType = changeType;
-            StoredRetainedMessages = allMessages ?? throw new ArgumentNullException(nameof(allMessages));
+            ChangedRetainedMessage = changedRetainedMessage ?? throw new ArgumentNullException(nameof(changedRetainedMessage));
+            StoredRetainedMessages = storedRetainedMessages ?? throw new ArgumentNullException(nameof(storedRetainedMessages));
         }
 
-        public RetainedMessageChangeType ChangeType { get; }
-
-        public MqttRetainedMessage ChangedRetainedMessage { get; }
+        public MqttApplicationMessage ChangedRetainedMessage { get; }
 
         public string ClientId { get; }
 
-        public List<MqttRetainedMessage> StoredRetainedMessages { get; }
+        public List<MqttApplicationMessage> StoredRetainedMessages { get; }
     }
 }

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -58,7 +58,7 @@ namespace MQTTnet.Server
             _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
         }
 
-        public async Task Stop()
+        public async Task CloseAllConnectionsAsync()
         {
             List<MqttClient> connections;
             lock (_clients)
@@ -131,7 +131,7 @@ namespace MQTTnet.Server
             {
                 if (applicationMessage.Retain)
                 {
-                    await _retainedMessagesManager.HandleMessage(senderId, applicationMessage).ConfigureAwait(false);
+                    await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
                 }
 
                 var deliveryCount = 0;

Source/MQTTnet/Server/Internal/MqttRetainedMessage.cs

@@ -1,18 +0,0 @@
-// 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; }
-
-        /// <summary>
-        ///     Gets the timestamp in UTC when this retained message eas published.
-        /// </summary>
-        public DateTime PublishedTimestamp { get; set; }
-    }
-}

Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs

@@ -3,7 +3,6 @@
 // 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;
@@ -16,42 +15,47 @@ namespace MQTTnet.Server
 {
     public sealed class MqttRetainedMessagesManager
     {
+        readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>(4096);
+        readonly AsyncLock _storageAccessLock = new AsyncLock();
+
         readonly MqttServerEventContainer _eventContainer;
         readonly MqttNetSourceLogger _logger;
-        readonly ConcurrentDictionary<string, MqttRetainedMessage> _messages = new ConcurrentDictionary<string, MqttRetainedMessage>();
-        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 ClearMessages()
+        
+        public async Task Start()
         {
-            _messages.Clear();
-
-            using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
+            try
             {
-                await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
+                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)
+            {
+                _logger.Error(exception, "Unhandled exception while loading retained messages.");
             }
         }
 
-        public Task<IList<MqttApplicationMessage>> GetMessages()
-        {
-            var result = new List<MqttApplicationMessage>(_messages.Select(m => m.Value.ApplicationMessage));
-            return Task.FromResult((IList<MqttApplicationMessage>)result);
-        }
-
-        public async Task HandleMessage(string clientId, MqttApplicationMessage applicationMessage)
+        public async Task UpdateMessage(string clientId, MqttApplicationMessage applicationMessage)
         {
             if (applicationMessage == null)
             {
@@ -60,185 +64,78 @@ namespace MQTTnet.Server
 
             try
             {
-                List<MqttRetainedMessage> allMessages = null;
+                List<MqttApplicationMessage> messagesForSave = null;
                 var saveIsRequired = false;
-
-                var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
-                var changeType = RetainedMessageChangeType.Created;
-
-                var retainedMessage = new MqttRetainedMessage
+                
+                lock (_messages)
                 {
-                    ApplicationMessage = applicationMessage,
-                    PublishedTimestamp = DateTime.UtcNow
-                };
+                    var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
 
-                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,
-                        _ =>
-                        {
-                            changeType = RetainedMessageChangeType.Created;
-                            changedMessage = retainedMessage;
-                            saveIsRequired = true;
-
-                            return retainedMessage;
-                        },
-                        (_, existingMessage) =>
-                        {
-                            var hasSamePayload = existingMessage.ApplicationMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray);
-                            var hasSameQoSLevel = existingMessage.ApplicationMessage.QualityOfServiceLevel == applicationMessage.QualityOfServiceLevel;
-
-                            if (!hasSameQoSLevel || !hasSamePayload)
-                            {
-                                changeType = RetainedMessageChangeType.Updated;
-                                saveIsRequired = true;
-                                messageChanged = true;
-                                return retainedMessage;
-                            }
-
-                            // Keep the existing message.
-                            return existingMessage;
-                        });
-
-                    if (messageChanged)
+                    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))
+                        {
+                            _messages[applicationMessage.Topic] = applicationMessage;
+                            saveIsRequired = true;
+                        }
+                        else
+                        {
+                            if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray))
+                            {
+                                _messages[applicationMessage.Topic] = applicationMessage;
+                                saveIsRequired = true;
+                            }
+                        }
+
                         _logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
                     }
+
+                    if (saveIsRequired)
+                    {
+                        messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
+                    }
                 }
 
                 if (saveIsRequired)
                 {
-                    allMessages = new List<MqttRetainedMessage>(_messages.Values);
-                }
-
-                if (saveIsRequired)
-                {
-                    await SaveMessages(clientId, retainedMessage, changeType, allMessages).ConfigureAwait(false);
+                    using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
+                    {
+                        var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessage, messagesForSave);
+                        await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+                    }
                 }
             }
             catch (Exception exception)
             {
-                _logger.Error(exception, "Error while handling retained messages.");
+                _logger.Error(exception, "Unhandled exception while handling retained messages.");
             }
         }
-
-        public async Task Start()
+        
+        public Task<IList<MqttApplicationMessage>> GetMessages()
         {
-            _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)
+            lock (_messages)
             {
-                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;
-                }
+                var result = new List<MqttApplicationMessage>(_messages.Values);
+                return Task.FromResult((IList<MqttApplicationMessage>)result);
             }
         }
 
-        async Task RemoveExpiredMessages()
+        public async Task ClearMessages()
         {
-            var expiredMessages = new List<MqttRetainedMessage>();
-
-            var now = DateTime.UtcNow;
-
-            foreach (var retainedMessage in _messages)
+            lock (_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);
+                _messages.Clear();
             }
 
-            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<MqttRetainedMessage>(_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<MqttRetainedMessage> allMessages)
-        {
             using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
             {
-                if (_eventContainer.RetainedMessageChangedEvent.HasHandlers)
-                {
-                    var eventArgs = new RetainedMessageChangedEventArgs(clientId, changedMessage, changeType, allMessages);
-                    await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
-                }
+                await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
             }
         }
     }
-}
+}

Source/MQTTnet/Server/MqttServer.cs

@@ -287,9 +287,8 @@ namespace MQTTnet.Server
 
                 _cancellationTokenSource.Cancel(false);
 
-                await _clientSessionsManager.Stop().ConfigureAwait(false);
-                await _retainedMessagesManager.Stop().ConfigureAwait(false);
-                
+                await _clientSessionsManager.CloseAllConnectionsAsync().ConfigureAwait(false);
+
                 foreach (var adapter in _adapters)
                 {
                     adapter.ClientHandler = null;
@@ -358,7 +357,7 @@ namespace MQTTnet.Server
             ThrowIfDisposed();
             ThrowIfNotStarted();
 
-            return _retainedMessagesManager?.HandleMessage(null, retainedMessage);
+            return _retainedMessagesManager?.UpdateMessage(null, retainedMessage);
         }
 
         protected override void Dispose(bool disposing)