Add support for retained application messages

MQTTnet.Core/Server/IMqttServer.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 
 namespace MQTTnet.Core.Server
@@ -12,7 +13,8 @@ namespace MQTTnet.Core.Server
         IList<ConnectedMqttClient> GetConnectedClients();
         void InjectClient(string identifier, IMqttCommunicationAdapter adapter);
         void Publish(MqttApplicationMessage applicationMessage);
-        void Start();
-        void Stop();
+
+        Task StartAsync();
+        Task StopAsync();
     }
 }

MQTTnet.Core/Server/IMqttServerStorage.cs

@@ -0,0 +1,12 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Core.Server
+{
+    public interface IMqttServerStorage
+    {
+        Task SaveRetainedMessagesAsync(IList<MqttApplicationMessage> messages);
+
+        Task<IList<MqttApplicationMessage>> LoadRetainedMessagesAsync();
+    }
+}

MQTTnet.Core/Server/MqttClientMessageQueue.cs

@@ -1,89 +0,0 @@
-using System;
-using System.Collections.Concurrent;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Core.Adapter;
-using MQTTnet.Core.Diagnostics;
-using MQTTnet.Core.Exceptions;
-using MQTTnet.Core.Packets;
-using System.Linq;
-
-namespace MQTTnet.Core.Server
-{
-    public sealed class MqttClientMessageQueue
-    {
-        private readonly BlockingCollection<MqttPublishPacket> _pendingPublishPackets = new BlockingCollection<MqttPublishPacket>();
-
-        private readonly MqttServerOptions _options;
-        private CancellationTokenSource _cancellationTokenSource;
-
-        public MqttClientMessageQueue(MqttServerOptions options)
-        {
-            _options = options ?? throw new ArgumentNullException(nameof(options));
-        }
-
-        public void Start(IMqttCommunicationAdapter adapter)
-        {
-            if (_cancellationTokenSource != null)
-            {
-                throw new InvalidOperationException($"{nameof(MqttClientMessageQueue)} already started.");
-            }
-
-            if (adapter == null) throw new ArgumentNullException(nameof(adapter));
-            _cancellationTokenSource = new CancellationTokenSource();
-
-            Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token, adapter), _cancellationTokenSource.Token);
-        }
-
-        public void Stop()
-        {
-            _cancellationTokenSource?.Cancel();
-            _cancellationTokenSource = null;
-            _pendingPublishPackets?.Dispose();
-        }
-
-        public void Enqueue(MqttPublishPacket publishPacket)
-        {
-            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
-
-            _pendingPublishPackets.Add(publishPacket);
-        }
-
-        private async Task SendPendingPublishPacketsAsync(CancellationToken cancellationToken, IMqttCommunicationAdapter adapter)
-        {
-            var consumable = _pendingPublishPackets.GetConsumingEnumerable();
-            while (!cancellationToken.IsCancellationRequested)
-            {
-                if (_pendingPublishPackets.Count == 0)
-                {
-                    await Task.Delay(TimeSpan.FromMilliseconds(5), cancellationToken).ConfigureAwait(false);
-                    continue;
-                }
-
-                var packets = consumable.Take(_pendingPublishPackets.Count).ToList();
-                try
-                {
-                    await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packets).ConfigureAwait(false);
-                }
-                catch (MqttCommunicationException exception)
-                {
-                    MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
-                    foreach (var publishPacket in packets)
-                    {
-                        publishPacket.Dup = true;
-                        _pendingPublishPackets.Add(publishPacket, cancellationToken);
-                    }
-                }
-                catch (Exception exception)
-                {
-                    MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
-                    foreach (var publishPacket in packets)
-                    {
-                        publishPacket.Dup = true;
-                        _pendingPublishPackets.Add(publishPacket, cancellationToken);
-                    }
-                }
-            }
-        }
-    }
-}

MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs

@@ -0,0 +1,90 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Core.Adapter;
+using MQTTnet.Core.Diagnostics;
+using MQTTnet.Core.Exceptions;
+using MQTTnet.Core.Packets;
+using MQTTnet.Core.Protocol;
+
+namespace MQTTnet.Core.Server
+{
+    public sealed class MqttClientPendingMessagesQueue
+    {
+        private readonly BlockingCollection<MqttPublishPacket> _pendingPublishPackets = new BlockingCollection<MqttPublishPacket>();
+        private readonly MqttClientSession _session;
+        private readonly MqttServerOptions _options;
+
+        public MqttClientPendingMessagesQueue(MqttServerOptions options, MqttClientSession session)
+        {
+            _session = session ?? throw new ArgumentNullException(nameof(session));
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+        }
+
+        public void Start(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
+        {
+            if (adapter == null) throw new ArgumentNullException(nameof(adapter));
+
+            Task.Run(() => SendPendingPublishPacketsAsync(adapter, cancellationToken), cancellationToken);
+        }
+
+        public void Enqueue(MqttPublishPacket publishPacket)
+        {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
+            _pendingPublishPackets.Add(publishPacket);
+        }
+
+        private async Task SendPendingPublishPacketsAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    await SendPendingPublishPacketAsync(adapter, cancellationToken);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClientPendingMessagesQueue), exception, "Unhandled exception while sending pending publish packets.");
+            }
+        }
+
+        private async Task SendPendingPublishPacketAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
+        {
+            var packet = _pendingPublishPackets.Take(cancellationToken);
+
+            try
+            {
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                if (exception is MqttCommunicationTimedOutException)
+                {
+                    MqttTrace.Warning(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed due to timeout.");
+                }
+                else if (exception is MqttCommunicationException)
+                {
+                    MqttTrace.Warning(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed due to communication exception.");
+                }
+                else
+                {
+                    MqttTrace.Error(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed.");
+                }
+
+                if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+                {
+                    packet.Dup = true;
+                    _pendingPublishPackets.Add(packet, cancellationToken);
+                }
+
+                _session.Stop();
+            }
+        }
+    }
+}

MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs

@@ -0,0 +1,104 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using MQTTnet.Core.Diagnostics;
+using MQTTnet.Core.Internal;
+using MQTTnet.Core.Packets;
+
+namespace MQTTnet.Core.Server
+{
+    public sealed class MqttClientRetainedMessagesManager
+    {
+        private readonly Dictionary<string, MqttPublishPacket> _retainedMessages = new Dictionary<string, MqttPublishPacket>();
+        private readonly MqttServerOptions _options;
+
+        public MqttClientRetainedMessagesManager(MqttServerOptions options)
+        {
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+        }
+
+        public async Task LoadMessagesAsync()
+        {
+            try
+            {
+                var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync();
+                lock (_retainedMessages)
+                {
+                    _retainedMessages.Clear();
+                    foreach (var retainedMessage in retainedMessages)
+                    {
+                        _retainedMessages[retainedMessage.Topic] = retainedMessage.ToPublishPacket();
+                    }
+                }
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClientRetainedMessagesManager), exception, "Unhandled exception while loading retained messages.");
+            }
+        }
+
+        public async Task HandleMessageAsync(string clientId, MqttPublishPacket publishPacket)
+        {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
+            List<MqttPublishPacket> allRetainedMessages;
+            lock (_retainedMessages)
+            {
+                if (publishPacket.Payload?.Any() == false)
+                {
+                    _retainedMessages.Remove(publishPacket.Topic);
+                    MqttTrace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' cleared retained message for topic '{1}'.", clientId, publishPacket.Topic);
+                }
+                else
+                {
+                    _retainedMessages[publishPacket.Topic] = publishPacket;
+                    MqttTrace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' updated retained message for topic '{1}'.", clientId, publishPacket.Topic);
+                }
+
+                allRetainedMessages = new List<MqttPublishPacket>(_retainedMessages.Values);
+            }
+
+            try
+            {
+                // ReSharper disable once UseNullPropagation
+                if (_options.Storage != null)
+                {
+                    await _options.Storage.SaveRetainedMessagesAsync(allRetainedMessages.Select(p => p.ToApplicationMessage()).ToList());
+                }
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClientRetainedMessagesManager), exception, "Unhandled exception while saving retained messages.");
+            }
+        }
+
+        public List<MqttPublishPacket> GetMessages(MqttSubscribePacket subscribePacket)
+        {
+            var retainedMessages = new List<MqttPublishPacket>();
+            lock (_retainedMessages)
+            {
+                foreach (var retainedMessage in _retainedMessages.Values)
+                {
+                    foreach (var topicFilter in subscribePacket.TopicFilters)
+                    {
+                        if (retainedMessage.QualityOfServiceLevel < topicFilter.QualityOfServiceLevel)
+                        {
+                            continue;
+                        }
+
+                        if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic))
+                        {
+                            continue;
+                        }
+
+                        retainedMessages.Add(retainedMessage);
+                        break;
+                    }
+                }
+            }
+
+            return retainedMessages;
+        }
+    }
+}

