Update MQTTnet.sln

Update Startup.cs

Build/MQTTnet.AspNetCore.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet.AspNetCore</id>
-    <version>2.7.0</version>
+    <version>2.7.5</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,13 +10,13 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>This is a support library to integrate MQTTnet into AspNetCore.</description>
-    <releaseNotes>* Updated to MQTTnet 2.7.0.
+    <releaseNotes>* Updated to MQTTnet 2.7.5.
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2018</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
     <dependencies>
       <group targetFramework="netstandard2.0">
-        <dependency id="MQTTnet" version="2.7.0" />
+        <dependency id="MQTTnet" version="2.7.5" />
       </group>
     </dependencies>
   </metadata>

Build/MQTTnet.Extensions.Rpc.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet.Extensions.Rpc</id>
-    <version>2.6.0</version>
+    <version>2.7.5</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,32 +10,32 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>This is a extension library which allows executing synchronous device calls including a response using MQTTnet.</description>
-    <releaseNotes>* Initial version.
+    <releaseNotes>* Updated to MQTTnet 2.7.5.
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2017</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
     <dependencies>
       <group targetFramework="netstandard2.0">
-        <dependency id="MQTTnet" version="2.6.0-rc1" />
+        <dependency id="MQTTnet" version="2.7.5" />
       </group>
       <group targetFramework="netstandard1.3">
-        <dependency id="MQTTnet" version="2.6.0-rc1" />  
+        <dependency id="MQTTnet" version="2.7.5" />  
       </group>
 
       <group targetFramework="netstandard2.0">
-        <dependency id="MQTTnet" version="2.6.0-rc1" />
+        <dependency id="MQTTnet" version="2.7.5" />
       </group>
       
       <group targetFramework="uap10.0">
-        <dependency id="MQTTnet" version="2.6.0-rc1" />
+        <dependency id="MQTTnet" version="2.7.5" />
       </group>
         
       <group targetFramework="net452">
-        <dependency id="MQTTnet" version="2.6.0-rc1" />
+        <dependency id="MQTTnet" version="2.7.5" />
       </group>
 
       <group targetFramework="net461">
-        <dependency id="MQTTnet" version="2.6.0-rc1" />
+        <dependency id="MQTTnet" version="2.7.5" />
       </group>
     </dependencies>
   </metadata>

Build/MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.7.1</version>
+    <version>2.7.5</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,15 +10,13 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
-    <releaseNotes>* [Core] Fixed wrong parsing of ConnAck packet for protocol version 3.1.0.
-* [Core] Log messages are now overriding ToString() and providing a ready to use text representation.
-* [Client] Optimized package dispatcher and added several new exceptions.
-* [Client] The _ManagedClient_ now has an event which is fired after a queued application message was processed (including exception).
-* [Client] The _ManagedClient_ now supports unsubscribing (thanks to @lerppana)
-* [Server] Fixed some minor async issues.
-* [Server] Fixed wrong comparison of the topic and QoS for retained messages.
-* [Server] Added a property which provides access to the used options (read only).
-* [Server] Fixed a null ref expection when using an interceptor and publishing via the server directly.
+    <releaseNotes> * [Client] Fixed a deadlock while the client disconnects.
+* [Client] Fixed broken support for protocol version 3.1.0.
+* [Server] The _MqttTcpServerAdapter_ is now added to the ASP.NET services.
+* [Server] _MqttServerAdapter_ is renamed to _MqttTcpServerAdapter_ (BREAKING CHANGE!).
+* [Server] The server no longer sends the will message of a client if the disconnect was clean (via _Disconnect_ packet).
+* [Server] The application message interceptor now allows closing the connection.
+* [Server] Added a new flag for the _ClientDisconnected_ event which contains a value indicating whether the disconnect was clean (via _Disconnect_ packet).
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2018</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
@@ -47,7 +45,7 @@
 
       <group targetFramework="net461">
       </group>
-
+      
     </dependencies>
   </metadata>
 

Frameworks/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs

@@ -4,6 +4,7 @@ using Microsoft.Extensions.Hosting;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Server;
+using MQTTnet.Implementations;
 
 namespace MQTTnet.AspNetCore
 {
@@ -20,7 +21,9 @@ namespace MQTTnet.AspNetCore
             services.AddSingleton<IMqttServer>(s => s.GetService<MqttHostedServer>());
             
             services.AddSingleton<MqttWebSocketServerAdapter>();
-            services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttWebSocketServerAdapter>());
+            services.AddSingleton<MqttTcpServerAdapter>();
+            services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttWebSocketServerAdapter>()); 
+            services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttTcpServerAdapter>());
 
             return services;
         }

Frameworks/MQTTnet.NetStandard/Adapter/IMqttChannelAdapter.cs

@@ -7,7 +7,7 @@ using MQTTnet.Serializer;
 
 namespace MQTTnet.Adapter
 {
-    public interface IMqttChannelAdapter
+    public interface IMqttChannelAdapter : IDisposable
     {
         IMqttPacketSerializer PacketSerializer { get; }
 

Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs

@@ -14,11 +14,12 @@ using MQTTnet.Serializer;
 
 namespace MQTTnet.Adapter
 {
-    public class MqttChannelAdapter : IMqttChannelAdapter
+    public sealed class MqttChannelAdapter : IMqttChannelAdapter
     {
         private const uint ErrorOperationAborted = 0x800703E3;
         private const int ReadBufferSize = 4096;  // TODO: Move buffer size to config
 
+        private bool _isDisposed;
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
         private readonly IMqttNetLogger _logger;
         private readonly IMqttChannel _channel;
@@ -34,20 +35,24 @@ namespace MQTTnet.Adapter
 
         public Task ConnectAsync(TimeSpan timeout)
         {
-            _logger.Trace<MqttChannelAdapter>("Connecting [Timeout={0}]", timeout);
+            ThrowIfDisposed();
+            _logger.Verbose<MqttChannelAdapter>("Connecting [Timeout={0}]", timeout);
 
             return ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout));
         }
 
         public Task DisconnectAsync(TimeSpan timeout)
         {
-            _logger.Trace<MqttChannelAdapter>("Disconnecting [Timeout={0}]", timeout);
+            ThrowIfDisposed();
+            _logger.Verbose<MqttChannelAdapter>("Disconnecting [Timeout={0}]", timeout);
 
             return ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout));
         }
 
         public Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable<MqttBasePacket> packets)
         {
+            ThrowIfDisposed();
+
             return ExecuteAndWrapExceptionAsync(async () =>
             {
                 await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
@@ -55,20 +60,35 @@ namespace MQTTnet.Adapter
                 {
                     foreach (var packet in packets)
                     {
+                        if (cancellationToken.IsCancellationRequested)
+                        {
+                            return;
+                        }
+
                         if (packet == null)
                         {
                             continue;
                         }
 
-                        _logger.Trace<MqttChannelAdapter>("TX >>> {0} [Timeout={1}]", packet, timeout);
+                        _logger.Verbose<MqttChannelAdapter>("TX >>> {0} [Timeout={1}]", packet, timeout);
 
                         var chunks = PacketSerializer.Serialize(packet);
                         foreach (var chunk in chunks)
                         {
+                            if (cancellationToken.IsCancellationRequested)
+                            {
+                                return;
+                            }
+
                             await _channel.SendStream.WriteAsync(chunk.Array, chunk.Offset, chunk.Count, cancellationToken).ConfigureAwait(false);
                         }
                     }
 
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return;
+                    }
+
                     if (timeout > TimeSpan.Zero)
                     {
                         await _channel.SendStream.FlushAsync(cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
@@ -87,6 +107,8 @@ namespace MQTTnet.Adapter
 
         public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
+            ThrowIfDisposed();
+
             MqttBasePacket packet = null;
             await ExecuteAndWrapExceptionAsync(async () =>
             {
@@ -113,7 +135,7 @@ namespace MQTTnet.Adapter
                         throw new MqttProtocolViolationException("Received malformed packet.");
                     }
 
-                    _logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
+                    _logger.Verbose<MqttChannelAdapter>("RX <<< {0}", packet);
                 }
                 finally
                 {
@@ -138,7 +160,7 @@ namespace MQTTnet.Adapter
             }
 
             var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(header.BodyLength) : new MemoryStream();
-            
+
             var buffer = new byte[ReadBufferSize];
             while (body.Length < header.BodyLength)
             {
@@ -149,7 +171,7 @@ namespace MQTTnet.Adapter
                 }
 
                 var readBytesCount = await stream.ReadAsync(buffer, 0, bytesLeft, cancellationToken).ConfigureAwait(false);
-                
+
                 // Check if the client closed the connection before sending the full body.
                 if (readBytesCount == 0)
                 {
@@ -162,7 +184,7 @@ namespace MQTTnet.Adapter
             }
 
             body.Seek(0L, SeekOrigin.Begin);
-            
+
             return new ReceivedMqttPacket(header, body);
         }
 
@@ -190,7 +212,7 @@ namespace MQTTnet.Adapter
             }
             catch (COMException comException)
             {
-                if ((uint) comException.HResult == ErrorOperationAborted)
+                if ((uint)comException.HResult == ErrorOperationAborted)
                 {
                     throw new OperationCanceledException();
                 }
@@ -214,5 +236,20 @@ namespace MQTTnet.Adapter
                 throw new MqttCommunicationException(exception);
             }
         }
+
+        public void Dispose()
+        {
+            _isDisposed = true;
+            _semaphore?.Dispose();
+            _channel?.Dispose();
+        }
+
+        private void ThrowIfDisposed()
+        {
+            if (_isDisposed)
+            {
+                throw new ObjectDisposedException(nameof(MqttChannelAdapter));
+            }
+        }
     }
 }

Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapterExtensions.cs

@@ -1,17 +0,0 @@
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Packets;
-
-namespace MQTTnet.Adapter
-{
-    public static class MqttChannelAdapterExtensions
-    {
-        public static Task SendPacketsAsync(this IMqttChannelAdapter adapter, TimeSpan timeout, CancellationToken cancellationToken, params MqttBasePacket[] packets)
-        {
-            if (adapter == null) throw new ArgumentNullException(nameof(adapter));
-
-            return adapter.SendPacketsAsync(timeout, cancellationToken, packets);
-        }
-    }
-}

Frameworks/MQTTnet.NetStandard/ApplicationMessagePublisherExtensions.cs

@@ -1,16 +1,41 @@
 using System;
 using System.Threading.Tasks;
