Merge from develop.

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs

@@ -10,6 +10,7 @@ namespace MQTTnet.Extensions.ManagedClient
         bool IsStarted { get; }
         bool IsConnected { get; }
         int PendingApplicationMessagesCount { get; }
+        IManagedMqttClientOptions Options { get; }
 
         event EventHandler<MqttClientConnectedEventArgs> Connected;
         event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;

Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj

@@ -1,7 +1,8 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard1.3;netstandard2.0;net452;net461</TargetFrameworks>
+    <TargetFrameworks>netstandard1.3;netstandard2.0</TargetFrameworks>
+    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
     <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core'">$(TargetFrameworks);uap10.0</TargetFrameworks>
     <Product />
     <Company />

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -27,8 +27,7 @@ namespace MQTTnet.Extensions.ManagedClient
         private CancellationTokenSource _publishingCancellationToken;
 
         private ManagedMqttClientStorageManager _storageManager;
-        private IManagedMqttClientOptions _options;
-
+        
         private bool _subscriptionsNotPushed;
 
         public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger)
@@ -47,6 +46,7 @@ namespace MQTTnet.Extensions.ManagedClient
         public bool IsConnected => _mqttClient.IsConnected;
         public bool IsStarted => _connectionCancellationToken != null;
         public int PendingApplicationMessagesCount => _messageQueue.Count;
+        public IManagedMqttClientOptions Options { get; private set; }
 
         public event EventHandler<MqttClientConnectedEventArgs> Connected;
         public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
@@ -70,11 +70,11 @@ namespace MQTTnet.Extensions.ManagedClient
 
             if (_connectionCancellationToken != null) throw new InvalidOperationException("The managed client is already started.");
 
-            _options = options;
+            Options = options;
 
-            if (_options.Storage != null)
+            if (Options.Storage != null)
             {
-                _storageManager = new ManagedMqttClientStorageManager(_options.Storage);
+                _storageManager = new ManagedMqttClientStorageManager(Options.Storage);
                 var messages = await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false);
 
                 foreach (var message in messages)