MQTTnet.Core/Server/MqttClientSession.cs

@@ -14,23 +14,22 @@ namespace MQTTnet.Core.Server
     public sealed class MqttClientSession : IDisposable
     {
         private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
-
+        
         private readonly MqttClientSubscriptionsManager _subscriptionsManager = new MqttClientSubscriptionsManager();
-        private readonly MqttClientMessageQueue _messageQueue;
-        private readonly Action<MqttClientSession, MqttPublishPacket> _publishPacketReceivedCallback;
+        private readonly MqttClientSessionsManager _mqttClientSessionsManager;
+        private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
         private readonly MqttServerOptions _options;
 
-        private CancellationTokenSource _cancellationTokenSource;
         private string _identifier;
-        private MqttApplicationMessage _willApplicationMessage;
+        private CancellationTokenSource _cancellationTokenSource;
+        private MqttApplicationMessage _willMessage;
 
-        public MqttClientSession(string clientId, MqttServerOptions options, Action<MqttClientSession, MqttPublishPacket> publishPacketReceivedCallback)
+        public MqttClientSession(string clientId, MqttServerOptions options, MqttClientSessionsManager mqttClientSessionsManager)
         {
             ClientId = clientId;
             _options = options ?? throw new ArgumentNullException(nameof(options));
-            _publishPacketReceivedCallback = publishPacketReceivedCallback ?? throw new ArgumentNullException(nameof(publishPacketReceivedCallback));
-
-            _messageQueue = new MqttClientMessageQueue(options);
+            _mqttClientSessionsManager = mqttClientSessionsManager ?? throw new ArgumentNullException(nameof(mqttClientSessionsManager));
+            _pendingMessagesQueue = new MqttClientPendingMessagesQueue(options, this);
         }
 
         public string ClientId { get; }
@@ -39,11 +38,11 @@ namespace MQTTnet.Core.Server
 
         public IMqttCommunicationAdapter Adapter { get; private set; }
 
-        public async Task RunAsync(string identifier, MqttApplicationMessage willApplicationMessage, IMqttCommunicationAdapter adapter)
+        public async Task RunAsync(string identifier, MqttApplicationMessage willMessage, IMqttCommunicationAdapter adapter)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
 
-            _willApplicationMessage = willApplicationMessage;
+            _willMessage = willMessage;
 
             try
             {
@@ -51,33 +50,36 @@ namespace MQTTnet.Core.Server
                 Adapter = adapter;
                 _cancellationTokenSource = new CancellationTokenSource();
 
-                _messageQueue.Start(adapter);
-                while (!_cancellationTokenSource.IsCancellationRequested)
-                {
-                    var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false);
-                    await HandleIncomingPacketAsync(packet).ConfigureAwait(false);
-                }
+                _pendingMessagesQueue.Start(adapter, _cancellationTokenSource.Token);
+                await ReceivePacketsAsync(adapter, _cancellationTokenSource.Token);
             }
-            catch (MqttCommunicationException)
+            catch (OperationCanceledException)
             {
             }
+            catch (MqttCommunicationException exception)
+            {
+                MqttTrace.Warning(nameof(MqttClientSession), exception, "Client '{0}': Communication exception while processing client packets.", _identifier);
+            }
             catch (Exception exception)
             {
                 MqttTrace.Error(nameof(MqttClientSession), exception, "Client '{0}': Unhandled exception while processing client packets.", _identifier);
             }