+using MQTTnet.Protocol;
 
 namespace MQTTnet
 {
     public static class ApplicationMessagePublisherExtensions
     {
-        public static Task PublishAsync(this IApplicationMessagePublisher client, params MqttApplicationMessage[] applicationMessages)
+        public static Task PublishAsync(this IApplicationMessagePublisher publisher, params MqttApplicationMessage[] applicationMessages)
         {
-            if (client == null) throw new ArgumentNullException(nameof(client));
+            if (publisher == null) throw new ArgumentNullException(nameof(publisher));
             if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
 
-            return client.PublishAsync(applicationMessages);
+            return publisher.PublishAsync(applicationMessages);
+        }
+
+        public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic)
+        {
+            if (publisher == null) throw new ArgumentNullException(nameof(publisher));
+            if (topic == null) throw new ArgumentNullException(nameof(topic));
+            
+            return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).Build());
+        }
+
+        public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload)
+        {
+            if (publisher == null) throw new ArgumentNullException(nameof(publisher));
+            if (topic == null) throw new ArgumentNullException(nameof(topic));
+
+            return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(payload).Build());
+        }
+
+        public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        {
+            if (publisher == null) throw new ArgumentNullException(nameof(publisher));
+            if (topic == null) throw new ArgumentNullException(nameof(topic));
+
+            return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(payload).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Channel/IMqttChannel.cs

@@ -1,9 +1,10 @@
-using System.IO;
+using System;
+using System.IO;
 using System.Threading.Tasks;
 
 namespace MQTTnet.Channel
 {
-    public interface IMqttChannel
+    public interface IMqttChannel : IDisposable
     {
         Stream SendStream { get; }
         Stream ReceiveStream { get; }

Frameworks/MQTTnet.NetStandard/Client/IMqttClientAdapterFactory.cs

@@ -5,6 +5,6 @@ namespace MQTTnet.Client
 {
     public interface IMqttClientAdapterFactory
     {
-        IMqttChannelAdapter CreateClientAdapter(IMqttClientChannelOptions options, IMqttNetLogger logger);
+        IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger);
     }
 }

Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs

@@ -13,6 +13,8 @@ namespace MQTTnet.Client
         
         TimeSpan CommunicationTimeout { get; }
         TimeSpan KeepAlivePeriod { get; }
+        TimeSpan? KeepAliveSendInterval { get; }
+
         MqttProtocolVersion ProtocolVersion { get; }
 
         IMqttClientChannelOptions ChannelOptions { get; }

Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -15,6 +16,8 @@ namespace MQTTnet.Client
     public class MqttClient : IMqttClient
     {
         private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
+        private readonly Stopwatch _sendTracker = new Stopwatch();
+        private readonly SemaphoreSlim _disconnectLock = new SemaphoreSlim(1, 1);
         private readonly IMqttClientAdapterFactory _adapterFactory;
         private readonly MqttPacketDispatcher _packetDispatcher;
         private readonly IMqttNetLogger _logger;
@@ -22,13 +25,15 @@ namespace MQTTnet.Client
         private IMqttClientOptions _options;
         private bool _isReceivingPackets;
         private CancellationTokenSource _cancellationTokenSource;
+        private Task _packetReceiverTask;
+        private Task _keepAliveMessageSenderTask;
         private IMqttChannelAdapter _adapter;
 
         public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
         {
             _adapterFactory = channelFactory ?? throw new ArgumentNullException(nameof(channelFactory));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
-            
+
             _packetDispatcher = new MqttPacketDispatcher(logger);
         }
 
@@ -52,30 +57,34 @@ namespace MQTTnet.Client
                 _packetIdentifierProvider.Reset();
                 _packetDispatcher.Reset();
 
-                _adapter = _adapterFactory.CreateClientAdapter(options.ChannelOptions, _logger);
-                
-                _logger.Trace<MqttClient>("Trying to connect with server.");
+                _adapter = _adapterFactory.CreateClientAdapter(options, _logger);
+
+                _logger.Verbose<MqttClient>("Trying to connect with server.");
                 await _adapter.ConnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
-                _logger.Trace<MqttClient>("Connection with server established.");
+                _logger.Verbose<MqttClient>("Connection with server established.");
+
+                await StartReceivingPacketsAsync().ConfigureAwait(false);
 
-                await StartReceivingPacketsAsync(_cancellationTokenSource.Token).ConfigureAwait(false);
                 var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false);
+                _logger.Verbose<MqttClient>("MQTT connection with server established.");
 
-                _logger.Trace<MqttClient>("MQTT connection with server established.");
+                _sendTracker.Restart();
 
                 if (_options.KeepAlivePeriod != TimeSpan.Zero)
                 {
-                    StartSendingKeepAliveMessages(_cancellationTokenSource.Token);
+                    StartSendingKeepAliveMessages();
                 }
 
                 IsConnected = true;
                 Connected?.Invoke(this, new MqttClientConnectedEventArgs(connectResponse.IsSessionPresent));
+
+                _logger.Info<MqttClient>("Connected.");
                 return new MqttClientConnectResult(connectResponse.IsSessionPresent);
             }
             catch (Exception exception)
             {
                 _logger.Error<MqttClient>(exception, "Error while connecting with server.");
-                await DisconnectInternalAsync(exception).ConfigureAwait(false);
+                await DisconnectInternalAsync(null, exception).ConfigureAwait(false);
 
                 throw;
             }
@@ -90,11 +99,14 @@ namespace MQTTnet.Client
 
             try
             {
-                await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false);
+                if (!_cancellationTokenSource.IsCancellationRequested)
+                {
+                    await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false);
+                }
             }
             finally
             {
-                await DisconnectInternalAsync(null).ConfigureAwait(false);
+                await DisconnectInternalAsync(null, null).ConfigureAwait(false);
             }
         }
 
@@ -140,7 +152,7 @@ namespace MQTTnet.Client
             ThrowIfNotConnected();
 
             var publishPackets = applicationMessages.Select(m => m.ToPublishPacket());
-            var packetGroups = publishPackets.GroupBy(p => p.QualityOfServiceLevel).OrderBy(g => g.Key); 
+            var packetGroups = publishPackets.GroupBy(p => p.QualityOfServiceLevel).OrderBy(g => g.Key);
 
             foreach (var qosGroup in packetGroups)
             {
@@ -149,7 +161,7 @@ namespace MQTTnet.Client
                     case MqttQualityOfServiceLevel.AtMostOnce:
                         {
                             // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
-                            await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosGroup).ConfigureAwait(false);
+                            await SendAsync(qosGroup.Cast<MqttBasePacket>().ToArray()).ConfigureAwait(false);
                             break;
                         }
                     case MqttQualityOfServiceLevel.AtLeastOnce:
@@ -167,8 +179,14 @@ namespace MQTTnet.Client
                             foreach (var publishPacket in qosGroup)
                             {
                                 publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
+
                                 var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
-                                await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).ConfigureAwait(false);
+                                var pubRelPacket = new MqttPubRelPacket
+                                {
+                                    PacketIdentifier = pubRecPacket.PacketIdentifier
+                                };
+
+                                await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket).ConfigureAwait(false);
                             }
 
                             break;
@@ -184,6 +202,9 @@ namespace MQTTnet.Client
         public void Dispose()
         {
             _cancellationTokenSource?.Dispose();
+            _cancellationTokenSource = null;
+
+            _adapter?.Dispose();
         }
 
         private async Task<MqttConnAckPacket> AuthenticateAsync(MqttApplicationMessage willApplicationMessage)
@@ -217,25 +238,17 @@ namespace MQTTnet.Client
             if (IsConnected) throw new MqttProtocolViolationException(message);
         }
 
-        private async Task DisconnectInternalAsync(Exception exception)
+        private async Task DisconnectInternalAsync(Task sender, Exception exception)
         {
-            var clientWasConnected = IsConnected;
-            IsConnected = false;
-
-            var cts = _cancellationTokenSource;
-            if (cts == null || cts.IsCancellationRequested)
-            {
-                return;
-            }
-
-            cts.Cancel(false);
-            cts.Dispose();
-            _cancellationTokenSource = null;
-
+            await _disconnectLock.WaitAsync();
             try
             {
-                await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
-                _logger.Info<MqttClient>("Disconnected from adapter.");
+                if (_cancellationTokenSource == null || _cancellationTokenSource.IsCancellationRequested)
+                {
+                    return;
+                }
+
+                _cancellationTokenSource.Cancel(false);
             }
             catch (Exception adapterException)
             {
@@ -243,6 +256,42 @@ namespace MQTTnet.Client
             }
             finally
             {
+                _disconnectLock.Release();
+            }
+
+            var clientWasConnected = IsConnected;
+            IsConnected = false;
+
+            try
+            {
+                if (_packetReceiverTask != null && _packetReceiverTask != sender)
+                {
+                    _packetReceiverTask.Wait();
+                }
+
+                if (_keepAliveMessageSenderTask != null && _keepAliveMessageSenderTask != sender)
+                {
+                    _keepAliveMessageSenderTask.Wait();
+                }
+
+                if (_adapter != null)
+                {
+                    await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
+                }
+                
+                _logger.Verbose<MqttClient>("Disconnected from adapter.");
+            }
+            catch (Exception adapterException)
+            {
+                _logger.Warning<MqttClient>(adapterException, "Error while disconnecting from adapter.");
+            }
+            finally
+            {
+                _adapter?.Dispose();
+                _adapter = null;
+                _cancellationTokenSource?.Dispose();
+                _cancellationTokenSource = null;
+
                 _logger.Info<MqttClient>("Disconnected.");
                 Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected, exception));
             }
@@ -252,8 +301,6 @@ namespace MQTTnet.Client
         {
             try
             {
-                _logger.Info<MqttClient>("Received <<< {0}", packet);
-                
                 if (packet is MqttPublishPacket publishPacket)
                 {
                     await ProcessReceivedPublishPacketAsync(publishPacket).ConfigureAwait(false);
@@ -301,6 +348,11 @@ namespace MQTTnet.Client
 
         private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket)
         {
+            if (_cancellationTokenSource.IsCancellationRequested)
+            {
+                return Task.FromResult(0);
+            }
+
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
                 FireApplicationMessageReceivedEvent(publishPacket);
@@ -325,151 +377,140 @@ namespace MQTTnet.Client
 
         private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket)
         {
-            return SendAsync(pubRelPacket.CreateResponse<MqttPubCompPacket>());
+            var response = new MqttPubCompPacket
+            {
+                PacketIdentifier = pubRelPacket.PacketIdentifier
+            };
+
+            return SendAsync(response);
         }
 
-        private Task SendAsync(MqttBasePacket packet)
+        private Task SendAsync(params MqttBasePacket[] packets)
         {
-            return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packet);
+            _sendTracker.Restart();
+            return _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, packets);
         }
 
         private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket
         {
-            ushort identifier = 0;
+            ushort? identifier = null;
             if (requestPacket is IMqttPacketWithIdentifier requestPacketWithIdentifier)
             {
                 identifier = requestPacketWithIdentifier.PacketIdentifier;
             }
 
             var packetAwaiter = _packetDispatcher.WaitForPacketAsync(typeof(TResponsePacket), identifier, _options.CommunicationTimeout);
-            await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false);
+            await SendAsync(requestPacket).ConfigureAwait(false);
+
             return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
         }
 
-        private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
+        private async Task SendKeepAliveMessagesAsync()
         {
-            _logger.Info<MqttClient>("Start sending keep alive packets.");
+            _logger.Verbose<MqttClient>("Start sending keep alive packets.");
 
             try
             {
-                while (!cancellationToken.IsCancellationRequested)
+                while (!_cancellationTokenSource.Token.IsCancellationRequested)
                 {
-                    await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
-                    await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false);
-                }
-            }
-            catch (OperationCanceledException)
-            {
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    return;
-                }
+                    var keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75);
+                    if (_options.KeepAliveSendInterval.HasValue)
+                    {
+                        keepAliveSendInterval = _options.KeepAliveSendInterval.Value;
+                    }
 