@@ -116,16 +116,16 @@ namespace MQTTnet.Extensions.ManagedClient
             ManagedMqttApplicationMessage removedMessage = null;
             lock (_messageQueue)
             {
-                if (_messageQueue.Count >= _options.MaxPendingMessages)
+                if (_messageQueue.Count >= Options.MaxPendingMessages)
                 {
-                    if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
+                    if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
                     {
                         _logger.Verbose("Skipping publish of new application message because internal queue is full.");
                         ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(applicationMessage));
                         return;
                     }
 
-                    if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
+                    if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
                     {
                         removedMessage = _messageQueue.RemoveFirst();
                         _logger.Verbose("Removed oldest application message from internal queue because it is full.");
@@ -219,7 +219,7 @@ namespace MQTTnet.Extensions.ManagedClient
                 if (connectionState == ReconnectionResult.NotConnected)
                 {
                     StopPublishing();
-                    await Task.Delay(_options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
+                    await Task.Delay(Options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
                     return;
                 }
 
@@ -232,7 +232,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
                 if (connectionState == ReconnectionResult.StillConnected)
                 {
-                    await Task.Delay(_options.ConnectionCheckInterval, cancellationToken).ConfigureAwait(false);
+                    await Task.Delay(Options.ConnectionCheckInterval, cancellationToken).ConfigureAwait(false);
                 }
             }
             catch (OperationCanceledException)
@@ -254,7 +254,15 @@ namespace MQTTnet.Extensions.ManagedClient
             {
                 while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected)
                 {
-                    var message = _messageQueue.Dequeue();
+                    //Peek at the message without dequeueing in order to prevent the
+                    //possibility of the queue growing beyond the configured cap.
+                    //Previously, messages could be re-enqueued if there was an
+                    //exception, and this re-enqueueing did not honor the cap.
+                    //Furthermore, because re-enqueueing would shuffle the order
+                    //of the messages, the DropOldestQueuedMessage strategy would
+                    //be unable to know which message is actually the oldest and would
+                    //instead drop the first item in the queue.
+                    var message = _messageQueue.PeekAndWait();
                     if (message == null)
                     {
                         continue;
@@ -284,6 +292,16 @@ namespace MQTTnet.Extensions.ManagedClient
             try
             {
                 _mqttClient.PublishAsync(message.ApplicationMessage).GetAwaiter().GetResult();
+                lock (_messageQueue) //lock to avoid conflict with this.PublishAsync
+                {
+                    //While publishing this message, this.PublishAsync could have booted this
+                    //message off the queue to make room for another (when using a cap
+                    //with the DropOldestQueuedMessage strategy).  If the first item
+                    //in the queue is equal to this message, then it's safe to remove
+                    //it from the queue.  If not, that means this.PublishAsync has already
+                    //removed it, in which case we don't want to do anything.
+                    _messageQueue.RemoveFirst(i => i.Id.Equals(message.Id));
+                }
                 _storageManager?.RemoveAsync(message).GetAwaiter().GetResult();
             }
             catch (MqttCommunicationException exception)
@@ -292,9 +310,19 @@ namespace MQTTnet.Extensions.ManagedClient
 
                 _logger.Warning(exception, $"Publishing application ({message.Id}) message failed.");
 
-                if (message.ApplicationMessage.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+                if (message.ApplicationMessage.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
                 {
-                    _messageQueue.Enqueue(message);
+                    //If QoS 0, we don't want this message to stay on the queue.
+                    //If QoS 1 or 2, it's possible that, when using a cap, this message
+                    //has been booted off the queue by this.PublishAsync, in which case this
+                    //thread will not continue to try to publish it. While this does
+                    //contradict the expected behavior of QoS 1 and 2, that's also true
+                    //for the usage of a message queue cap, so it's still consistent
+                    //with prior behavior in that way.
+                    lock (_messageQueue) //lock to avoid conflict with this.PublishAsync
+                    {
+                        _messageQueue.RemoveFirst(i => i.Id.Equals(message.Id));
+                    }
                 }
             }
             catch (Exception exception)
@@ -360,7 +388,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
             try
             {
-                await _mqttClient.ConnectAsync(_options.ClientOptions).ConfigureAwait(false);
+                await _mqttClient.ConnectAsync(Options.ClientOptions).ConfigureAwait(false);
                 return ReconnectionResult.Reconnected;
             }
             catch (Exception exception)

Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj

@@ -1,7 +1,8 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard1.3;netstandard2.0;net452;net461</TargetFrameworks>
+    <TargetFrameworks>netstandard1.3;netstandard2.0</TargetFrameworks>
+    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
     <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core'">$(TargetFrameworks);uap10.0</TargetFrameworks>
     <Product />
     <Company />

Source/MQTTnet/Client/IMqttClient.cs

@@ -7,6 +7,7 @@ namespace MQTTnet.Client
     public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
     {
         bool IsConnected { get; }
+        IMqttClientOptions Options { get; }
 
         event EventHandler<MqttClientConnectedEventArgs> Connected;
         event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;

Source/MQTTnet/Client/MqttClient.cs

@@ -25,7 +25,6 @@ namespace MQTTnet.Client
         private readonly IMqttClientAdapterFactory _adapterFactory;
         private readonly IMqttNetChildLogger _logger;
 
-        private IMqttClientOptions _options;
         private CancellationTokenSource _cancellationTokenSource;
         internal Task _packetReceiverTask;
         internal Task _keepAliveMessageSenderTask;
@@ -46,6 +45,7 @@ namespace MQTTnet.Client
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
         public bool IsConnected { get; private set; }
+        public IMqttClientOptions Options { get; private set; }
 
         public async Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options)
         {
@@ -56,16 +56,17 @@ namespace MQTTnet.Client
 
             try
             {
-                _options = options;
+                Options = options;
+
                 _packetIdentifierProvider.Reset();
                 _packetDispatcher.Reset();
 
                 _cancellationTokenSource = new CancellationTokenSource();
                 _disconnectGate = 0;
                 _adapter = _adapterFactory.CreateClientAdapter(options, _logger);
-                
-                _logger.Verbose($"Trying to connect with server ({_options.ChannelOptions}).");
-                await _adapter.ConnectAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token).ConfigureAwait(false);
+
+                _logger.Verbose($"Trying to connect with server ({Options.ChannelOptions}).");
+                await _adapter.ConnectAsync(Options.CommunicationTimeout, _cancellationTokenSource.Token).ConfigureAwait(false);
                 _logger.Verbose("Connection with server established.");
 
                 StartReceivingPackets(_cancellationTokenSource.Token);
@@ -75,7 +76,7 @@ namespace MQTTnet.Client
 
                 _sendTracker.Restart();
 
-                if (_options.KeepAlivePeriod != TimeSpan.Zero)
+                if (Options.KeepAlivePeriod != TimeSpan.Zero)
                 {
                     StartSendingKeepAliveMessages(_cancellationTokenSource.Token);
                 }
@@ -141,7 +142,7 @@ namespace MQTTnet.Client
             return subscribePacket.TopicFilters.Select((t, i) => new MqttSubscribeResult(t, response.SubscribeReturnCodes[i])).ToList();
         }
 
-        public async Task UnsubscribeAsync(IEnumerable<string> topicFilters)
+        public Task UnsubscribeAsync(IEnumerable<string> topicFilters)
         {
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
@@ -153,7 +154,7 @@ namespace MQTTnet.Client
                 TopicFilters = topicFilters.ToList()
             };
 
-            await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false);
+            return SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, _cancellationTokenSource.Token);
         }
 
         public Task PublishAsync(MqttApplicationMessage applicationMessage)
@@ -201,11 +202,11 @@ namespace MQTTnet.Client
         {
             var connectPacket = new MqttConnectPacket
             {
-                ClientId = _options.ClientId,
-                Username = _options.Credentials?.Username,
-                Password = _options.Credentials?.Password,
-                CleanSession = _options.CleanSession,
-                KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds,
+                ClientId = Options.ClientId,
+                Username = Options.Credentials?.Username,
+                Password = Options.Credentials?.Password,
+                CleanSession = Options.CleanSession,
+                KeepAlivePeriod = (ushort)Options.KeepAlivePeriod.TotalSeconds,
                 WillMessage = willApplicationMessage
             };
 
@@ -241,14 +242,14 @@ namespace MQTTnet.Client
 
             try
             {
-                await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false);
-                await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false);
-
                 if (_adapter != null)
                 {
-                    await _adapter.DisconnectAsync(_options.CommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
+                    await _adapter.DisconnectAsync(Options.CommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
                 }
 
+                await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false);
+                await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false);
+                
                 _logger.Verbose("Disconnected from adapter.");
             }
             catch (Exception adapterException)
@@ -310,7 +311,7 @@ namespace MQTTnet.Client
             try
             {
                 await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
-                var respone = await Internal.TaskExtensions.TimeoutAfterAsync(ct => packetAwaiter.Task, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
+                var respone = await Internal.TaskExtensions.TimeoutAfterAsync(ct => packetAwaiter.Task, Options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
 
                 return (TResponsePacket)respone;
             }
@@ -327,16 +328,16 @@ namespace MQTTnet.Client
 
         private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
         {
-            _logger.Verbose("Start sending keep alive packets.");
-
             try
             {
+                _logger.Verbose("Start sending keep alive packets.");
+
                 while (!cancellationToken.IsCancellationRequested)
                 {
-                    var keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75);
-                    if (_options.KeepAliveSendInterval.HasValue)
+                    var keepAliveSendInterval = TimeSpan.FromSeconds(Options.KeepAlivePeriod.TotalSeconds * 0.75);
+                    if (Options.KeepAliveSendInterval.HasValue)
                     {
-                        keepAliveSendInterval = _options.KeepAliveSendInterval.Value;
+                        keepAliveSendInterval = Options.KeepAliveSendInterval.Value;
                     }
 
                     var waitTime = keepAliveSendInterval - _sendTracker.Elapsed;
@@ -381,10 +382,10 @@ namespace MQTTnet.Client
 
         private async Task ReceivePacketsAsync(CancellationToken cancellationToken)
         {
-            _logger.Verbose("Start receiving packets.");
-
             try
             {
+                _logger.Verbose("Start receiving packets.");
+
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken)
@@ -507,7 +508,7 @@ namespace MQTTnet.Client
                 () => ReceivePacketsAsync(cancellationToken),
                 cancellationToken,
                 TaskCreationOptions.LongRunning, 
-                TaskScheduler.Default);
+                TaskScheduler.Default).Unwrap();
         }
 
         private void StartSendingKeepAliveMessages(CancellationToken cancellationToken)
@@ -516,7 +517,7 @@ namespace MQTTnet.Client
                 () => SendKeepAliveMessagesAsync(cancellationToken),
                 cancellationToken,
                 TaskCreationOptions.LongRunning,
-                TaskScheduler.Default);
+                TaskScheduler.Default).Unwrap();
         }
 
         private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
@@ -525,7 +526,7 @@ namespace MQTTnet.Client
             {
                 // TODO: Move conversion to formatter.
                 var applicationMessage = publishPacket.ToApplicationMessage();
-                ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage));
+                ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage));
             }
             catch (Exception exception)
             {

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -2,7 +2,6 @@
 using System;
 using System.Net.Security;
 using System.Net.Sockets;
-using System.Security.Authentication;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading.Tasks;
 using System.IO;
@@ -17,7 +16,7 @@ namespace MQTTnet.Implementations
     {
         private readonly IMqttClientOptions _clientOptions;
         private readonly MqttClientTcpOptions _options;
-        
+
         private Socket _socket;
         private Stream _stream;
 
@@ -87,8 +86,9 @@ namespace MQTTnet.Implementations
 
         public void Dispose()
         {
-            Cleanup(ref _stream, (s) => s.Dispose());
-            Cleanup(ref _socket, (s) => {
+            Cleanup(ref _stream, s => s.Dispose());
+            Cleanup(ref _socket, s =>
+            {
                 if (s.Connected)
                 {
                     s.Shutdown(SocketShutdown.Both);
@@ -102,7 +102,7 @@ namespace MQTTnet.Implementations
             // Try the instance callback.
             if (_options.TlsOptions.CertificateValidationCallback != null)
             {
-                return _options.TlsOptions.CertificateValidationCallback(x509Certificate, chain, sslPolicyErrors,_clientOptions);
+                return _options.TlsOptions.CertificateValidationCallback(x509Certificate, chain, sslPolicyErrors, _clientOptions);
             }
 
             // Try static callback.

Source/MQTTnet/Internal/BlockingQueue.cs

@@ -55,6 +55,40 @@ namespace MQTTnet.Internal
                 _gate.WaitOne();
             }
         }
+        
+        public TItem PeekAndWait()
+        {
+            while (true)
+            {
+                lock (_syncRoot)
+                {
+                    if (_items.Count > 0)
+                    {
+                        return _items.First.Value;
+                    }
+
+                    if (_items.Count == 0)
+                    {
+                        _gate.Reset();
+                    }
+                }
+
+                _gate.WaitOne();
+            }
+        }
+
+        public void RemoveFirst(Predicate<TItem> match)
+        {
+            if (match == null) throw new ArgumentNullException(nameof(match));
+
+            lock (_syncRoot)
+            {
+                if (_items.Count > 0 && match(_items.First.Value))
+                {
+                    _items.RemoveFirst();
+                }
+            }
+        }
 
         public TItem RemoveFirst()
         {

Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs

@@ -36,8 +36,8 @@ namespace MQTTnet.Server
             {
                 return;
             }
-
-            Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken);
+            
+            Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
         public void Pause()
@@ -50,6 +50,12 @@ namespace MQTTnet.Server
             _isPaused = false;
         }
 
+        public void Reset()
+        {
+            _lastPacketReceivedTracker.Restart();
+            _lastNonKeepAlivePacketReceivedTracker.Restart();
+        }
+
         public void PacketReceived(MqttBasePacket packet)
         {
             _lastPacketReceivedTracker.Restart();

Source/MQTTnet/Server/MqttClientSession.cs

@@ -9,6 +9,7 @@ using MQTTnet.Exceptions;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Serializer;
 
 namespace MQTTnet.Server
 {
@@ -17,6 +18,7 @@ namespace MQTTnet.Server
         private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
+        private readonly MqttServerEventDispatcher _eventDispatcher;
         private readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
         private readonly MqttClientPendingPacketsQueue _pendingPacketsQueue;
         private readonly MqttClientSubscriptionsManager _subscriptionsManager;
@@ -28,29 +30,33 @@ namespace MQTTnet.Server
         private CancellationTokenSource _cancellationTokenSource;
         private MqttApplicationMessage _willMessage;
         private bool _wasCleanDisconnect;
-        private IMqttChannelAdapter _adapter;
         private Task _workerTask;
         private IDisposable _cleanupHandle;
 
+        private string _adapterEndpoint;
+        private MqttProtocolVersion? _adapterProtocolVersion;
+        
         public MqttClientSession(
             string clientId,
             IMqttServerOptions options,
             MqttClientSessionsManager sessionsManager,
             MqttRetainedMessagesManager retainedMessagesManager,
+            MqttServerEventDispatcher eventDispatcher,
             IMqttNetChildLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
             _options = options ?? throw new ArgumentNullException(nameof(options));
-            _sessionsManager = sessionsManager;
+            _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
+            _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
 
             ClientId = clientId;
 
             _logger = logger.CreateChildLogger(nameof(MqttClientSession));
 
             _keepAliveMonitor = new MqttClientKeepAliveMonitor(this, _logger);
-            _subscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server);
+            _subscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, eventDispatcher);
             _pendingPacketsQueue = new MqttClientPendingPacketsQueue(_options, this, _logger);
         }
 
@@ -59,9 +65,9 @@ namespace MQTTnet.Server
         public void FillStatus(MqttClientSessionStatus status)
         {
             status.ClientId = ClientId;
-            status.IsConnected = _adapter != null;
-            status.Endpoint = _adapter?.Endpoint;
-            status.ProtocolVersion = _adapter?.PacketFormatterAdapter?.ProtocolVersion;
+            status.IsConnected = _cancellationTokenSource != null;
+            status.Endpoint = _adapterEndpoint;
+            status.ProtocolVersion = _adapterProtocolVersion;
             status.PendingApplicationMessagesCount = _pendingPacketsQueue.Count;
             status.LastPacketReceived = _keepAliveMonitor.LastPacketReceived;
             status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived;
@@ -73,24 +79,154 @@ namespace MQTTnet.Server
             return _workerTask;
         }
 
-        private async Task RunInternalAsync(MqttApplicationMessage willMessage, int keepAlivePeriod, IMqttChannelAdapter adapter)
+        public void Stop(MqttClientDisconnectType type)
+        {
+            Stop(type, false);
+        }
+
+        public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
+            var checkSubscriptionsResult = _subscriptionsManager.CheckSubscriptions(publishPacket.Topic, publishPacket.QualityOfServiceLevel);
+            if (!checkSubscriptionsResult.IsSubscribed)
+            {
+                return;
+            }
+
+            publishPacket = new MqttPublishPacket
+            {
+                Topic = publishPacket.Topic,
+                Payload = publishPacket.Payload,
+                QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel,
+                Retain = publishPacket.Retain,
+                Dup = false
+            };
+
+            if (publishPacket.QualityOfServiceLevel > 0)
+            {
+                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
+            }
+
+            if (_options.ClientMessageQueueInterceptor != null)
+            {
+                var context = new MqttClientMessageQueueInterceptorContext(
+                    senderClientSession?.ClientId,
+                    ClientId,
+                    publishPacket.ToApplicationMessage());
+
+                _options.ClientMessageQueueInterceptor?.Invoke(context);
+
+                if (!context.AcceptEnqueue || context.ApplicationMessage == null)
+                {
+                    return;
+                }
+
+                publishPacket.Topic = context.ApplicationMessage.Topic;
+                publishPacket.Payload = context.ApplicationMessage.Payload;
+                publishPacket.QualityOfServiceLevel = context.ApplicationMessage.QualityOfServiceLevel;
+            }
+
+            _pendingPacketsQueue.Enqueue(publishPacket);
+        }
+
+        public Task SubscribeAsync(IList<TopicFilter> topicFilters)
+        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            _subscriptionsManager.Subscribe(new MqttSubscribePacket
+            {
+                TopicFilters = topicFilters
+            });
+
+            EnqueueSubscribedRetainedMessages(topicFilters);
+            return Task.FromResult(0);
+        }
+
+        public Task UnsubscribeAsync(IList<string> topicFilters)
+        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            _subscriptionsManager.Unsubscribe(new MqttUnsubscribePacket
+            {
+                TopicFilters = topicFilters
+            });
+
+            return Task.FromResult(0);
+        }
+
+        public void ClearPendingApplicationMessages()
+        {
+            _pendingPacketsQueue.Clear();
+        }
+
+        public void Dispose()
+        {
+            _pendingPacketsQueue?.Dispose();
+
+            _cancellationTokenSource?.Cancel ();
+            _cancellationTokenSource?.Dispose();
+            _cancellationTokenSource = null;
+        }
+
+        private void Stop(MqttClientDisconnectType type, bool isInsideSession)
+        {
+            try
+            {
+                var cts = _cancellationTokenSource;
+                if (cts == null || cts.IsCancellationRequested)
+                {
+                    return;
+                }
+
+                _cancellationTokenSource?.Cancel(false);
+
+                _wasCleanDisconnect = type == MqttClientDisconnectType.Clean;
+
+                if (_willMessage != null && !_wasCleanDisconnect)
+                {
+                    _sessionsManager.EnqueueApplicationMessage(this, _willMessage.ToPublishPacket());
+                }
+
+                _willMessage = null;
+
+                if (!isInsideSession)
+                {
+                    _workerTask?.GetAwaiter().GetResult();
+                }
+            }
+            finally
+            {
+                _logger.Info("Client '{0}': Disconnected (clean={1}).", ClientId, _wasCleanDisconnect);
+                _eventDispatcher.OnClientDisconnected(ClientId, _wasCleanDisconnect);
+            }
+        }
+
+        private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
 
             try
             {
-                _adapter = adapter;
+                if (_cancellationTokenSource != null)
+                {
+                    Stop(MqttClientDisconnectType.Clean, true);
+                }
 
                 adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted;
                 adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted;
 
                 _cancellationTokenSource = new CancellationTokenSource();
 
-                //woraround for https://github.com/dotnet/corefx/issues/24430
+                //workaround for https://github.com/dotnet/corefx/issues/24430
 #pragma warning disable 4014
-                _cleanupHandle = _cancellationTokenSource.Token.Register(() => CleanupAsync());
+                _cleanupHandle = _cancellationTokenSource.Token.Register(async () =>
+                {
+                    await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false);
+                    TryDisposeAdapter(adapter);
+                });
 #pragma warning restore 4014
-                //endworkaround
+                //end workaround
 
                 _wasCleanDisconnect = false;
                 _willMessage = willMessage;
@@ -98,6 +234,9 @@ namespace MQTTnet.Server
                 _pendingPacketsQueue.Start(adapter, _cancellationTokenSource.Token);
                 _keepAliveMonitor.Start(keepAlivePeriod, _cancellationTokenSource.Token);
 
+                _adapterEndpoint = adapter.Endpoint;
+                _adapterProtocolVersion = adapter.PacketSerializer.ProtocolVersion;
+
                 while (!_cancellationTokenSource.IsCancellationRequested)
                 {
                     var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false);
@@ -133,34 +272,35 @@ namespace MQTTnet.Server
             }
             finally
             {
-                await CleanupAsync().ConfigureAwait(false);
+                _adapterEndpoint = null;
+                _adapterProtocolVersion = null;
+
+                // Uncomment as soon as the workaround above is no longer needed.
+                //await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false);
+                //TryDisposeAdapter(adapter);
 
                 _cleanupHandle?.Dispose();
                 _cleanupHandle = null;
                 
-                _adapter = null;
                 _cancellationTokenSource?.Cancel(false);
                 _cancellationTokenSource?.Dispose();
                 _cancellationTokenSource = null;
             }
         }
 