-            finally
+        }
+
+        public void Stop()
+        {
+            if (_willMessage != null)
             {
-                if (willApplicationMessage != null)
-                {
-                    _publishPacketReceivedCallback(this, _willApplicationMessage.ToPublishPacket());
-                }
-
-                _messageQueue.Stop();
-                _cancellationTokenSource.Cancel();
-                Adapter = null;
-
-                MqttTrace.Information(nameof(MqttClientSession), "Client '{0}': Disconnected.", _identifier);
+                _mqttClientSessionsManager.DispatchPublishPacket(this, _willMessage.ToPublishPacket());
             }
+
+            _cancellationTokenSource?.Cancel(false);
+            _cancellationTokenSource?.Dispose();
+            _cancellationTokenSource = null;
+
+            Adapter = null;
+
+            MqttTrace.Information(nameof(MqttClientSession), "Client '{0}': Disconnected.", _identifier);
         }
 
         public void EnqueuePublishPacket(MqttPublishPacket publishPacket)
@@ -89,7 +91,7 @@ namespace MQTTnet.Core.Server
                 return;
             }
 
-            _messageQueue.Enqueue(publishPacket);
+            _pendingMessagesQueue.Enqueue(publishPacket);
             MqttTrace.Verbose(nameof(MqttClientSession), "Client '{0}': Enqueued pending publish packet.", _identifier);
         }
 
@@ -99,68 +101,100 @@ namespace MQTTnet.Core.Server
             _cancellationTokenSource?.Dispose();
         }
 
-        private Task HandleIncomingPacketAsync(MqttBasePacket packet)
+        private async Task ReceivePacketsAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
+                    await ProcessReceivedPacketAsync(packet).ConfigureAwait(false);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            catch (MqttCommunicationException exception)
+            {
+                MqttTrace.Warning(nameof(MqttClientSession), exception, "Client '{0}': Communication exception while processing client packets.", _identifier);
+                Stop();
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClientSession), exception, "Client '{0}': Unhandled exception while processing client packets.", _identifier);
+                Stop();
+            }
+        }
+
+        private async Task ProcessReceivedPacketAsync(MqttBasePacket packet)
         {
             if (packet is MqttSubscribePacket subscribePacket)
             {
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Subscribe(subscribePacket));
+                await Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Subscribe(subscribePacket));
+                EnqueueRetainedMessages(subscribePacket);
             }
-
-            if (packet is MqttUnsubscribePacket unsubscribePacket)
+            else if (packet is MqttUnsubscribePacket unsubscribePacket)
             {
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Unsubscribe(unsubscribePacket));
+                await Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Unsubscribe(unsubscribePacket));
             }
-
-            if (packet is MqttPublishPacket publishPacket)
+            else if (packet is MqttPublishPacket publishPacket)
             {
-                return HandleIncomingPublishPacketAsync(publishPacket);
+                await HandleIncomingPublishPacketAsync(publishPacket);
             }
-
-            if (packet is MqttPubRelPacket pubRelPacket)
+            else if (packet is MqttPubRelPacket pubRelPacket)
             {
-                return HandleIncomingPubRelPacketAsync(pubRelPacket);
+                await HandleIncomingPubRelPacketAsync(pubRelPacket);
             }
-
-            if (packet is MqttPubRecPacket pubRecPacket)
+            else if (packet is MqttPubRecPacket pubRecPacket)
             {
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, pubRecPacket.CreateResponse<MqttPubRelPacket>());
+                await Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, pubRecPacket.CreateResponse<MqttPubRelPacket>());
             }
-
-            if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
+            else if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
             {
                 // Discard message.
-                return Task.FromResult((object)null);
             }
-
-            if (packet is MqttPingReqPacket)
+            else if (packet is MqttPingReqPacket)
             {
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPingRespPacket());
+                await Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPingRespPacket());
             }
-
-            if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
+            else if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
             {
                 _cancellationTokenSource.Cancel();
-                return Task.FromResult((object)null);
             }
-
-            MqttTrace.Warning(nameof(MqttClientSession), "Client '{0}': Received not supported packet ({1}). Closing connection.", _identifier, packet);
-            _cancellationTokenSource.Cancel();
-
-            return Task.FromResult((object)null);
+            else
+            {
+                MqttTrace.Warning(nameof(MqttClientSession), "Client '{0}': Received not supported packet ({1}). Closing connection.", _identifier, packet);
+                _cancellationTokenSource.Cancel();
+            }
         }
 