-                await DisconnectInternalAsync(null).ConfigureAwait(false);
-            }
-            catch (MqttCommunicationException exception)
-            {
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    return;
-                }
+                    if (_sendTracker.Elapsed > keepAliveSendInterval)
+                    {
+                        await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
+                    }
 
-                _logger.Warning<MqttClient>(exception, "MQTT communication exception while sending/receiving keep alive packets.");
-                await DisconnectInternalAsync(exception).ConfigureAwait(false);
+                    await Task.Delay(keepAliveSendInterval, _cancellationTokenSource.Token).ConfigureAwait(false);
+                }
             }
             catch (Exception exception)
             {
-                _logger.Warning<MqttClient>(exception, "Unhandled exception while sending/receiving keep alive packets.");
-                await DisconnectInternalAsync(exception).ConfigureAwait(false);
+                if (exception is OperationCanceledException)
+                {
+                }
+                else if (exception is MqttCommunicationException)
+                {
+                    _logger.Warning<MqttClient>(exception, "MQTT communication exception while sending/receiving keep alive packets.");
+                }
+                else
+                {
+                    _logger.Error<MqttClient>(exception, "Unhandled exception while sending/receiving keep alive packets.");
+                }
+                
+                await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false);
             }
             finally
             {
-                _logger.Info<MqttClient>("Stopped sending keep alive packets.");
+                _logger.Verbose<MqttClient>("Stopped sending keep alive packets.");
             }
         }
 
-        private async Task ReceivePacketsAsync(CancellationToken cancellationToken)
+        private async Task ReceivePacketsAsync()
         {
-            _logger.Info<MqttClient>("Start receiving packets.");
+            _logger.Verbose<MqttClient>("Start receiving packets.");
 
             try
             {
-                while (!cancellationToken.IsCancellationRequested)
+                while (!_cancellationTokenSource.Token.IsCancellationRequested)
                 {
                     _isReceivingPackets = true;
 
-                    var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
+                    var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false);
 
-                    if (cancellationToken.IsCancellationRequested)
+                    if (_cancellationTokenSource.Token.IsCancellationRequested)
                     {
                         return;
                     }
 
-                    StartProcessReceivedPacket(packet, cancellationToken);
+                    StartProcessReceivedPacket(packet);
                 }
             }
-            catch (OperationCanceledException)
-            {
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    return;
-                }
-
-                await DisconnectInternalAsync(null).ConfigureAwait(false);
-            }
-            catch (MqttCommunicationException exception)
-            {
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    return;
-                }
-
-                _logger.Warning<MqttClient>(exception, "MQTT communication exception while receiving packets.");
-                await DisconnectInternalAsync(exception).ConfigureAwait(false);
-            }
             catch (Exception exception)
             {
-                _logger.Error<MqttClient>(exception, "Unhandled exception while receiving packets.");
-                await DisconnectInternalAsync(exception).ConfigureAwait(false);
+                if (exception is OperationCanceledException)
+                {
+                }
+                else if (exception is MqttCommunicationException)
+                {
+                    _logger.Warning<MqttClient>(exception, "MQTT communication exception while receiving packets.");
+                }
+                else
+                {
+                    _logger.Error<MqttClient>(exception, "Unhandled exception while receiving packets.");
+                }
+
+                await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false);
             }
             finally
             {
-                _logger.Info<MqttClient>("Stopped receiving packets.");
+                _logger.Verbose<MqttClient>("Stopped receiving packets.");
             }
         }
 
-        private void StartProcessReceivedPacket(MqttBasePacket packet, CancellationToken cancellationToken)
+        private void StartProcessReceivedPacket(MqttBasePacket packet)
         {
-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-            Task.Run(
-                async () => await ProcessReceivedPacketAsync(packet).ConfigureAwait(false),
-                cancellationToken).ConfigureAwait(false);
-#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            Task.Run(() => ProcessReceivedPacketAsync(packet), _cancellationTokenSource.Token);
         }
 
-        private async Task StartReceivingPacketsAsync(CancellationToken cancellationToken)
+        private async Task StartReceivingPacketsAsync()
         {
             _isReceivingPackets = false;
 
-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-            Task.Run(
-                async () => await ReceivePacketsAsync(cancellationToken).ConfigureAwait(false),
-                cancellationToken).ConfigureAwait(false);
-#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            _packetReceiverTask = Task.Run(ReceivePacketsAsync, _cancellationTokenSource.Token);
 
-            while (!_isReceivingPackets && !cancellationToken.IsCancellationRequested)
+            while (!_isReceivingPackets && !_cancellationTokenSource.Token.IsCancellationRequested)
             {
-                await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken).ConfigureAwait(false);
+                await Task.Delay(TimeSpan.FromMilliseconds(100), _cancellationTokenSource.Token).ConfigureAwait(false);
             }
         }
 
-        private void StartSendingKeepAliveMessages(CancellationToken cancellationToken)
+        private void StartSendingKeepAliveMessages()
         {
-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-            Task.Run(
-                async () => await SendKeepAliveMessagesAsync(cancellationToken).ConfigureAwait(false), 
-                cancellationToken).ConfigureAwait(false);
-#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            _keepAliveMessageSenderTask = Task.Run(SendKeepAliveMessagesAsync, _cancellationTokenSource.Token);
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Client/MqttClientExtensions.cs

@@ -24,6 +24,14 @@ namespace MQTTnet.Client
             return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
         }
 
+        public static Task<IList<MqttSubscribeResult>> SubscribeAsync(this IMqttClient client, string topic)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+            if (topic == null) throw new ArgumentNullException(nameof(topic));
+
+            return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());
+        }
+
         public static Task UnsubscribeAsync(this IMqttClient client, params string[] topicFilters)
         {
             if (client == null) throw new ArgumentNullException(nameof(client));

Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs

@@ -13,7 +13,9 @@ namespace MQTTnet.Client
 
         public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials();
 
-        public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5);
+        public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15);
+
+        public TimeSpan? KeepAliveSendInterval { get; set; }
 
         public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
 

Frameworks/MQTTnet.NetStandard/Client/MqttClientTcpOptions.cs

@@ -6,6 +6,8 @@
 
         public int? Port { get; set; }
 
+        public int BufferSize { get; set; } = 20 * 4096;
+
         public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
     }
 }

Frameworks/MQTTnet.NetStandard/Client/MqttClientWebSocketOptions.cs

@@ -9,7 +9,7 @@ namespace MQTTnet.Client
 
         public IDictionary<string, string> RequestHeaders { get; set; }
 
-        public ICollection<string> SubProtocols { get; set; }
+        public ICollection<string> SubProtocols { get; set; } = new List<string> { "mqtt" };
 
         public CookieContainer CookieContainer { get; set; }
 

Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs

@@ -18,7 +18,7 @@ namespace MQTTnet.Client
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
-        public async Task<MqttBasePacket> WaitForPacketAsync(Type responseType, ushort identifier, TimeSpan timeout)
+        public async Task<MqttBasePacket> WaitForPacketAsync(Type responseType, ushort? identifier, TimeSpan timeout)
         {
             var packetAwaiter = AddPacketAwaiter(responseType, identifier);
             try
@@ -44,13 +44,13 @@ namespace MQTTnet.Client
 
             if (_awaiters.TryGetValue(type, out var byId))
             {
-                ushort identifier = 0;
+                ushort? identifier = 0;
                 if (packet is IMqttPacketWithIdentifier packetWithIdentifier)
                 {
                     identifier = packetWithIdentifier.PacketIdentifier;
                 }
 
-                if (byId.TryRemove(identifier, out var tcs))
+                if (byId.TryRemove(identifier.Value, out var tcs))
                 {
                     tcs.TrySetResult(packet);
                     return;
@@ -65,12 +65,17 @@ namespace MQTTnet.Client
             _awaiters.Clear();
         }
 
-        private TaskCompletionSource<MqttBasePacket> AddPacketAwaiter(Type responseType, ushort identifier)
+        private TaskCompletionSource<MqttBasePacket> AddPacketAwaiter(Type responseType, ushort? identifier)
         {
             var tcs = new TaskCompletionSource<MqttBasePacket>();
 
+            if (!identifier.HasValue)
+            {
+                identifier = 0;
+            }
+
             var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
-            if (!byId.TryAdd(identifier, tcs))
+            if (!byId.TryAdd(identifier.Value, tcs))
             {
                 throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{responseType}' with identifier {identifier}.");
             }
@@ -78,10 +83,15 @@ namespace MQTTnet.Client
             return tcs;
         }
 
-        private void RemovePacketAwaiter(Type responseType, ushort identifier)
+        private void RemovePacketAwaiter(Type responseType, ushort? identifier)
         {
+            if (!identifier.HasValue)
+            {
+                identifier = 0;
+            }
+
             var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
-            byId.TryRemove(identifier, out var _);
+            byId.TryRemove(identifier.Value, out var _);
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Diagnostics/IMqttNetLogger.cs

@@ -6,7 +6,7 @@ namespace MQTTnet.Diagnostics
     {
         event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
 
-        void Trace<TSource>(string message, params object[] parameters);
+        void Verbose<TSource>(string message, params object[] parameters);
 
         void Info<TSource>(string message, params object[] parameters);
 

Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs

@@ -13,7 +13,7 @@ namespace MQTTnet.Diagnostics
 
         public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
 
-        public void Trace<TSource>(string message, params object[] parameters)
+        public void Verbose<TSource>(string message, params object[] parameters)
         {
             Publish<TSource>(MqttNetLogLevel.Verbose, null, message, parameters);
         }

Frameworks/MQTTnet.NetStandard/Implementations/MqttClientAdapterFactory.cs

@@ -8,18 +8,28 @@ namespace MQTTnet.Implementations
 {
     public class MqttClientAdapterFactory : IMqttClientAdapterFactory
     {
-        public IMqttChannelAdapter CreateClientAdapter(IMqttClientChannelOptions options, IMqttNetLogger logger)
+        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            switch (options)
+            var serializer = new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion };
+
+            switch (options.ChannelOptions)
             {
                 case MqttClientTcpOptions tcpOptions:
-                    return new MqttChannelAdapter(new MqttTcpChannel(tcpOptions), new MqttPacketSerializer(), logger);
+                    {
+                        return new MqttChannelAdapter(new MqttTcpChannel(tcpOptions), serializer, logger);
+                    }
+
                 case MqttClientWebSocketOptions webSocketOptions:
-                    return new MqttChannelAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializer(), logger);
+                    {
+                        return new MqttChannelAdapter(new MqttWebSocketChannel(webSocketOptions), serializer, logger);
+                    }
+
                 default:
-                    throw new NotSupportedException();
+                    {
+                        throw new NotSupportedException();
+                    }
             }
         }
     }

Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.Uwp.cs

@@ -13,14 +13,21 @@ using MQTTnet.Client;
 
 namespace MQTTnet.Implementations
 {
-    public sealed class MqttTcpChannel : IMqttChannel, IDisposable
+    public sealed class MqttTcpChannel : IMqttChannel
     {
+        // ReSharper disable once MemberCanBePrivate.Global
+        // ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
+        public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user.
+
+        private readonly int _bufferSize = BufferSize;
         private readonly MqttClientTcpOptions _options;
+
         private StreamSocket _socket;
 
         public MqttTcpChannel(MqttClientTcpOptions options)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            _bufferSize = _options.BufferSize;
         }
 
         public MqttTcpChannel(StreamSocket socket)
@@ -69,20 +76,56 @@ namespace MQTTnet.Implementations
 
         public void Dispose()
         {
-            SendStream?.Dispose();
-            SendStream = null;
+            try
+            {
+                SendStream?.Dispose();
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            catch (NullReferenceException)
+            {
+            }
+            finally
+            {
+                SendStream = null;
+            }
 
-            ReceiveStream?.Dispose();
-            ReceiveStream = null;
+            try
+            {
+                ReceiveStream?.Dispose();
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            catch (NullReferenceException)
+            {
+            }
+            finally
+            {
+                ReceiveStream = null;
+            }
 
-            _socket?.Dispose();
-            _socket = null;
+            try
+            {
+                _socket?.Dispose();
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            catch (NullReferenceException)
+            {
+            }
+            finally
+            {
+                _socket = null;
+            }
         }
 
         private void CreateStreams()
         {
-            SendStream = _socket.OutputStream.AsStreamForWrite();
-            ReceiveStream = _socket.InputStream.AsStreamForRead();
+            SendStream = _socket.OutputStream.AsStreamForWrite(_bufferSize);
+            ReceiveStream = _socket.InputStream.AsStreamForRead(_bufferSize);
         }
 
         private static Certificate LoadCertificate(MqttClientTcpOptions options)

Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs

@@ -12,17 +12,20 @@ using MQTTnet.Client;
 
 namespace MQTTnet.Implementations
 {
-    public sealed class MqttTcpChannel : IMqttChannel, IDisposable
+    public sealed class MqttTcpChannel : IMqttChannel
     {
-        private readonly MqttClientTcpOptions _options;
-
-        //todo: this can be used with min dependency NetStandard1.6
 #if NET452 || NET461
         // ReSharper disable once MemberCanBePrivate.Global
         // ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
         public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user.
+
+        private readonly int _bufferSize = BufferSize;
+#else
+        private readonly int _bufferSize = 0;
 #endif
 
+        private readonly MqttClientTcpOptions _options;
+
         private Socket _socket;
         private SslStream _sslStream;
 
@@ -32,6 +35,7 @@ namespace MQTTnet.Implementations
         public MqttTcpChannel(MqttClientTcpOptions options)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            _bufferSize = options.BufferSize;
         }
 
         /// <summary>
@@ -43,7 +47,7 @@ namespace MQTTnet.Implementations
             _socket = socket ?? throw new ArgumentNullException(nameof(socket));
             _sslStream = sslStream;
 
-            CreateStreams(socket, sslStream);
+            CreateStreams();
         }
 
         public Stream SendStream { get; private set; }
@@ -58,7 +62,6 @@ namespace MQTTnet.Implementations
                 _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
             }
 
-            //todo: else brach can be used with min dependency NET46
 #if NET452 || NET461
             await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
 #else
@@ -68,10 +71,10 @@ namespace MQTTnet.Implementations
             if (_options.TlsOptions.UseTls)
             {
                 _sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
-                await _sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(_options), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+                await _sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
             }
             
-            CreateStreams(_socket, _sslStream);
+            CreateStreams();
         }
 
         public Task DisconnectAsync()
@@ -82,11 +85,70 @@ namespace MQTTnet.Implementations
 
         public void Dispose()
         {
-            _socket?.Dispose();
-            _socket = null;
+            var oneStreamIsUsed = SendStream != null && ReceiveStream != null && ReferenceEquals(SendStream, ReceiveStream);
 
-            _sslStream?.Dispose();
-            _sslStream = null;
+            try
+            {
+                SendStream?.Dispose();
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            catch (NullReferenceException)
+            {
+            }
+            finally
+            {
+                SendStream = null;
+            }
+
+            try
+            {
+                if (!oneStreamIsUsed)
+                {
+                    ReceiveStream?.Dispose();
+                }
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            catch (NullReferenceException)
+            {
+            }
+            finally
+            {
+                ReceiveStream = null;
+            }
+
+            try
+            {
+                _sslStream?.Dispose();
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            catch (NullReferenceException)
+            {
+            }
+            finally
+            {
+                _sslStream = null;
+            }
+
+            try
+            {
+                _socket?.Dispose();
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            catch (NullReferenceException)
+            {
+            }
+            finally
+            {
+                _socket = null;
+            }
         }
 
         private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
@@ -120,15 +182,15 @@ namespace MQTTnet.Implementations
             return _options.TlsOptions.AllowUntrustedCertificates;
         }
 
-        private static X509CertificateCollection LoadCertificates(MqttClientTcpOptions options)
+        private X509CertificateCollection LoadCertificates()
         {
             var certificates = new X509CertificateCollection();
-            if (options.TlsOptions.Certificates == null)
+            if (_options.TlsOptions.Certificates == null)
             {
                 return certificates;
             }
 
-            foreach (var certificate in options.TlsOptions.Certificates)
+            foreach (var certificate in _options.TlsOptions.Certificates)
             {
                 certificates.Add(new X509Certificate2(certificate));
             }
@@ -136,20 +198,26 @@ namespace MQTTnet.Implementations
             return certificates;
         }
 
-        private void CreateStreams(Socket socket, Stream sslStream)
+        private void CreateStreams()
         {
-            var stream = sslStream ?? new NetworkStream(socket);
+            Stream stream;
+            if (_sslStream != null)
+            {
+                stream = _sslStream;
+            }
+            else
+            {
+                stream = new NetworkStream(_socket, true);
+            }
             
-            //todo: if branch can be used with min dependency NetStandard1.6
 #if NET452 || NET461
-            SendStream = new BufferedStream(stream, BufferSize);
-            ReceiveStream = new BufferedStream(stream, BufferSize);
+            SendStream = new BufferedStream(stream, _bufferSize);
+            ReceiveStream = new BufferedStream(stream, _bufferSize);
 #else
             SendStream = stream;
             ReceiveStream = stream;
 #endif
         }
-
     }
 }
 #endif

Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.Uwp.cs

@@ -9,12 +9,12 @@ using MQTTnet.Server;
 
 namespace MQTTnet.Implementations
 {
-    public class MqttServerAdapter : IMqttServerAdapter, IDisposable
+    public class MqttTcpServerAdapter : IMqttServerAdapter, IDisposable
     {
         private readonly IMqttNetLogger _logger;
         private StreamSocketListener _defaultEndpointSocket;
 
-        public MqttServerAdapter(IMqttNetLogger logger)
+        public MqttTcpServerAdapter(IMqttNetLogger logger)
         {
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
@@ -67,7 +67,7 @@ namespace MQTTnet.Implementations
             }
             catch (Exception exception)
             {
-                _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
+                _logger.Error<MqttTcpServerAdapter>(exception, "Error while accepting connection at default endpoint.");
             }
         }
     }

Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpServerAdapter.cs

@@ -14,7 +14,7 @@ using MQTTnet.Server;
 
 namespace MQTTnet.Implementations
 {
-    public class MqttServerAdapter : IMqttServerAdapter, IDisposable
+    public class MqttTcpServerAdapter : IMqttServerAdapter, IDisposable
     {
         private readonly IMqttNetLogger _logger;
 
@@ -23,7 +23,7 @@ namespace MQTTnet.Implementations
         private Socket _tlsEndpointSocket;
         private X509Certificate2 _tlsCertificate;
 
-        public MqttServerAdapter(IMqttNetLogger logger)
+        public MqttTcpServerAdapter(IMqttNetLogger logger)
         {
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
@@ -117,7 +117,7 @@ namespace MQTTnet.Implementations
                         return;
                     }
 
-                    _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
+                    _logger.Error<MqttTcpServerAdapter>(exception, "Error while accepting connection at default endpoint.");
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }
             }
@@ -152,7 +152,7 @@ namespace MQTTnet.Implementations
                         return;
                     }
 
-                    _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at TLS endpoint.");
+                    _logger.Error<MqttTcpServerAdapter>(exception, "Error while accepting connection at TLS endpoint.");
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }
             }

Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs

@@ -9,7 +9,7 @@ using MQTTnet.Client;
 
 namespace MQTTnet.Implementations
 {
-    public sealed class MqttWebSocketChannel : IMqttChannel, IDisposable
+    public sealed class MqttWebSocketChannel : IMqttChannel
     {
         // ReSharper disable once MemberCanBePrivate.Global
         // ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
@@ -96,8 +96,17 @@ namespace MQTTnet.Implementations
 
         public void Dispose()
         {
-            _webSocket?.Dispose();
-            _webSocket = null;
+            try
+            {
+                _webSocket?.Dispose();
+            }
+            catch (ObjectDisposedException)
+            {
+            }
+            finally
+            {
+                _webSocket = null;
+            }            
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs

@@ -11,9 +11,9 @@ namespace MQTTnet.Implementations
 {
     public class WebSocketStream : Stream
     {
-        private readonly WebSocket _webSocket;
-        private readonly byte[] _chunkBuffer = new byte[MqttWebSocketChannel.BufferSize];
+        private readonly byte[] _chunckBuffer = new byte[MqttWebSocketChannel.BufferSize];
         private readonly Queue<byte> _buffer = new Queue<byte>(MqttWebSocketChannel.BufferSize);
+        private readonly WebSocket _webSocket;
 
         public WebSocketStream(WebSocket webSocket)
         {
@@ -55,25 +55,28 @@ namespace MQTTnet.Implementations
             // Use existing date from buffer.
             while (count > 0 && _buffer.Any())
             {
-                buffer[offset++] = _buffer.Dequeue();
+                buffer[offset] = _buffer.Dequeue();
                 count--;
                 bytesRead++;
+                offset++;
             }
 
             if (count == 0)
             {
                 return bytesRead;
             }
-            
+
+            // Fetch new data if the buffer is not full.
             while (_webSocket.State == WebSocketState.Open)
             {
-                await FetchChunkAsync(cancellationToken);
+                await FetchChunkAsync(cancellationToken).ConfigureAwait(false);
 
                 while (count > 0 && _buffer.Any())
                 {
-                    buffer[offset++] = _buffer.Dequeue();
+                    buffer[offset] = _buffer.Dequeue();
                     count--;
                     bytesRead++;
+                    offset++;
                 }
 
                 if (count == 0)
@@ -111,19 +114,24 @@ namespace MQTTnet.Implementations
         }
 
         private async Task FetchChunkAsync(CancellationToken cancellationToken)
-        {
-            var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(_chunkBuffer, 0, _chunkBuffer.Length), cancellationToken).ConfigureAwait(false);
-
-            for (var i = 0; i < response.Count; i++)
-            {
-                var @byte = _chunkBuffer[i];
-                _buffer.Enqueue(@byte);
-            }
+        {   
+            var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(_chunckBuffer, 0, _chunckBuffer.Length), cancellationToken).ConfigureAwait(false);
 
             if (response.MessageType == WebSocketMessageType.Close)
             {
                 await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
             }
+            else if (response.MessageType == WebSocketMessageType.Binary)
+            {
+                for (var i = 0; i < response.Count; i++)
+                {
+                    _buffer.Enqueue(_chunckBuffer[i]);
+                }
+            }
+            else if (response.MessageType == WebSocketMessageType.Text)
+            {
+                throw new MqttProtocolViolationException("WebSocket channel received TEXT message.");
+            }
         }
     }
 }

Frameworks/MQTTnet.NetStandard/ManagedClient/IManagedMqttClient.cs

@@ -7,6 +7,7 @@ namespace MQTTnet.ManagedClient
 {
     public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
     {
+        bool IsStarted { get; }
         bool IsConnected { get; }
 
         event EventHandler<MqttClientConnectedEventArgs> Connected;

Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClient.cs

@@ -40,6 +40,7 @@ namespace MQTTnet.ManagedClient
         }
 
         public bool IsConnected => _mqttClient.IsConnected;
+        public bool IsStarted => _connectionCancellationToken != null;
 
         public event EventHandler<MqttClientConnectedEventArgs> Connected;
         public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
@@ -243,7 +244,7 @@ namespace MQTTnet.ManagedClient
             }
             finally
             {
-                _logger.Trace<ManagedMqttClient>("Stopped publishing messages.");
+                _logger.Verbose<ManagedMqttClient>("Stopped publishing messages.");
             }
         }
 

Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientExtensions.cs

@@ -21,6 +21,14 @@ namespace MQTTnet.ManagedClient
             return managedClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
         }
 
+        public static Task SubscribeAsync(this IManagedMqttClient managedClient, string topic)
+        {
+            if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
+            if (topic == null) throw new ArgumentNullException(nameof(topic));
+
+            return managedClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());
+        }
+
         public static Task UnsubscribeAsync(this IManagedMqttClient managedClient, params string[] topicFilters)
         {
             if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));

Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClientOptionsBuilder.cs

@@ -6,6 +6,7 @@ namespace MQTTnet.ManagedClient
     public class ManagedMqttClientOptionsBuilder
     {
         private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
+        private MqttClientOptionsBuilder _clientOptionsBuilder;
 
         public ManagedMqttClientOptionsBuilder WithAutoReconnectDelay(TimeSpan value)
         {
@@ -21,7 +22,24 @@ namespace MQTTnet.ManagedClient
 
         public ManagedMqttClientOptionsBuilder WithClientOptions(IMqttClientOptions value)
         {
+            if (_clientOptionsBuilder != null)
+            {
+                throw new InvalidOperationException("Cannot use client options builder and client options at the same time.");
+            }
+
             _options.ClientOptions = value ?? throw new ArgumentNullException(nameof(value));
+
+            return this;
+        }
+
+        public ManagedMqttClientOptionsBuilder WithClientOptions(MqttClientOptionsBuilder builder)
+        {
+            if (_options.ClientOptions != null)
+            {
+                throw new InvalidOperationException("Cannot use client options builder and client options at the same time.");
+            }
+
+            _clientOptionsBuilder = builder;
             return this;
         }
 
@@ -29,15 +47,22 @@ namespace MQTTnet.ManagedClient
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            var builder = new MqttClientOptionsBuilder();
-            options(builder);
-            _options.ClientOptions = builder.Build();
+            if (_clientOptionsBuilder == null)
+            {
+                _clientOptionsBuilder = new MqttClientOptionsBuilder();
+            }
 
+            options(_clientOptionsBuilder);
             return this;
         }
 
         public ManagedMqttClientOptions Build()
         {
+            if (_clientOptionsBuilder != null)
+            {
+                _options.ClientOptions = _clientOptionsBuilder.Build();
+            }
+
             if (_options.ClientOptions == null)
             {
                 throw new InvalidOperationException("The ClientOptions cannot be null.");

Frameworks/MQTTnet.NetStandard/MqttApplicationMessageExtensions.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Text;
+
+namespace MQTTnet
+{
+    public static class MqttApplicationMessageExtensions
+    {
+        public static string ConvertPayloadToString(this MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+
+            if (applicationMessage.Payload == null)
+            {
+                return null;
+            }
+
+            if (applicationMessage.Payload.Length == 0)
+            {
+                return string.Empty;
+            }
+
+            return Encoding.UTF8.GetString(applicationMessage.Payload, 0, applicationMessage.Payload.Length);
+        }
+    }
+}

Frameworks/MQTTnet.NetStandard/MqttFactory.cs

@@ -38,7 +38,14 @@ namespace MQTTnet
         public IMqttServer CreateMqttServer()
         {
             var logger = new MqttNetLogger();
-            return CreateMqttServer(new List<IMqttServerAdapter> { new MqttServerAdapter(logger) }, logger);
+            return CreateMqttServer(logger);
+        }
+
+        public IMqttServer CreateMqttServer(IMqttNetLogger logger)
+        {
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
+
+            return CreateMqttServer(new List<IMqttServerAdapter> { new MqttTcpServerAdapter(logger) }, logger);
         }
 
         public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)

Frameworks/MQTTnet.NetStandard/Packets/IMqttPacketWithIdentifier.cs

@@ -2,6 +2,6 @@
 {
     public interface IMqttPacketWithIdentifier
     {
-        ushort PacketIdentifier { get; set; }
+        ushort? PacketIdentifier { get; set; }
     }
 }

Frameworks/MQTTnet.NetStandard/Packets/MqttBasePublishPacket.cs

@@ -2,6 +2,6 @@
 {
     public class MqttBasePublishPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
-        public ushort PacketIdentifier { get; set; }
+        public ushort? PacketIdentifier { get; set; }
     }
 }

Frameworks/MQTTnet.NetStandard/Packets/MqttPacketExtensions.cs

@@ -1,26 +0,0 @@
-using System;
-
-namespace MQTTnet.Packets
-{
-    public static class MqttPacketExtensions
-    {
-        public static TResponsePacket CreateResponse<TResponsePacket>(this MqttBasePacket packet)
-        {
-            if (packet == null) throw new ArgumentNullException(nameof(packet));
-
-            var responsePacket = Activator.CreateInstance<TResponsePacket>();
-
-            if (responsePacket is IMqttPacketWithIdentifier responsePacketWithIdentifier)
-            {
-                if (!(packet is IMqttPacketWithIdentifier requestPacketWithIdentifier))
-                {
-                    throw new InvalidOperationException("Response packet has PacketIdentifier but request packet does not.");
-                }
-
-                responsePacketWithIdentifier.PacketIdentifier = requestPacketWithIdentifier.PacketIdentifier;
-            }
-
-            return responsePacket;
-        }
-    }
-}

Frameworks/MQTTnet.NetStandard/Packets/MqttSubAckPacket.cs

@@ -6,7 +6,7 @@ namespace MQTTnet.Packets
 {
     public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
-        public ushort PacketIdentifier { get; set; }
+        public ushort? PacketIdentifier { get; set; }
 
         public IList<MqttSubscribeReturnCode> SubscribeReturnCodes { get; } = new List<MqttSubscribeReturnCode>();
 

Frameworks/MQTTnet.NetStandard/Packets/MqttSubscribePacket.cs

@@ -5,7 +5,7 @@ namespace MQTTnet.Packets
 {
     public sealed class MqttSubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
-        public ushort PacketIdentifier { get; set; }
+        public ushort? PacketIdentifier { get; set; }
 
         public IList<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>();
 

Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubAckPacket.cs

@@ -2,7 +2,7 @@
 {
     public sealed class MqttUnsubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
-        public ushort PacketIdentifier { get; set; }
+        public ushort? PacketIdentifier { get; set; }
 
         public override string ToString()
         {

Frameworks/MQTTnet.NetStandard/Packets/MqttUnsubscribe.cs

@@ -4,7 +4,7 @@ namespace MQTTnet.Packets
 {
     public sealed class MqttUnsubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
-        public ushort PacketIdentifier { get; set; }
+        public ushort? PacketIdentifier { get; set; }
 
         public IList<string> TopicFilters { get; set; } = new List<string>();
 

Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs

@@ -24,6 +24,11 @@ namespace MQTTnet.Serializer
 
         public static async Task<MqttPacketHeader> ReadHeaderAsync(Stream stream, CancellationToken cancellationToken)
         {
+            if (cancellationToken.IsCancellationRequested)
+            {
+                return null;
+            }
+
             // Wait for the next package which starts with the header. At this point there will probably
             // some large delay and thus the thread should be put back to the pool (await). So ReadByte()
             // is not an option here.

Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs

@@ -397,7 +397,12 @@ namespace MQTTnet.Serializer
 
         private static byte Serialize(MqttPubRelPacket packet, MqttPacketWriter writer)
         {
-            writer.Write(packet.PacketIdentifier);
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("PubRel packet has no packet identifier.");
+            }
+
+            writer.Write(packet.PacketIdentifier.Value);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
         }
@@ -410,7 +415,12 @@ namespace MQTTnet.Serializer
 
             if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
             {
-                writer.Write(packet.PacketIdentifier);
+                if (!packet.PacketIdentifier.HasValue)
+                {
+                    throw new MqttProtocolViolationException("Publish packet has no packet identifier.");
+                }
+
+                writer.Write(packet.PacketIdentifier.Value);
             }
             else
             {
@@ -444,21 +454,36 @@ namespace MQTTnet.Serializer
 
         private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer)
         {
-            writer.Write(packet.PacketIdentifier);
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("PubAck packet has no packet identifier.");
+            }
+
+            writer.Write(packet.PacketIdentifier.Value);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
         }
 
         private static byte Serialize(MqttPubRecPacket packet, MqttPacketWriter writer)
         {
-            writer.Write(packet.PacketIdentifier);
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("PubRec packet has no packet identifier.");
+            }
+
+            writer.Write(packet.PacketIdentifier.Value);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
         }
 
         private static byte Serialize(MqttPubCompPacket packet, MqttPacketWriter writer)
         {
-            writer.Write(packet.PacketIdentifier);
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("PubComp packet has no packet identifier.");
+            }
+
+            writer.Write(packet.PacketIdentifier.Value);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
         }
@@ -467,7 +492,12 @@ namespace MQTTnet.Serializer
         {
             if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
 
-            writer.Write(packet.PacketIdentifier);
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("Subscribe packet has no packet identifier.");
+            }
+
+            writer.Write(packet.PacketIdentifier.Value);
 
             if (packet.TopicFilters?.Count > 0)
             {
@@ -483,7 +513,12 @@ namespace MQTTnet.Serializer
 
         private static byte Serialize(MqttSubAckPacket packet, MqttPacketWriter writer)
         {
-            writer.Write(packet.PacketIdentifier);
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("SubAck packet has no packet identifier.");
+            }
+
+            writer.Write(packet.PacketIdentifier.Value);
 
             if (packet.SubscribeReturnCodes?.Any() == true)
             {
@@ -500,7 +535,12 @@ namespace MQTTnet.Serializer
         {
             if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
 
-            writer.Write(packet.PacketIdentifier);
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("Unsubscribe packet has no packet identifier.");
+            }
+
+            writer.Write(packet.PacketIdentifier.Value);
 
             if (packet.TopicFilters?.Any() == true)
             {
@@ -513,9 +553,14 @@ namespace MQTTnet.Serializer
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
         }
 
-        private static byte Serialize(IMqttPacketWithIdentifier packet, BinaryWriter writer)
+        private static byte Serialize(MqttUnsubAckPacket packet, BinaryWriter writer)
         {
-            writer.Write(packet.PacketIdentifier);
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("UnsubAck packet has no packet identifier.");
+            }
+
+            writer.Write(packet.PacketIdentifier.Value);
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck);
         }
 

Frameworks/MQTTnet.NetStandard/Server/MqttApplicationMessageInterceptorContext.cs

@@ -11,5 +11,9 @@
         public string ClientId { get; }
 
         public MqttApplicationMessage ApplicationMessage { get; set; }
+
+        public bool AcceptPublish { get; set; } = true;
+
+        public bool CloseConnection { get; set; }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectedEventArgs.cs

@@ -4,11 +4,14 @@ namespace MQTTnet.Server
 {
     public class MqttClientDisconnectedEventArgs : EventArgs
     {
-        public MqttClientDisconnectedEventArgs(ConnectedMqttClient client)
+        public MqttClientDisconnectedEventArgs(ConnectedMqttClient client, bool wasCleanDisconnect)
         {
             Client = client ?? throw new ArgumentNullException(nameof(client));
+            WasCleanDisconnect = wasCleanDisconnect;
         }
         
         public ConnectedMqttClient Client { get; }
+
+        public bool WasCleanDisconnect { get; }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs

@@ -16,6 +16,8 @@ namespace MQTTnet.Server
         private readonly Func<Task> _timeoutCallback;
         private readonly IMqttNetLogger _logger;
 
+        private Task _workerTask;
+
         public MqttClientKeepAliveMonitor(string clientId, Func<Task> timeoutCallback, IMqttNetLogger logger)
         {
             _clientId = clientId;
@@ -34,7 +36,15 @@ namespace MQTTnet.Server
                 return;
             }
 
-            Task.Run(async () => await RunAsync(keepAlivePeriod, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
+            _workerTask = Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken).ConfigureAwait(false), cancellationToken);
+        }
+
+        public void WaitForCompletion()
+        {
+            if (_workerTask != null)
+            {
+                Task.WaitAll(_workerTask);
+            }
         }
 
         private async Task RunAsync(int keepAlivePeriod, CancellationToken cancellationToken)
@@ -71,7 +81,7 @@ namespace MQTTnet.Server
             }
             finally
             {
-                _logger.Trace<MqttClientSession>("Client {0}: Stopped checking keep alive timeout.", _clientId);
+                _logger.Verbose<MqttClientSession>("Client {0}: Stopped checking keep alive timeout.", _clientId);
             }
         }
 

Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Concurrent;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
@@ -18,6 +19,8 @@ namespace MQTTnet.Server
         private readonly MqttClientSession _clientSession;
         private readonly IMqttNetLogger _logger;
 
+        private Task _workerTask;
+
         public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetLogger logger)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -36,7 +39,15 @@ namespace MQTTnet.Server
                 return;
             }
 
-            Task.Run(async () => await SendQueuedPacketsAsync(adapter, cancellationToken), cancellationToken).ConfigureAwait(false);
+            _workerTask = Task.Run(() => SendQueuedPacketsAsync(adapter, cancellationToken), cancellationToken);
+        }
+
+        public void WaitForCompletion()
+        {
+            if (_workerTask != null)
+            {
+                Task.WaitAll(_workerTask);
+            }
         }
 
         public void Enqueue(MqttBasePacket packet)