-        private async Task CleanupAsync()
+        private void TryDisposeAdapter(IMqttChannelAdapter adapter)
         {
-            var adapter = _adapter;
+            if (adapter == null)
+            {
+                return;
+            }
+
             try
             {
-                if (adapter == null)
-                {
-                    return;
-                }
-
-                _adapter = null;
-
                 adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
                 adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
 
-                await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
+                adapter.Dispose();
             }
             catch (Exception exception)
             {
@@ -231,65 +371,21 @@ namespace MQTTnet.Server
                 publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
             }
 
-            if (_options.ClientMessageQueueInterceptor != null)
+        private async Task TryDisconnectAdapterAsync(IMqttChannelAdapter adapter)
+        {
+            if (adapter == null)
             {
-                var context = new MqttClientMessageQueueInterceptorContext(
-                    senderClientSession?.ClientId,
-                    ClientId,
-                    publishPacket.ToApplicationMessage());
-
-                _options.ClientMessageQueueInterceptor?.Invoke(context);
-
-                if (!context.AcceptEnqueue || context.ApplicationMessage == null)
-                {
-                    return;
-                }
-
-                publishPacket.Topic = context.ApplicationMessage.Topic;
-                publishPacket.Payload = context.ApplicationMessage.Payload;
-                publishPacket.QualityOfServiceLevel = context.ApplicationMessage.QualityOfServiceLevel;
+                return;
             }
 
-            _pendingPacketsQueue.Enqueue(publishPacket);
-        }
-
-        public Task SubscribeAsync(IList<TopicFilter> topicFilters)
-        {
-            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
-
-            _subscriptionsManager.Subscribe(new MqttSubscribePacket
+            try
             {
-                TopicFilters = topicFilters
-            });
-
-            EnqueueSubscribedRetainedMessages(topicFilters);
-            return Task.FromResult(0);
-        }
-
-        public Task UnsubscribeAsync(IList<string> topicFilters)
-        {
-            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
-
-            _subscriptionsManager.Unsubscribe(new MqttUnsubscribePacket
+                await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
+            }
+            catch (Exception exception)
             {
-                TopicFilters = topicFilters
-            });
-
-            return Task.FromResult(0);
-        }
-
-        public void ClearPendingApplicationMessages()
-        {
-            _pendingPacketsQueue.Clear();
-        }
-
-        public void Dispose()
-        {
-            _pendingPacketsQueue?.Dispose();
-
-            _cancellationTokenSource?.Cancel ();
-            _cancellationTokenSource?.Dispose();
-            _cancellationTokenSource = null;
+                _logger.Error(exception, "Error while disconnecting channel adapter.");
+            }
         }
 
         private void ProcessReceivedPacket(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
@@ -366,7 +462,12 @@ namespace MQTTnet.Server
             var retainedMessages = _retainedMessagesManager.GetSubscribedMessages(topicFilters);
             foreach (var applicationMessage in retainedMessages)
             {
-                EnqueueApplicationMessage(null, applicationMessage);
+                var publishPacket = applicationMessage.ToPublishPacket();
+
+                // Set the retain flag to true according to [MQTT-3.3.1-8].
+                publishPacket.Retain = true;
+
+                EnqueueApplicationMessage(null, publishPacket);
             }
         }
 

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -23,25 +23,29 @@ namespace MQTTnet.Server
         private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
 
         private readonly CancellationToken _cancellationToken;