-        private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
+        private void EnqueueRetainedMessages(MqttSubscribePacket subscribePacket)
         {
+            var retainedMessages = _mqttClientSessionsManager.RetainedMessagesManager.GetMessages(subscribePacket);
+            foreach (var publishPacket in retainedMessages)
+            {
+                EnqueuePublishPacket(publishPacket);
+            }
+        }
+
+        private async Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
+        {
+            if (publishPacket.Retain)
+            {
+                await _mqttClientSessionsManager.RetainedMessagesManager.HandleMessageAsync(_identifier, publishPacket);
+            }
+
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
-                _publishPacketReceivedCallback(this, publishPacket);
-                return Task.FromResult(0);
+                _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket);
+                return;
             }
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
-                _publishPacketReceivedCallback(this, publishPacket);
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
+                _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket);
+                await Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
+                return;
             }
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -171,9 +205,10 @@ namespace MQTTnet.Core.Server
                     _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
                 }
 
-                _publishPacketReceivedCallback(this, publishPacket);
+                _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket);
 
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
+                await Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
+                return;
             }
 
             throw new MqttCommunicationException("Received a not supported QoS level.");

MQTTnet.Core/Server/MqttClientSessionsManager.cs

@@ -14,19 +14,24 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSessionsManager
     {
-        private readonly object _syncRoot = new object();
         private readonly Dictionary<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>();
         private readonly MqttServerOptions _options;
 
         public MqttClientSessionsManager(MqttServerOptions options)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            RetainedMessagesManager = new MqttClientRetainedMessagesManager(options);
         }
 
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
+        public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
+
+        public MqttClientRetainedMessagesManager RetainedMessagesManager { get; }
+
         public async Task RunClientSessionAsync(MqttClientConnectedEventArgs eventArgs)
         {
+            var clientId = string.Empty;
             try
             {
                 if (!(await eventArgs.ClientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false) is MqttConnectPacket connectPacket))
@@ -34,9 +39,11 @@ namespace MQTTnet.Core.Server
                     throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
                 }
 
+                clientId = connectPacket.ClientId;
+
                 // Switch to the required protocol version before sending any response.
                 eventArgs.ClientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion;
-
+                
                 var connectReturnCode = ValidateConnection(connectPacket);
                 if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
                 {
@@ -65,12 +72,13 @@ namespace MQTTnet.Core.Server
             finally
             {
                 await eventArgs.ClientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
+                ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientId, eventArgs.ClientAdapter));
             }
         }
 
         public void Clear()
         {
-            lock (_syncRoot)
+            lock (_clientSessions)
             {
                 _clientSessions.Clear();
             }
@@ -78,7 +86,7 @@ namespace MQTTnet.Core.Server
 
         public IList<ConnectedMqttClient> GetConnectedClients()
         {
-            lock (_syncRoot)
+            lock (_clientSessions)
             {
                 return _clientSessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
                 {
@@ -88,6 +96,27 @@ namespace MQTTnet.Core.Server
             }
         }
 
+        public void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        {
+            try
+            {
+                var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, publishPacket.ToApplicationMessage());
+                ApplicationMessageReceived?.Invoke(this, eventArgs);
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClientSessionsManager), exception, "Error while processing application message");
+            }
+
+            lock (_clientSessions)
+            {
+                foreach (var clientSession in _clientSessions.Values.ToList())
+                {
+                    clientSession.EnqueuePublishPacket(publishPacket);
+                }
+            }
+        }
+
         private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
         {
             if (_options.ConnectionValidator != null)
@@ -100,10 +129,9 @@ namespace MQTTnet.Core.Server
 
         private GetOrCreateClientSessionResult GetOrCreateClientSession(MqttConnectPacket connectPacket)
         {
-            lock (_syncRoot)
+            lock (_clientSessions)
             {
-                MqttClientSession clientSession;
-                var isSessionPresent = _clientSessions.TryGetValue(connectPacket.ClientId, out clientSession);
+                var isSessionPresent = _clientSessions.TryGetValue(connectPacket.ClientId, out var clientSession);
                 if (isSessionPresent)
                 {
                     if (connectPacket.CleanSession)
@@ -124,7 +152,7 @@ namespace MQTTnet.Core.Server
                 {
                     isExistingSession = false;
 
-                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, DispatchPublishPacket);
+                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, this);
                     _clientSessions[connectPacket.ClientId] = clientSession;
 
                     MqttTrace.Verbose(nameof(MqttClientSessionsManager), "Created a new session for client '{0}'.", connectPacket.ClientId);
@@ -133,26 +161,5 @@ namespace MQTTnet.Core.Server
                 return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
             }
         }
-
-        public void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
-        {
-            try
-            {
-                var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, publishPacket.ToApplicationMessage());
-                ApplicationMessageReceived?.Invoke(this, eventArgs);
-            }
-            catch (Exception exception)
-            {
-                MqttTrace.Error(nameof(MqttClientSessionsManager), exception, "Error while processing application message");
-            }
-
-            lock (_syncRoot)
-            {
-                foreach (var clientSession in _clientSessions.Values.ToList())
-                {
-                    clientSession.EnqueuePublishPacket(publishPacket);
-                }
-            }
-        }
     }
 }