@@ -46,7 +57,7 @@ namespace MQTTnet.Server
             _queue.Enqueue(packet);
             _queueWaitSemaphore.Release();
 
-            _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
+            _logger.Verbose<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
         }
 
         private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -55,7 +66,7 @@ namespace MQTTnet.Server
             {
                 while (!cancellationToken.IsCancellationRequested)
                 {
-                    await SendQueuedPacketAsync(adapter, cancellationToken);
+                    await SendNextQueuedPacketAsync(adapter, cancellationToken).ConfigureAwait(false);
                 }
             }
             catch (OperationCanceledException)
@@ -67,7 +78,7 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task SendQueuedPacketAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
+        private async Task SendNextQueuedPacketAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
         {
             MqttBasePacket packet = null;
             try
@@ -78,9 +89,14 @@ namespace MQTTnet.Server
                     throw new InvalidOperationException(); // should not happen
                 }
 
-                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return;
+                }
 
-                _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false);
+
+                _logger.Verbose<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
             }
             catch (Exception exception)
             {
@@ -110,7 +126,10 @@ namespace MQTTnet.Server
                     }
                 }
 
-                await _clientSession.StopAsync();
+                if (!cancellationToken.IsCancellationRequested)
+                {
+                    await _clientSession.StopAsync().ConfigureAwait(false);
+                }
             }
         }
 

Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs

@@ -3,6 +3,7 @@ using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
+using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Internal;
@@ -14,6 +15,7 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClientSession : IDisposable
     {
+        private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         private readonly IMqttServerOptions _options;
         private readonly IMqttNetLogger _logger;
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
@@ -21,6 +23,7 @@ namespace MQTTnet.Server
         private IMqttChannelAdapter _adapter;
         private CancellationTokenSource _cancellationTokenSource;
         private MqttApplicationMessage _willMessage;
+        private bool _wasCleanDisconnect;
 
         public MqttClientSession(
             string clientId,
@@ -53,7 +56,7 @@ namespace MQTTnet.Server
 
         public bool IsConnected => _adapter != null;
 
-        public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
+        public async Task<bool> RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
         {
             if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket));
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -62,6 +65,7 @@ namespace MQTTnet.Server
             {
                 var cancellationTokenSource = new CancellationTokenSource();
 
+                _wasCleanDisconnect = false;
                 _willMessage = connectPacket.WillMessage;
                 _adapter = adapter;
                 _cancellationTokenSource = cancellationTokenSource;
@@ -82,9 +86,11 @@ namespace MQTTnet.Server
             {
                 _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
             }
+
+            return _wasCleanDisconnect;
         }
 
-        public async Task StopAsync()
+        public async Task StopAsync(bool wasCleanDisconnect = false)
         {
             try
             {
@@ -93,24 +99,27 @@ namespace MQTTnet.Server
                     return;
                 }
 
+                _wasCleanDisconnect = wasCleanDisconnect;
+
                 _cancellationTokenSource?.Cancel(false);
+
+                PendingMessagesQueue.WaitForCompletion();
+                KeepAliveMonitor.WaitForCompletion();
+
                 _cancellationTokenSource?.Dispose();
                 _cancellationTokenSource = null;
 
-                if (_adapter != null)
-                {
-                    await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
-                    _adapter = null;
-                }
+                _adapter = null;
 
                 _logger.Info<MqttClientSession>("Client '{0}': Session stopped.", ClientId);
             }
             finally
             {
                 var willMessage = _willMessage;
-                if (willMessage != null)
+                _willMessage = null; // clear willmessage so it is send just once
+
+                if (willMessage != null && !wasCleanDisconnect)
                 {
-                    _willMessage = null; // clear willmessage so it is send just once
                     await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
                 }
             }
@@ -129,6 +138,11 @@ namespace MQTTnet.Server
             var publishPacket = applicationMessage.ToPublishPacket();
             publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
 
+            if (publishPacket.QualityOfServiceLevel > 0)
+            {
+                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
+            }
+
             PendingMessagesQueue.Enqueue(publishPacket);
         }
 
@@ -205,7 +219,7 @@ namespace MQTTnet.Server
 
             if (packet is MqttPingReqPacket)
             {
-                return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPingRespPacket());
+                return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttPingRespPacket() });
             }
 
             if (packet is MqttPubRelPacket pubRelPacket)
@@ -215,7 +229,12 @@ namespace MQTTnet.Server
 
             if (packet is MqttPubRecPacket pubRecPacket)
             {
-                return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, pubRecPacket.CreateResponse<MqttPubRelPacket>());
+                var responsePacket = new MqttPubRelPacket
+                {
+                    PacketIdentifier = pubRecPacket.PacketIdentifier
+                };
+
+                return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { responsePacket });
             }
 
             if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
@@ -234,7 +253,12 @@ namespace MQTTnet.Server
                 return HandleIncomingUnsubscribePacketAsync(adapter, unsubscribePacket, cancellationToken);
             }
 
-            if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
+            if (packet is MqttDisconnectPacket)
+            {
+                return StopAsync(true);
+            }
+
+            if (packet is MqttConnectPacket)
             {
                 return StopAsync();
             }
@@ -246,11 +270,10 @@ namespace MQTTnet.Server
         private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
         {
             var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { subscribeResult.ResponsePacket }).ConfigureAwait(false);
 
             if (subscribeResult.CloseConnection)
             {
-                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false);
                 await StopAsync().ConfigureAwait(false);
             }
 
@@ -260,7 +283,7 @@ namespace MQTTnet.Server
         private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
         {
             var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false);
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, unsubscribeResult);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { unsubscribeResult });
         }
 
         private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
@@ -302,7 +325,7 @@ namespace MQTTnet.Server
             await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
 
             var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier };
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false);
         }
 
         private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
@@ -311,13 +334,13 @@ namespace MQTTnet.Server
             await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
 
             var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier };
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false);
         }
 
         private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
         {
             var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier };
-            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response);
+            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response });
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs

@@ -29,20 +29,24 @@ namespace MQTTnet.Server
         }
 
         public Action<ConnectedMqttClient> ClientConnectedCallback { get; set; }
-        public Action<ConnectedMqttClient> ClientDisconnectedCallback { get; set; }
+        public Action<ConnectedMqttClient, bool> ClientDisconnectedCallback { get; set; }
         public Action<string, TopicFilter> ClientSubscribedTopicCallback { get; set; }
         public Action<string, string> ClientUnsubscribedTopicCallback { get; set; }
         public Action<string, MqttApplicationMessage> ApplicationMessageReceivedCallback { get; set; }
-        
+
         public async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
         {
             var clientId = string.Empty;
+            var wasCleanDisconnect = false;
             MqttClientSession clientSession = null;
+            
             try
             {
-                if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false) is MqttConnectPacket connectPacket))
+                if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken)
+                    .ConfigureAwait(false) is MqttConnectPacket connectPacket))
                 {
-                    throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
+                    throw new MqttProtocolViolationException(
+                        "The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
                 }
 
                 clientId = connectPacket.ClientId;
@@ -53,9 +57,12 @@ namespace MQTTnet.Server
                 var connectReturnCode = ValidateConnection(connectPacket);
                 if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
                 {
-                    await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
+                    await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[]
                     {
-                        ConnectReturnCode = connectReturnCode
+                        new MqttConnAckPacket
+                        {
+                            ConnectReturnCode = connectReturnCode
+                        }
                     }).ConfigureAwait(false);
 
                     return;
@@ -64,10 +71,13 @@ namespace MQTTnet.Server
                 var result = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
                 clientSession = result.Session;
 
-                await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
+                await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[]
                 {
-                    ConnectReturnCode = connectReturnCode,
-                    IsSessionPresent = result.IsExistingSession
+                    new MqttConnAckPacket
+                    {
+                        ConnectReturnCode = connectReturnCode,
+                        IsSessionPresent = result.IsExistingSession
+                    }
                 }).ConfigureAwait(false);
 
                 ClientConnectedCallback?.Invoke(new ConnectedMqttClient
@@ -76,7 +86,10 @@ namespace MQTTnet.Server
                     ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
                 });
 