+        private readonly MqttServerEventDispatcher _eventDispatcher;
 
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
         private readonly IMqttServerOptions _options;
         private readonly IMqttNetChildLogger _logger;
 
-        public MqttClientSessionsManager(IMqttServerOptions options, MqttServer server, MqttRetainedMessagesManager retainedMessagesManager, CancellationToken cancellationToken, IMqttNetChildLogger logger)
+        public MqttClientSessionsManager(
+            IMqttServerOptions options, 
+            MqttRetainedMessagesManager retainedMessagesManager, 
+            CancellationToken cancellationToken,
+            MqttServerEventDispatcher eventDispatcher,
+            IMqttNetChildLogger logger)
         {
-            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            _cancellationToken = cancellationToken;
 
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateChildLogger(nameof(MqttClientSessionsManager));
 
-            _cancellationToken = cancellationToken;
+            _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
             _options = options ?? throw new ArgumentNullException(nameof(options));
-            Server = server ?? throw new ArgumentNullException(nameof(server));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
         }
 
-        public MqttServer Server { get; }
-
         public void Start()
         {
             Task.Factory.StartNew(() => TryProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
@@ -62,7 +66,7 @@ namespace MQTTnet.Server
 
         public Task StartSession(IMqttChannelAdapter clientAdapter)
         {
-            return Task.Run(() => RunSession(clientAdapter, _cancellationToken), _cancellationToken);
+            return Task.Run(() => RunSessionAsync(clientAdapter, _cancellationToken), _cancellationToken);
         }
 
         public IList<IMqttClientSessionStatus> GetClientStatus()
@@ -177,7 +181,7 @@ namespace MQTTnet.Server
                     applicationMessage = interceptorContext.ApplicationMessage;
                 }
 
-                Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
+                _eventDispatcher.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
 
                 if (applicationMessage.Retain)
                 {
@@ -186,7 +190,12 @@ namespace MQTTnet.Server
 
                 foreach (var clientSession in GetSessions())
                 {
-                    clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage);
+                    var publishPacket = applicationMessage.ToPublishPacket();
+
+                    // Set the retain flag to true according to [MQTT-3.3.1-9].
+                    publishPacket.Retain = false;
+
+                    clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, publishPacket);
                 }
             }
             catch (OperationCanceledException)