MQTTnet.Core/Server/MqttServer.cs

@@ -23,6 +23,7 @@ namespace MQTTnet.Core.Server
 
             _clientSessionsManager = new MqttClientSessionsManager(options);
             _clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
+            _clientSessionsManager.ClientDisconnected += OnClientDisconnected;
         }
 
         public IList<ConnectedMqttClient> GetConnectedClients()
@@ -31,7 +32,7 @@ namespace MQTTnet.Core.Server
         }
 
         public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
-
+        public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
         public void Publish(MqttApplicationMessage applicationMessage)
@@ -44,28 +45,29 @@ namespace MQTTnet.Core.Server
         public void InjectClient(string identifier, IMqttCommunicationAdapter adapter)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
-
             if (_cancellationTokenSource == null) throw new InvalidOperationException("The MQTT server is not started.");
 
             OnClientConnected(this, new MqttClientConnectedEventArgs(identifier, adapter));
         }
 
-        public void Start()
+        public async Task StartAsync()
         {
             if (_cancellationTokenSource != null) throw new InvalidOperationException("The MQTT server is already started.");
 
             _cancellationTokenSource = new CancellationTokenSource();
 
+            await _clientSessionsManager.RetainedMessagesManager.LoadMessagesAsync();
+
             foreach (var adapter in _adapters)
             {
                 adapter.ClientConnected += OnClientConnected;
-                adapter.Start(_options);
+                await adapter.StartAsync(_options);
             }
 
             MqttTrace.Information(nameof(MqttServer), "Started.");
         }
 
-        public void Stop()
+        public async Task StopAsync()
         {
             _cancellationTokenSource?.Cancel(false);
             _cancellationTokenSource?.Dispose();
@@ -74,7 +76,7 @@ namespace MQTTnet.Core.Server
             foreach (var adapter in _adapters)
             {
                 adapter.ClientConnected -= OnClientConnected;
-                adapter.Stop();
+                await adapter.StopAsync();
             }
 
             _clientSessionsManager.Clear();
@@ -89,5 +91,11 @@ namespace MQTTnet.Core.Server
 
             Task.Run(() => _clientSessionsManager.RunClientSessionAsync(eventArgs), _cancellationTokenSource.Token);
         }
+
+        private void OnClientDisconnected(object sender, MqttClientDisconnectedEventArgs eventArgs)
+        {
+            MqttTrace.Information(nameof(MqttServer), "Client '{0}': Disconnected.", eventArgs.Identifier);
+            ClientDisconnected?.Invoke(this, eventArgs);
+        }
     }
 }

MQTTnet.Core/Server/MqttServerOptions.cs

@@ -15,5 +15,7 @@ namespace MQTTnet.Core.Server
         public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);
 
         public Func<MqttConnectPacket, MqttConnectReturnCode> ConnectionValidator { get; set; }
+
+        public IMqttServerStorage Storage { get; set; }
     }
 }