-                await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
+                wasCleanDisconnect = await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
             }
             catch (Exception exception)
             {
@@ -87,18 +100,20 @@ namespace MQTTnet.Server
                 try
                 {
                     await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
+                    clientAdapter.Dispose();
                 }
-                catch (Exception)
+                catch (Exception exception)
                 {
-                    // ignored
+                    _logger.Error<MqttClientSessionsManager>(exception, exception.Message);
                 }
-
+                
                 ClientDisconnectedCallback?.Invoke(new ConnectedMqttClient
                 {
                     ClientId = clientId,
                     ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion,
                     PendingApplicationMessages = clientSession?.PendingMessagesQueue.Count ?? 0
-                });
+                },
+                wasCleanDisconnect);
             }
         }
 
@@ -144,8 +159,13 @@ namespace MQTTnet.Server
         {
             try
             {
-                applicationMessage = InterceptApplicationMessage(senderClientSession, applicationMessage);
-                if (applicationMessage == null)
+                var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage);
+                if (interceptorContext.CloseConnection)
+                {
+                    await senderClientSession.StopAsync().ConfigureAwait(false);
+                }
+
+                if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
                 {
                     return;
                 }
@@ -218,20 +238,20 @@ namespace MQTTnet.Server
             }
         }
 
-        private MqttApplicationMessage InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
         {
-            var interceptor = _options.ApplicationMessageInterceptor;
-            if (interceptor == null)
-            {
-                return applicationMessage;
-            }
-
             var interceptorContext = new MqttApplicationMessageInterceptorContext(
                 senderClientSession?.ClientId,
                 applicationMessage);
 
+            var interceptor = _options.ApplicationMessageInterceptor;
+            if (interceptor == null)
+            {
+                return interceptorContext;
+            }
+            
             interceptor(interceptorContext);
-            return interceptorContext.ApplicationMessage;
+            return interceptorContext;
         }
 
         private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
@@ -267,11 +287,11 @@ namespace MQTTnet.Server
                         clientSession.Dispose();
                         clientSession = null;
 
-                        _logger.Trace<MqttClientSessionsManager>("Stopped existing session of client '{0}'.", connectPacket.ClientId);
+                        _logger.Verbose<MqttClientSessionsManager>("Stopped existing session of client '{0}'.", connectPacket.ClientId);
                     }
                     else
                     {
-                        _logger.Trace<MqttClientSessionsManager>("Reusing existing session of client '{0}'.", connectPacket.ClientId);
+                        _logger.Verbose<MqttClientSessionsManager>("Reusing existing session of client '{0}'.", connectPacket.ClientId);
                     }
                 }
 
@@ -290,7 +310,7 @@ namespace MQTTnet.Server
 
                     _sessions[connectPacket.ClientId] = clientSession;
 
-                    _logger.Trace<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
+                    _logger.Verbose<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
                 }
 
                 return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
@@ -308,7 +328,7 @@ namespace MQTTnet.Server
             ClientSubscribedTopicCallback = null;
             ClientUnsubscribedTopicCallback = null;
             ApplicationMessageReceivedCallback = null;
-            
+
             _semaphore?.Dispose();
         }
     }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs

@@ -14,7 +14,7 @@ namespace MQTTnet.Server
         private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly IMqttServerOptions _options;
         private readonly string _clientId;
-        
+
         public MqttClientSubscriptionsManager(IMqttServerOptions options, string clientId)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -30,7 +30,11 @@ namespace MQTTnet.Server
 
             var result = new MqttClientSubscribeResult
             {
-                ResponsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>(),
+                ResponsePacket = new MqttSubAckPacket
+                {
+                    PacketIdentifier = subscribePacket.PacketIdentifier
+                },
+
                 CloseConnection = false
             };
 
@@ -87,7 +91,10 @@ namespace MQTTnet.Server
                 _semaphore.Release();
             }
 
-            return unsubscribePacket.CreateResponse<MqttUnsubAckPacket>();
+            return new MqttUnsubAckPacket
+            {
+                PacketIdentifier = unsubscribePacket.PacketIdentifier
+            };
         }
 
         public async Task<CheckSubscriptionsResult> CheckSubscriptionsAsync(MqttApplicationMessage applicationMessage)

Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs

@@ -132,7 +132,7 @@ namespace MQTTnet.Server
 
             if (!saveIsRequired)
             {
-                _logger.Trace<MqttRetainedMessagesManager>("Skipped saving retained messages because no changes were detected.");
+                _logger.Verbose<MqttRetainedMessagesManager>("Skipped saving retained messages because no changes were detected.");
             }
 
             if (saveIsRequired && _options.Storage != null)

Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs

@@ -138,10 +138,10 @@ namespace MQTTnet.Server
             ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client));
         }
 
-        private void OnClientDisconnected(ConnectedMqttClient client)
+        private void OnClientDisconnected(ConnectedMqttClient client, bool wasCleanDisconnect)
         {
-            _logger.Info<MqttServer>("Client '{0}': Disconnected.", client.ClientId);
-            ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client));
+            _logger.Info<MqttServer>("Client '{0}': Disconnected (clean={1}).", client.ClientId, wasCleanDisconnect);
+            ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client, wasCleanDisconnect));
         }
 
         private void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
@@ -162,7 +162,7 @@ namespace MQTTnet.Server
         private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
         {
             eventArgs.SessionTask = Task.Run(
-                async () => await _clientSessionsManager.RunSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
+                () => _clientSessionsManager.RunSessionAsync(eventArgs.Client, _cancellationTokenSource.Token),
                 _cancellationTokenSource.Token);
         }
     }

Frameworks/MQTTnet.Netstandard/MQTTnet.NetStandard.csproj

@@ -1,65 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">netstandard1.3;netstandard2.0;net452;net461;uap10.0</TargetFrameworks>
-    <TargetFrameworks Condition=" '$(OS)' != 'Windows_NT' ">netstandard1.3;netstandard2.0</TargetFrameworks>
-    <AssemblyName>MQTTnet</AssemblyName>
-    <RootNamespace>MQTTnet</RootNamespace>
-    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
-    <DebugType>Full</DebugType>
-    <AssemblyVersion>0.0.0.0</AssemblyVersion>
-    <FileVersion>0.0.0.0</FileVersion>
-    <Version>0.0.0.0</Version>
-    <Company />
-    <Product />
-    <Description />
-    <Authors />
-    <PackageId />
-    <SignAssembly>false</SignAssembly>
-    <DelaySign>false</DelaySign>
-  </PropertyGroup>
-
-  <PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
-    <CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
-    <NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
-    <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
-    <TargetPlatformVersion>10.0.16299.0</TargetPlatformVersion>
-    <TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
-    <TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
-    <TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
-    <DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
-    <DefaultLanguage>en</DefaultLanguage>
-    <LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
-  </PropertyGroup>
-
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x86'" />
-
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'" />
-
-  <PropertyGroup Condition="'$(Configuration)|$(TargetFramework)|$(Platform)'=='Release|netstandard1.3|AnyCPU'">
-    <DefineConstants>RELEASE;NETSTANDARD1_3</DefineConstants>
-  </PropertyGroup>
-
-  <ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
-    <PackageReference Include="System.Net.Security" Version="4.3.2" />
-    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
-    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
-  </ItemGroup>
-
-  <ItemGroup Condition="'$(TargetFramework)'=='netstandard1.3'">
-    <PackageReference Include="System.Net.Security" Version="4.3.2" />
-    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
-    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
-  </ItemGroup>
-
-  <ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
-    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="5.4.1" />
-  </ItemGroup>
-
-  <ItemGroup Condition="'$(TargetFramework)'=='net452'">
-  </ItemGroup>
-
-  <ItemGroup Condition="'$(TargetFramework)'=='net461'">
-  </ItemGroup>
-
-</Project>

MQTTnet.sln

@@ -11,13 +11,14 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Frameworks", "Frameworks",
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.TestApp.UniversalWindows", "Tests\MQTTnet.TestApp.UniversalWindows\MQTTnet.TestApp.UniversalWindows.csproj", "{FF1F72D6-9524-4422-9497-3CC0002216ED}"
 EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.NetStandard", "Frameworks\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj", "{3587E506-55A2-4EB3-99C7-DC01E42D25D2}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.NetStandard", "Frameworks\MQTTnet.NetStandard\MQTTnet.Netstandard.csproj", "{3587E506-55A2-4EB3-99C7-DC01E42D25D2}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{67C28AC1-BC3A-420A-BE9C-FA2401431CF9}"
 	ProjectSection(SolutionItems) = preProject
 		appveyor.yml = appveyor.yml
 		Build\build.ps1 = Build\build.ps1
 		Build\MQTTnet.AspNetCore.nuspec = Build\MQTTnet.AspNetCore.nuspec
+		Build\MQTTnet.Extensions.Rpc.nuspec = Build\MQTTnet.Extensions.Rpc.nuspec
 		Build\MQTTnet.nuspec = Build\MQTTnet.nuspec
 	EndProjectSection
 EndProject

README.md

@@ -78,7 +78,7 @@ If you want to contribute to this project just create a pull request. But only p
 This library is used in the following projects:
 
 * MQTT Client Rx (Wrapper for Reactive Extensions, <https://github.com/1iveowl/MQTTClient.rx>)
-* MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester), [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8), and [Windows 10](https://www.microsoft.com/en-us/store/p/mqtt-tester/9wzdncrd272c))
+* MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester) and [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8))
 * Wirehome (Open Source Home Automation system for .NET, <https://github.com/chkr1011/Wirehome>)
 
 

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -75,7 +75,7 @@ namespace MQTTnet.Core.Tests
                 await s.StopAsync();
             }
 
-            Assert.AreEqual(1, receivedMessagesCount);
+            Assert.AreEqual(0, receivedMessagesCount);
         }
 
         [TestMethod]
@@ -156,7 +156,7 @@ namespace MQTTnet.Core.Tests
                 var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build();
                 await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
 
-                s.PublishAsync(message).Wait();
+                await s.PublishAsync(message);
                 await Task.Delay(500);
             }
             finally
@@ -209,12 +209,9 @@ namespace MQTTnet.Core.Tests
                 await s.StartAsync(new MqttServerOptions());
 
                 var c1 = await serverAdapter.ConnectTestClient(s, "c1");
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAndWaitForAsync(s, new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
                 await c1.DisconnectAsync();
 
-                await Task.Delay(TimeSpan.FromSeconds(2));
-                // TODO: Find another way to wait for the retained components.
-
                 var c2 = await serverAdapter.ConnectTestClient(s, "c2");
                 c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
                 await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("retained").Build());
@@ -244,7 +241,7 @@ namespace MQTTnet.Core.Tests
                 await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
                 await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[0]).WithRetainFlag().Build());
                 await c1.DisconnectAsync();