@@ -206,7 +215,7 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task RunSession(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
+        private async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
         {
             var clientId = string.Empty;
             
@@ -239,7 +248,6 @@ namespace MQTTnet.Server
                 }
 
                 var result = PrepareClientSession(connectPacket);
-                var clientSession = result.Session;
 
                 await clientAdapter.SendPacketAsync(
                     new MqttConnAckPacket
@@ -249,9 +257,10 @@ namespace MQTTnet.Server
                     },
                     cancellationToken).ConfigureAwait(false);
 
-                Server.OnClientConnected(clientId);
+                _logger.Info("Client '{0}': Connected.", clientId);
+                _eventDispatcher.OnClientConnected(clientId);
 
-                await clientSession.RunAsync(connectPacket.WillMessage, connectPacket.KeepAlivePeriod, clientAdapter).ConfigureAwait(false);
+                await result.Session.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
             }
             catch (OperationCanceledException)
             {
@@ -315,7 +324,7 @@ namespace MQTTnet.Server
                 {
                     isExistingSession = false;
 
-                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _retainedMessagesManager, _logger);
+                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _retainedMessagesManager, _eventDispatcher, _logger);
                     _sessions[connectPacket.ClientId] = clientSession;
 
                     _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);

Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs

@@ -10,14 +10,14 @@ namespace MQTTnet.Server
     {
         private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly IMqttServerOptions _options;
-        private readonly MqttServer _server;
+        private readonly MqttServerEventDispatcher _eventDispatcher;
         private readonly string _clientId;
 
-        public MqttClientSubscriptionsManager(string clientId, IMqttServerOptions options, MqttServer server)
+        public MqttClientSubscriptionsManager(string clientId, IMqttServerOptions options, MqttServerEventDispatcher eventDispatcher)
         {
             _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
             _options = options ?? throw new ArgumentNullException(nameof(options));
-            _server = server;
+            _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
         }
 
         public MqttClientSubscribeResult Subscribe(MqttSubscribePacket subscribePacket)
@@ -58,7 +58,7 @@ namespace MQTTnet.Server
                         _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
                     }
 
-                    _server.OnClientSubscribedTopic(_clientId, topicFilter);
+                    _eventDispatcher.OnClientSubscribedTopic(_clientId, topicFilter);
                 }
             }
 
@@ -75,7 +75,7 @@ namespace MQTTnet.Server
                 {
                     _subscriptions.Remove(topicFilter);
 
-                    _server.OnClientUnsubscribedTopic(_clientId, topicFilter);
+                    _eventDispatcher.OnClientUnsubscribedTopic(_clientId, topicFilter);
                 }
             }
 

Source/MQTTnet/Server/MqttServer.cs

@@ -11,6 +11,7 @@ namespace MQTTnet.Server
 {
     public class MqttServer : IMqttServer
     {
+        private readonly MqttServerEventDispatcher _eventDispatcher = new MqttServerEventDispatcher();
         private readonly ICollection<IMqttServerAdapter> _adapters;
         private readonly IMqttNetChildLogger _logger;
 
@@ -21,10 +22,16 @@ namespace MQTTnet.Server
         public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetChildLogger logger)
         {
             if (adapters == null) throw new ArgumentNullException(nameof(adapters));
+            _adapters = adapters.ToList();
+
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateChildLogger(nameof(MqttServer));
 
-            _adapters = adapters.ToList();
+            _eventDispatcher.ClientConnected += (s, e) => ClientConnected?.Invoke(s, e);
+            _eventDispatcher.ClientDisconnected += (s, e) => ClientDisconnected?.Invoke(s, e);
+            _eventDispatcher.ClientSubscribedTopic += (s, e) => ClientSubscribedTopic?.Invoke(s, e);
+            _eventDispatcher.ClientUnsubscribedTopic += (s, e) => ClientUnsubscribedTopic?.Invoke(s, e);
+            _eventDispatcher.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
         }
 
         public event EventHandler Started;
@@ -92,7 +99,7 @@ namespace MQTTnet.Server
             _retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
             await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false);
 