-
+                
                 var c2 = await serverAdapter.ConnectTestClient(s, "c2");
                 c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
                 await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce));
@@ -274,7 +271,8 @@ namespace MQTTnet.Core.Tests
                 await s.StartAsync(options);
 
                 var c1 = await serverAdapter.ConnectTestClient(s, "c1");
-                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+
+                await c1.PublishAndWaitForAsync(s, new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
                 await c1.DisconnectAsync();
             }
             finally
@@ -282,8 +280,7 @@ namespace MQTTnet.Core.Tests
                 await s.StopAsync();
             }
 
-            await Task.Delay(TimeSpan.FromSeconds(2));
-            // TODO: Find another way to wait for the retained components.
+            Assert.AreEqual(1, storage.Messages.Count);
 
             s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
 
@@ -385,17 +382,17 @@ namespace MQTTnet.Core.Tests
 
         private class TestStorage : IMqttServerStorage
         {
-            private IList<MqttApplicationMessage> _messages = new List<MqttApplicationMessage>();
+            public IList<MqttApplicationMessage> Messages = new List<MqttApplicationMessage>();
 
             public Task SaveRetainedMessagesAsync(IList<MqttApplicationMessage> messages)
             {
-                _messages = messages;
+                Messages = messages;
                 return Task.CompletedTask;
             }
 
             public Task<IList<MqttApplicationMessage>> LoadRetainedMessagesAsync()
             {
-                return Task.FromResult(_messages);
+                return Task.FromResult(Messages);
             }
         }
 

Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs

@@ -17,6 +17,10 @@ namespace MQTTnet.Core.Tests
 
         public IMqttPacketSerializer PacketSerializer { get; } = new MqttPacketSerializer();
 
+        public void Dispose()
+        {
+        }
+
         public Task ConnectAsync(TimeSpan timeout)
         {
             return Task.FromResult(0);
@@ -43,7 +47,17 @@ namespace MQTTnet.Core.Tests
         {
             ThrowIfPartnerIsNull();
 
-            return Task.Run(() => _incomingPackets.Take(), cancellationToken);
+            return Task.Run(() =>
+            {
+                try
+                {
+                    return _incomingPackets.Take(cancellationToken);
+                }
+                catch
+                {
+                    return null;
+                }
+            }, cancellationToken);
         }
 
         private void EnqueuePacketInternal(MqttBasePacket packet)

Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapterFactory.cs

@@ -13,7 +13,7 @@ namespace MQTTnet.Core.Tests
             _adapter = adapter;
         }
         
-        public IMqttChannelAdapter CreateClientAdapter(IMqttClientChannelOptions options, IMqttNetLogger logger)
+        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
         {
             return _adapter;
         }

Tests/MQTTnet.Core.Tests/TestServerExtensions.cs

@@ -0,0 +1,38 @@
+using MQTTnet.Client;
+using MQTTnet.Server;
+using System;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Core.Tests
+{
+    public static class TestServerExtensions
+    {
+        /// <summary>
+        /// publishes a message with a client and waits in the server until a message with the same topic is received
+        /// </summary>
+        /// <returns></returns>
+        public static async Task PublishAndWaitForAsync(this IMqttClient client, IMqttServer server, MqttApplicationMessage message) 
+        {
+            var tcs = new TaskCompletionSource<object>();
+
+            EventHandler<MqttApplicationMessageReceivedEventArgs> handler = (sender, args) =>
+            {
+                if (args.ApplicationMessage.Topic == message.Topic)
+                {
+                    tcs.SetResult(true);
+                }
+            };
+            server.ApplicationMessageReceived += handler;
+            
+            try
+            {
+                await client.PublishAsync(message).ConfigureAwait(false);
+                await tcs.Task.ConfigureAwait(false);
+            }
+            finally
+            {
+                server.ApplicationMessageReceived -= handler;
+            }
+        }
+    }
+}

Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs

@@ -5,6 +5,7 @@ using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.FileProviders;
+using MQTTnet;
 using MQTTnet.AspNetCore;
 using MQTTnet.Server;
 

Tests/MQTTnet.TestApp.NetCore/ServerTest.cs

@@ -3,7 +3,6 @@ using System.Text;
 using System.Threading.Tasks;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
-using Newtonsoft.Json.Linq;
 
 namespace MQTTnet.TestApp.NetCore
 {
@@ -38,6 +37,12 @@ namespace MQTTnet.TestApp.NetCore
                             // based payload with the timestamp is a suitable use case.
                             context.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(DateTime.Now.ToString("O"));
                         }
+
+                        if (context.ApplicationMessage.Topic == "not_allowed_topic")
+                        {
+                            context.AcceptPublish = false;
+                            context.CloseConnection = true;
+                        }
                     },
                     SubscriptionInterceptor = context =>
                     {
@@ -72,27 +77,27 @@ namespace MQTTnet.TestApp.NetCore
                         ConsoleColor.Magenta);
                 };
 
-                options.ApplicationMessageInterceptor = c =>
-                {
-                    if (c.ApplicationMessage.Payload == null || c.ApplicationMessage.Payload.Length == 0)
-                    {
-                        return;
-                    }
+                //options.ApplicationMessageInterceptor = c =>
+                //{
+                //    if (c.ApplicationMessage.Payload == null || c.ApplicationMessage.Payload.Length == 0)
+                //    {
+                //        return;
+                //    }
 
-                    try
-                    {
-                        var content = JObject.Parse(Encoding.UTF8.GetString(c.ApplicationMessage.Payload));
-                        var timestampProperty = content.Property("timestamp");
-                        if (timestampProperty != null && timestampProperty.Value.Type == JTokenType.Null)
-                        {
-                            timestampProperty.Value = DateTime.Now.ToString("O");
-                            c.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(content.ToString());
-                        }
-                    }
-                    catch (Exception e)
-                    {
-                    }
-                };
+                //    try
+                //    {
+                //        var content = JObject.Parse(Encoding.UTF8.GetString(c.ApplicationMessage.Payload));
+                //        var timestampProperty = content.Property("timestamp");
+                //        if (timestampProperty != null && timestampProperty.Value.Type == JTokenType.Null)
+                //        {
+                //            timestampProperty.Value = DateTime.Now.ToString("O");
+                //            c.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(content.ToString());
+                //        }
+                //    }
+                //    catch (Exception)
+                //    {
+                //    }
+                //};
 
                 mqttServer.ClientDisconnected += (s, e) =>
                 {

Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj

@@ -12,7 +12,7 @@
     <DefaultLanguage>en-US</DefaultLanguage>
     <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
     <TargetPlatformVersion Condition=" '$(TargetPlatformVersion)' == '' ">10.0.16299.0</TargetPlatformVersion>
-    <TargetPlatformMinVersion>10.0.15063.0</TargetPlatformMinVersion>
+    <TargetPlatformMinVersion>10.0.16299.0</TargetPlatformMinVersion>
     <MinimumVisualStudioVersion>14</MinimumVisualStudioVersion>
     <FileAlignment>512</FileAlignment>
     <ProjectTypeGuids>{A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>

Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml

@@ -148,7 +148,7 @@
             </PivotItem>
             <PivotItem Header="Log">
                 <StackPanel Background="{ThemeResource ApplicationPageBackgroundThemeBrush}">
-                    <Button Click="Clear" Width="120">Clear</Button>
+                    <Button Click="ClearLog" Width="120">Clear</Button>
                 </StackPanel>
             </PivotItem>
         </Pivot>

Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs

@@ -13,6 +13,8 @@ using MQTTnet.Implementations;
 using MQTTnet.ManagedClient;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
+using MqttClientConnectedEventArgs = MQTTnet.Client.MqttClientConnectedEventArgs;
+using MqttClientDisconnectedEventArgs = MQTTnet.Client.MqttClientDisconnectedEventArgs;
 
 namespace MQTTnet.TestApp.UniversalWindows
 {
@@ -33,6 +35,11 @@ namespace MQTTnet.TestApp.UniversalWindows
         private async void OnTraceMessagePublished(object sender, MqttNetLogMessagePublishedEventArgs e)
         {
             _traceMessages.Enqueue(e.TraceMessage);
+            await UpdateLogAsync();
+        }
+
+        private async Task UpdateLogAsync()
+        {
             while (_traceMessages.Count > 100)
             {
                 _traceMessages.TryDequeue(out _);
@@ -42,7 +49,8 @@ namespace MQTTnet.TestApp.UniversalWindows
             foreach (var traceMessage in _traceMessages)
             {
                 logText.AppendFormat(
-                    "[{0:yyyy-MM-dd HH:mm:ss.fff}] [{1}] [{2}] [{3}] [{4}]{5}", traceMessage.Timestamp,
+                    "[{0:yyyy-MM-dd HH:mm:ss.fff}] [{1}] [{2}] [{3}] [{4}]{5}",
+                    traceMessage.Timestamp,
                     traceMessage.Level,
                     traceMessage.Source,
                     traceMessage.ThreadId,
@@ -112,11 +120,15 @@ namespace MQTTnet.TestApp.UniversalWindows
                 {
                     await _mqttClient.DisconnectAsync();
                     _mqttClient.ApplicationMessageReceived -= OnApplicationMessageReceived;
+                    _mqttClient.Connected -= OnConnected;
+                    _mqttClient.Disconnected -= OnDisconnected;
                 }
 
                 var factory = new MqttFactory();
                 _mqttClient = factory.CreateMqttClient();
                 _mqttClient.ApplicationMessageReceived += OnApplicationMessageReceived;
+                _mqttClient.Connected += OnConnected;
+                _mqttClient.Disconnected += OnDisconnected;
 
                 await _mqttClient.ConnectAsync(options);
             }
@@ -126,6 +138,22 @@ namespace MQTTnet.TestApp.UniversalWindows
             }
         }
 
+        private void OnDisconnected(object sender, MqttClientDisconnectedEventArgs e)
+        {
+            _traceMessages.Enqueue(new MqttNetLogMessage("", DateTime.Now, -1,
+                "", MqttNetLogLevel.Info, "! DISCONNECTED EVENT FIRED", null));
+
+            Task.Run(UpdateLogAsync);
+        }
+
+        private void OnConnected(object sender, MqttClientConnectedEventArgs e)
+        {
+            _traceMessages.Enqueue(new MqttNetLogMessage("", DateTime.Now, -1,
+                "", MqttNetLogLevel.Info, "! CONNECTED EVENT FIRED", null));
+
+            Task.Run(UpdateLogAsync);
+        }
+
         private async void OnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs eventArgs)
         {
             var item = $"Timestamp: {DateTime.Now:O} | Topic: {eventArgs.ApplicationMessage.Topic} | Payload: {Encoding.UTF8.GetString(eventArgs.ApplicationMessage.Payload)} | QoS: {eventArgs.ApplicationMessage.QualityOfServiceLevel}";
@@ -197,8 +225,13 @@ namespace MQTTnet.TestApp.UniversalWindows
             }
         }
 
-        private void Clear(object sender, RoutedEventArgs e)
+        private void ClearLog(object sender, RoutedEventArgs e)
         {
+            while (_traceMessages.Count > 0)
+            {
+                _traceMessages.TryDequeue(out _);
+            }
+
             Trace.Text = string.Empty;
         }