-            _clientSessionsManager = new MqttClientSessionsManager(Options, this, _retainedMessagesManager, _cancellationTokenSource.Token, _logger);
+            _clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _cancellationTokenSource.Token, _eventDispatcher, _logger);
             _clientSessionsManager.Start();
 
             foreach (var adapter in _adapters)
@@ -144,33 +151,6 @@ namespace MQTTnet.Server
             return _retainedMessagesManager?.ClearMessagesAsync();
         }
 
-        internal void OnClientConnected(string clientId)
-        {
-            _logger.Info("Client '{0}': Connected.", clientId);
-            ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientId));
-        }
-
-        internal void OnClientDisconnected(string clientId, bool wasCleanDisconnect)
-        {
-            _logger.Info("Client '{0}': Disconnected (clean={1}).", clientId, wasCleanDisconnect);
-            ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientId, wasCleanDisconnect));
-        }
-
-        internal void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
-        {
-            ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
-        }
-
-        internal void OnClientUnsubscribedTopic(string clientId, string topicFilter)
-        {
-            ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
-        }
-
-        internal void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
-        {
-            ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(clientId, applicationMessage));
-        }
-
         private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
         {
             eventArgs.SessionTask = _clientSessionsManager.StartSession(eventArgs.Client);

Source/MQTTnet/Server/MqttServerEventDispatcher.cs

@@ -0,0 +1,42 @@
+using System;
+
+namespace MQTTnet.Server
+{
+    public class MqttServerEventDispatcher
+    {
+        public event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
+
+        public event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
+
+        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+
+        public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
+
+        public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
+
+        public void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
+        {
+            ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
+        }
+
+        public void OnClientUnsubscribedTopic(string clientId, string topicFilter)
+        {
+            ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
+        }
+
+        public void OnClientDisconnected(string clientId, bool wasCleanDisconnect)
+        {
+            ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientId, wasCleanDisconnect));
+        }
+
+        public void OnApplicationMessageReceived(string senderClientId, MqttApplicationMessage applicationMessage)
+        {
+            ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage));
+        }
+
+        public void OnClientConnected(string clientId)
+        {
+            ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientId));
+        }
+    }
+}