Merge branch 'develop'

Fix ManagedMqttClient missing call to unsubscribe topics

Build/MQTTnet.AspNetCore.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet.AspNetCore</id>
-    <version>2.6.0</version>
+    <version>2.7.1</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.6.0.
+    <releaseNotes>* Updated to MQTTnet 2.7.1.
     </releaseNotes>
-    <copyright>Copyright Christian Kratky 2016-2017</copyright>
+    <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.6.0" />
+        <dependency id="MQTTnet" version="2.7.1" />
       </group>
     </dependencies>
   </metadata>

Build/MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.6.0</version>
+    <version>2.7.2</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,28 +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] Merged projects (BREAKING CHANGE! But only namespace changes).
-* [Core] Added a strong name for the assembly.
-* [Core] Performance optimizations.
-* [Core] Fixed a logging issue when dealing with IOExceptions.
-* [Core] Fixed a typo in the global logger class (BREAKING CHANGE! Please find new example in Wiki).
-* [Core] Added support for project compilation under macOS (Thanks to @FourOnes).
-* [Client] The certificate for encryption is now loaded as a _X509Certificate2_ which prodives more data (Thanks to @cvellan).
-* [Client] Fixed an issue in _ManagedClient_ which can cause the client to stop when publishing subscriptions.
-* [Client] Fixed an issue in _ManagedClient_ which prevents changing the QoS of an existing subscription (BREAKING CHANGE!).
-* [Client] Fixed an issue in _MqttClientOptionsBuilder_ which prevents adding TLS options to the client options when calling _Build()_. (Thanks to @cvellan).
-* [Client] Added an overload for subscribing at both clients which reduces required code.
-* [Client] Added the _ClientId_ to the application message interceptor context.
-* [Server] The application message interceptor can now delete any received application message.
-* [Server] Added a ConnectionValidator context to align with other APIs (BREAKING CHANGE! Please find new example in Wiki).
-* [Server] Added an interface for the _MqttServerOptions_.
-* [Server] Added packet statistics for the connected clients.
-* [Server] Fixed a security issue which sends retained packages to a failed subscription.
-* [Server] Fixed the response (MaximumQoS) of a subscription (Thanks to @redbeans2017).
-* [Server] The keep alive timeouts are now checked for every client (Thanks to @RainerMueller82).
-* [Server] The used IP addresses can be specified now (IPAddress.Any as default).
+    <releaseNotes>* [Client] Added the subprotocol "mqtt" as default for web socket based connections.
+* [Client] Added a new client setting called "KeepAliveSendInterval". It allows configuring the effective interval for sending ping requests.
+* [Client] The client will no longer send ping requests if other packets are sent within the configured interval.
+* [Server] The server now generates a valid packet identifier when disaptching publish packets to clients.
+* [Core] Add several new extension methods.
     </releaseNotes>
-    <copyright>Copyright Christian Kratky 2016-2017</copyright>
+    <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>
 
@@ -75,7 +60,6 @@
 
     <!-- .NET Framework -->
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net452\MQTTnet.*" target="lib\net452\"/>
-
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net461\MQTTnet.*" target="lib\net461\"/>
     
   </files>

Build/build.ps1

@@ -1,6 +1,7 @@
 param([string]$assemblyVersion, [string]$nugetVersion)
 
-if ([string]::IsNullOrEmpty($version)) {$version = "0.0.1"}
+if ([string]::IsNullOrEmpty($assemblyVersion)) {$assemblyVersion = "0.0.1"}
+if ([string]::IsNullOrEmpty($nugetVersion)) {$nugetVersion = "0.0.1"}
 
 $vswhere = ${Env:\ProgramFiles(x86)} + '\Microsoft Visual Studio\Installer\vswhere'
 
@@ -9,14 +10,14 @@ if ($path) {
     $msbuild = join-path $path 'MSBuild\15.0\Bin\MSBuild.exe'
 
 	# Build the core library
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
+    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
     
 	# Build the ASP.NET Core 2.0 extension
-	&$msbuild ..\Frameworks\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
+	&$msbuild ..\Frameworks\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 
 	# Build the RPC extension
 	&$msbuild ..\Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
@@ -25,7 +26,8 @@ if ($path) {
     &$msbuild ..\Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
     &$msbuild ..\Extensions\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m
 
-    Remove-Item .\NuGet -Force -Recurse
+    Remove-Item .\NuGet -Force -Recurse -ErrorAction SilentlyContinue
+
     New-Item -ItemType Directory -Force -Path .\NuGet
     .\NuGet.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
     .\NuGet.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion

Documents/Import_CodeSigningKey.md

@@ -0,0 +1,9 @@
+# Import codeSigningKey.pfx
+In order to import the key for code signing on a new developer machine use the following command within the VS developer command line:
+
+> sn –i codeSigningKey.pfx VS_KEY_EFCA4C5B6DFD4B4F
+
+The container name may be different.
+
+# Check if the assembly has a strong name
+> sn -vf MQTTnet.dll

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

@@ -20,6 +20,7 @@
     <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>
 

Frameworks/MQTTnet.AspnetCore/MQTTnet.AspnetCore.csproj

@@ -16,9 +16,9 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.0" />
-    <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.0" />
-    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.0" />
+    <PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.1" />
+    <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.1" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.1" />
   </ItemGroup>
 
   <ItemGroup>

Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs

@@ -17,8 +17,7 @@ namespace MQTTnet.Adapter
     public class MqttChannelAdapter : IMqttChannelAdapter
     {
         private const uint ErrorOperationAborted = 0x800703E3;
-
-        private static readonly byte[] EmptyBody = new byte[0];
+        private const int ReadBufferSize = 4096;  // TODO: Move buffer size to config
 
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
         private readonly IMqttNetLogger _logger;
@@ -91,28 +90,35 @@ namespace MQTTnet.Adapter
             MqttBasePacket packet = null;
             await ExecuteAndWrapExceptionAsync(async () =>
             {
-                ReceivedMqttPacket receivedMqttPacket;
-                if (timeout > TimeSpan.Zero)
+                ReceivedMqttPacket receivedMqttPacket = null;
+                try
                 {
-                    receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
-                }
-                else
-                {
-                    receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
-                }
+                    if (timeout > TimeSpan.Zero)
+                    {
+                        receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
+                    }
 
-                if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
-                {
-                    throw new TaskCanceledException();
-                }
+                    if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
+                    {
+                        throw new TaskCanceledException();
+                    }
 
-                packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
-                if (packet == null)
-                {
-                    throw new MqttProtocolViolationException("Received malformed packet.");
-                }
+                    packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
+                    if (packet == null)
+                    {
+                        throw new MqttProtocolViolationException("Received malformed packet.");
+                    }
 
-                _logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
+                    _logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
+                }
+                finally
+                {
+                    receivedMqttPacket?.Dispose();
+                }
             }).ConfigureAwait(false);
 
             return packet;
@@ -120,7 +126,7 @@ namespace MQTTnet.Adapter
 
         private static async Task<ReceivedMqttPacket> ReceiveAsync(Stream stream, CancellationToken cancellationToken)
         {
-            var header = MqttPacketReader.ReadHeaderFromSource(stream, cancellationToken);
+            var header = await MqttPacketReader.ReadHeaderAsync(stream, cancellationToken).ConfigureAwait(false);
             if (header == null)
             {
                 return null;
@@ -128,18 +134,35 @@ namespace MQTTnet.Adapter
 
             if (header.BodyLength == 0)
             {
-                return new ReceivedMqttPacket(header, EmptyBody);
+                return new ReceivedMqttPacket(header, new MemoryStream(new byte[0], false));
             }
 
-            var body = new byte[header.BodyLength];
+            var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(header.BodyLength) : new MemoryStream();
 
-            var offset = 0;
-            do
+            var buffer = new byte[ReadBufferSize];
+            while (body.Length < header.BodyLength)
             {
-                var readBytesCount = await stream.ReadAsync(body, offset, body.Length - offset, cancellationToken).ConfigureAwait(false);
-                offset += readBytesCount;
-            } while (offset < header.BodyLength);
-            
+                var bytesLeft = header.BodyLength - (int)body.Length;
+                if (bytesLeft > buffer.Length)
+                {
+                    bytesLeft = buffer.Length;
+                }
+
+                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)
+                {
+                    throw new MqttCommunicationException("Connection closed while reading remaining packet body.");
+                }
+
+                // Here is no need to await because internally only an array is used and no real I/O operation is made.
+                // Using async here will only generate overhead.
+                body.Write(buffer, 0, readBytesCount);
+            }
+
+            body.Seek(0L, SeekOrigin.Begin);
+
             return new ReceivedMqttPacket(header, body);
         }
 
@@ -167,7 +190,7 @@ namespace MQTTnet.Adapter
             }
             catch (COMException comException)
             {
-                if ((uint) comException.HResult == ErrorOperationAborted)
+                if ((uint)comException.HResult == ErrorOperationAborted)
                 {
                     throw new OperationCanceledException();
                 }

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/Adapter/ReceivedMqttPacket.cs

@@ -1,11 +1,12 @@
 using System;
+using System.IO;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Adapter
 {
-    public class ReceivedMqttPacket
+    public sealed class ReceivedMqttPacket : IDisposable
     {
-        public ReceivedMqttPacket(MqttPacketHeader header, byte[] body)
+        public ReceivedMqttPacket(MqttPacketHeader header, MemoryStream body)
         {
             Header = header ?? throw new ArgumentNullException(nameof(header));
             Body = body ?? throw new ArgumentNullException(nameof(body));
@@ -13,6 +14,11 @@ namespace MQTTnet.Adapter
 
         public MqttPacketHeader Header { get; }
 
-        public byte[] Body { get; }
+        public MemoryStream Body { get; }
+
+        public void Dispose()
+        {
+            Body?.Dispose();
+        }
     }
 }

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/Client/IMqttClient.cs

@@ -4,7 +4,7 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.Client
 {
-    public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
+    public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
     {
         bool IsConnected { get; }
 

Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs

@@ -13,6 +13,8 @@ namespace MQTTnet.Client
         
         TimeSpan CommunicationTimeout { get; }
         TimeSpan KeepAlivePeriod { get; }
+        TimeSpan? KeepAliveSendInterval { get; set; }
+
         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;
@@ -12,15 +13,16 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Client
 {
-    public class MqttClient : IMqttClient, IDisposable
+    public class MqttClient : IMqttClient
     {
+        private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
+        private readonly Stopwatch _sendTracker = new Stopwatch();
         private readonly IMqttClientAdapterFactory _adapterFactory;
         private readonly MqttPacketDispatcher _packetDispatcher;
         private readonly IMqttNetLogger _logger;
 
         private IMqttClientOptions _options;
         private bool _isReceivingPackets;
-        private int _latestPacketIdentifier;
         private CancellationTokenSource _cancellationTokenSource;
         private IMqttChannelAdapter _adapter;
 
@@ -49,7 +51,7 @@ namespace MQTTnet.Client
             {
                 _options = options;
                 _cancellationTokenSource = new CancellationTokenSource();
-                _latestPacketIdentifier = 0;
+                _packetIdentifierProvider.Reset();
                 _packetDispatcher.Reset();
 
                 _adapter = _adapterFactory.CreateClientAdapter(options.ChannelOptions, _logger);
@@ -59,10 +61,12 @@ namespace MQTTnet.Client
                 _logger.Trace<MqttClient>("Connection with server established.");
 
                 await StartReceivingPacketsAsync(_cancellationTokenSource.Token).ConfigureAwait(false);
-                var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false);
 
+                var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false);
                 _logger.Trace<MqttClient>("MQTT connection with server established.");
 
+                _sendTracker.Restart();
+
                 if (_options.KeepAlivePeriod != TimeSpan.Zero)
                 {
                     StartSendingKeepAliveMessages(_cancellationTokenSource.Token);
@@ -75,7 +79,8 @@ namespace MQTTnet.Client
             catch (Exception exception)
             {
                 _logger.Error<MqttClient>(exception, "Error while connecting with server.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
+
                 throw;
             }
         }
@@ -93,7 +98,7 @@ namespace MQTTnet.Client
             }
             finally
             {
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(null).ConfigureAwait(false);
             }
         }
 
@@ -105,7 +110,7 @@ namespace MQTTnet.Client
 
             var subscribePacket = new MqttSubscribePacket
             {
-                PacketIdentifier = GetNewPacketIdentifier(),
+                PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(),
                 TopicFilters = topicFilters.ToList()
             };
 
@@ -127,7 +132,7 @@ namespace MQTTnet.Client
 
             var unsubscribePacket = new MqttUnsubscribePacket
             {
-                PacketIdentifier = GetNewPacketIdentifier(),
+                PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(),
                 TopicFilters = topicFilters.ToList()
             };
 
@@ -148,14 +153,14 @@ 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.ToArray()).ConfigureAwait(false);
                             break;
                         }
                     case MqttQualityOfServiceLevel.AtLeastOnce:
                         {
                             foreach (var publishPacket in qosGroup)
                             {
-                                publishPacket.PacketIdentifier = GetNewPacketIdentifier();
+                                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
                                 await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket).ConfigureAwait(false);
                             }
 
@@ -165,9 +170,15 @@ namespace MQTTnet.Client
                         {
                             foreach (var publishPacket in qosGroup)
                             {
-                                publishPacket.PacketIdentifier = GetNewPacketIdentifier();
+                                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;
@@ -216,7 +227,7 @@ namespace MQTTnet.Client
             if (IsConnected) throw new MqttProtocolViolationException(message);
         }
 
-        private async Task DisconnectInternalAsync()
+        private async Task DisconnectInternalAsync(Exception exception)
         {
             var clientWasConnected = IsConnected;
             IsConnected = false;
@@ -236,14 +247,14 @@ namespace MQTTnet.Client
                 await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
                 _logger.Info<MqttClient>("Disconnected from adapter.");
             }
-            catch (Exception exception)
+            catch (Exception adapterException)
             {
-                _logger.Warning<MqttClient>(exception, "Error while disconnecting from adapter.");
+                _logger.Warning<MqttClient>(adapterException, "Error while disconnecting from adapter.");
             }
             finally
             {
                 _logger.Info<MqttClient>("Disconnected.");
-                Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected));
+                Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected, exception));
             }
         }
 
@@ -324,24 +335,32 @@ 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
         {
-            var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.CommunicationTimeout);
-            await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false);
-            return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
-        }
+            ushort? identifier = null;
+            if (requestPacket is IMqttPacketWithIdentifier requestPacketWithIdentifier)
+            {
+                identifier = requestPacketWithIdentifier.PacketIdentifier;
+            }
 
-        private ushort GetNewPacketIdentifier()
-        {
-            return (ushort)Interlocked.Increment(ref _latestPacketIdentifier);
+            var packetAwaiter = _packetDispatcher.WaitForPacketAsync(typeof(TResponsePacket), identifier, _options.CommunicationTimeout);
+            await SendAsync(requestPacket).ConfigureAwait(false);
+
+            return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
         }
 
         private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
@@ -352,8 +371,22 @@ namespace MQTTnet.Client
             {
                 while (!cancellationToken.IsCancellationRequested)
                 {
-                    await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
-                    await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false);
+                    TimeSpan keepAliveSendInterval;
+                    if (_options.KeepAliveSendInterval.HasValue)
+                    {
+                        keepAliveSendInterval = _options.KeepAliveSendInterval.Value;
+                    }
+                    else
+                    {
+                        keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75);
+                    }
+
+                    if (_sendTracker.Elapsed > keepAliveSendInterval)
+                    {
+                        await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
+                    }
+                    
+                    await Task.Delay(keepAliveSendInterval, cancellationToken).ConfigureAwait(false);
                 }
             }
             catch (OperationCanceledException)
@@ -363,7 +396,7 @@ namespace MQTTnet.Client
                     return;
                 }
 
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(null).ConfigureAwait(false);
             }
             catch (MqttCommunicationException exception)
             {
@@ -373,12 +406,12 @@ namespace MQTTnet.Client
                 }
 
                 _logger.Warning<MqttClient>(exception, "MQTT communication exception while sending/receiving keep alive packets.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
                 _logger.Warning<MqttClient>(exception, "Unhandled exception while sending/receiving keep alive packets.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
             }
             finally
             {
@@ -413,7 +446,7 @@ namespace MQTTnet.Client
                     return;
                 }
 
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(null).ConfigureAwait(false);
             }
             catch (MqttCommunicationException exception)
             {
@@ -423,12 +456,12 @@ namespace MQTTnet.Client
                 }
 
                 _logger.Warning<MqttClient>(exception, "MQTT communication exception while receiving packets.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
                 _logger.Error<MqttClient>(exception, "Unhandled exception while receiving packets.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
             }
             finally
             {

Frameworks/MQTTnet.NetStandard/Client/MqttClientDisconnectedEventArgs.cs

@@ -4,11 +4,14 @@ namespace MQTTnet.Client
 {
     public class MqttClientDisconnectedEventArgs : EventArgs
     {
-        public MqttClientDisconnectedEventArgs(bool clientWasConnected)
+        public MqttClientDisconnectedEventArgs(bool clientWasConnected, Exception exception)
         {
             ClientWasConnected = clientWasConnected;
+            Exception = exception;
         }
 
         public bool ClientWasConnected { get; }
+
+        public Exception Exception { get; }
     }
 }

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/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

@@ -10,8 +10,7 @@ namespace MQTTnet.Client
 {
     public class MqttPacketDispatcher
     {
-        private readonly ConcurrentDictionary<Type, TaskCompletionSource<MqttBasePacket>> _packetByResponseType = new ConcurrentDictionary<Type, TaskCompletionSource<MqttBasePacket>>();
-        private readonly ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>> _packetByResponseTypeAndIdentifier = new ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>>();
+        private readonly ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>> _awaiters = new ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>>();
         private readonly IMqttNetLogger _logger;
 
         public MqttPacketDispatcher(IMqttNetLogger logger)
@@ -19,11 +18,9 @@ namespace MQTTnet.Client
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
-        public async Task<MqttBasePacket> WaitForPacketAsync(MqttBasePacket request, Type responseType, TimeSpan timeout)
+        public async Task<MqttBasePacket> WaitForPacketAsync(Type responseType, ushort? identifier, TimeSpan timeout)
         {
-            if (request == null) throw new ArgumentNullException(nameof(request));
-
-            var packetAwaiter = AddPacketAwaiter(request, responseType);
+            var packetAwaiter = AddPacketAwaiter(responseType, identifier);
             try
             {
                 return await packetAwaiter.Task.TimeoutAfter(timeout).ConfigureAwait(false);
@@ -35,7 +32,7 @@ namespace MQTTnet.Client
             }
             finally
             {
-                RemovePacketAwaiter(request, responseType);
+                RemovePacketAwaiter(responseType, identifier);
             }
         }
 
@@ -44,21 +41,20 @@ namespace MQTTnet.Client
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
             var type = packet.GetType();
-            if (packet is IMqttPacketWithIdentifier withIdentifier)
+
+            if (_awaiters.TryGetValue(type, out var byId))
             {
-                if (_packetByResponseTypeAndIdentifier.TryGetValue(type, out var byid))
+                ushort? identifier = 0;
+                if (packet is IMqttPacketWithIdentifier packetWithIdentifier)
                 {
-                    if (byid.TryRemove(withIdentifier.PacketIdentifier, out var tcs))
-                    {
-                        tcs.TrySetResult(packet);
-                        return;
-                    }
+                    identifier = packetWithIdentifier.PacketIdentifier;
+                }
+
+                if (byId.TryRemove(identifier.Value, out var tcs))
+                {
+                    tcs.TrySetResult(packet);
+                    return;
                 }
-            }
-            else if (_packetByResponseType.TryRemove(type, out var tcs))
-            {
-                tcs.TrySetResult(packet);
-                return;
             }
 
             throw new InvalidOperationException($"Packet of type '{type.Name}' not handled or dispatched.");
@@ -66,38 +62,36 @@ namespace MQTTnet.Client
 
         public void Reset()
         {
-            _packetByResponseTypeAndIdentifier.Clear();
-            _packetByResponseType.Clear();
+            _awaiters.Clear();
         }
 
-        private TaskCompletionSource<MqttBasePacket> AddPacketAwaiter(MqttBasePacket request, Type responseType)
+        private TaskCompletionSource<MqttBasePacket> AddPacketAwaiter(Type responseType, ushort? identifier)
         {
             var tcs = new TaskCompletionSource<MqttBasePacket>();
 
-            if (request is IMqttPacketWithIdentifier requestWithIdentifier)
+            if (!identifier.HasValue)
             {
-                var byId = _packetByResponseTypeAndIdentifier.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
-                byId[requestWithIdentifier.PacketIdentifier] = tcs;
+                identifier = 0;
             }
-            else
+
+            var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
+            if (!byId.TryAdd(identifier.Value, tcs))
             {
-                _packetByResponseType[responseType] = tcs;
+                throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{responseType}' with identifier {identifier}.");
             }
 
             return tcs;
         }
 
-        private void RemovePacketAwaiter(MqttBasePacket request, Type responseType)
+        private void RemovePacketAwaiter(Type responseType, ushort? identifier)
         {
-            if (request is IMqttPacketWithIdentifier requestWithIdentifier)
+            if (!identifier.HasValue)
             {
-                var byId = _packetByResponseTypeAndIdentifier.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
-                byId.TryRemove(requestWithIdentifier.PacketIdentifier, out var _);
-            }
-            else
-            {
-                _packetByResponseType.TryRemove(responseType, out var _);
+                identifier = 0;
             }
+
+            var byId = _awaiters.GetOrAdd(responseType, key => new ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>());
+            byId.TryRemove(identifier.Value, out var _);
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Client/MqttPacketIdentifierProvider.cs

@@ -0,0 +1,32 @@
+namespace MQTTnet.Client
+{
+    public class MqttPacketIdentifierProvider
+    {
+        private readonly object _syncRoot = new object();
+        private ushort _value;
+
+        public void Reset()
+        {
+            lock (_syncRoot)
+            {
+                _value = 0;
+            }
+        }
+
+        public ushort GetNewPacketIdentifier()
+        {
+            lock (_syncRoot)
+            {
+                _value++;
+
+                if (_value == 0)
+                {
+                    // As per official MQTT documentation the package identifier should never be 0.
+                    _value = 1;
+                }
+
+                return _value;
+            }
+        }
+    }
+}

Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogMessage.cs

@@ -28,5 +28,16 @@ namespace MQTTnet.Diagnostics
         public string Message { get; }
 
         public Exception Exception { get; }
+
+        public override string ToString()
+        {
+            var result = $"[{Timestamp:O}] [{LogId}] [{ThreadId}] [{Source}] [{Level}]: {Message}";
+            if (Exception != null)
+            {
+                result += Environment.NewLine + Exception;
+            }
+
+            return result;
+        }
     }
 }

Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs

@@ -110,22 +110,14 @@ namespace MQTTnet.Implementations
                 {
                     // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
                 }
-                catch (SocketException exception)
+                catch (Exception exception)
                 {
-                    if (exception.SocketErrorCode == SocketError.OperationAborted)
+                    if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
                     {
                         return;
                     }
 
                     _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
-                }
-                catch (Exception exception)
-                {
-                    _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
-                }
-                finally
-                {
-                    //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }
             }
@@ -153,22 +145,14 @@ namespace MQTTnet.Implementations
                 {
                     // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
                 }
-                catch (SocketException exception)
+                catch (Exception exception)
                 {
-                    if (exception.SocketErrorCode == SocketError.OperationAborted)
+                    if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
                     {
                         return;
                     }
 
-                    _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
-                }
-                catch (Exception exception)
-                {
                     _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at TLS endpoint.");
-                }
-                finally
-                {
-                    //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }
             }

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

@@ -69,12 +69,6 @@ namespace MQTTnet.Implementations
 
         public void Dispose()
         {
-            SendStream?.Dispose();
-            SendStream = null;
-
-            ReceiveStream?.Dispose();
-            ReceiveStream = null;
-
             _socket?.Dispose();
             _socket = 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/Internal/TaskExtensions.cs

@@ -30,7 +30,7 @@ namespace MQTTnet.Internal
 
                     if (task.IsFaulted)
                     {
-                        throw new MqttCommunicationException(task.Exception.GetBaseException());
+                        throw new MqttCommunicationException(task.Exception?.GetBaseException());
                     }
                 }
                 finally

Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj

@@ -15,8 +15,7 @@
     <Description />
     <Authors />
     <PackageId />
-    <SignAssembly>true</SignAssembly>
-    <AssemblyOriginatorKeyFile>codeSigningKey.pfx</AssemblyOriginatorKeyFile>
+    <SignAssembly>false</SignAssembly>
     <DelaySign>false</DelaySign>
   </PropertyGroup>
 
@@ -29,6 +28,7 @@
     <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>
 

Frameworks/MQTTnet.NetStandard/ManagedClient/ApplicationMessageProcessedEventArgs.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace MQTTnet.ManagedClient
+{
+    public class ApplicationMessageProcessedEventArgs : EventArgs
+    {
+        public ApplicationMessageProcessedEventArgs(MqttApplicationMessage applicationMessage, Exception exception)
+        {
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            Exception = exception;
+        }
+
+        public MqttApplicationMessage ApplicationMessage { get; }
+        public Exception Exception { get; }
+
+        public bool HasFailed => Exception != null;
+        public bool HasSucceeded => Exception == null;
+    }
+}

Frameworks/MQTTnet.NetStandard/ManagedClient/IManagedMqttClient.cs

@@ -5,13 +5,15 @@ using MQTTnet.Client;
 
 namespace MQTTnet.ManagedClient
 {
-    public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
+    public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
     {
         bool IsConnected { get; }
 
         event EventHandler<MqttClientConnectedEventArgs> Connected;
         event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
 
+        event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
+
         Task StartAsync(IManagedMqttClientOptions options);
         Task StopAsync();
 

Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClient.cs

@@ -11,11 +11,12 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.ManagedClient
 {
-    public class ManagedMqttClient : IManagedMqttClient, IDisposable
+    public class ManagedMqttClient : IManagedMqttClient
     {
         private readonly BlockingCollection<MqttApplicationMessage> _messageQueue = new BlockingCollection<MqttApplicationMessage>();
         private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly SemaphoreSlim _subscriptionsSemaphore = new SemaphoreSlim(1, 1);
+        private readonly List<string> _unsubscriptions = new List<string>();
 
         private readonly IMqttClient _mqttClient;
         private readonly IMqttNetLogger _logger;
@@ -27,7 +28,7 @@ namespace MQTTnet.ManagedClient
         private IManagedMqttClientOptions _options;
 
         private bool _subscriptionsNotPushed;
-        
+
         public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
         {
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -43,6 +44,8 @@ namespace MQTTnet.ManagedClient
         public event EventHandler<MqttClientConnectedEventArgs> Connected;
         public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
+        public event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
+        public event EventHandler SynchronizingSubscriptionsFailed;
 
         public async Task StartAsync(IManagedMqttClientOptions options)
         {
@@ -57,7 +60,7 @@ namespace MQTTnet.ManagedClient
             if (_connectionCancellationToken != null) throw new InvalidOperationException("The managed client is already started.");
 
             _options = options;
-            
+
             if (_options.Storage != null)
             {
                 _storageManager = new ManagedMqttClientStorageManager(_options.Storage);
@@ -65,9 +68,9 @@ namespace MQTTnet.ManagedClient
             }
 
             _connectionCancellationToken = new CancellationTokenSource();
-            
+
 #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 MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token).ConfigureAwait(false);
+            Task.Run(async () => await MaintainConnectionAsync(_connectionCancellationToken.Token).ConfigureAwait(false), _connectionCancellationToken.Token).ConfigureAwait(false);
 #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
 
             _logger.Info<ManagedMqttClient>("Started");
@@ -85,7 +88,7 @@ namespace MQTTnet.ManagedClient
 
             return Task.FromResult(0);
         }
-        
+
         public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
         {
             if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
@@ -96,7 +99,7 @@ namespace MQTTnet.ManagedClient
                 {
                     await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
                 }
-                
+
                 _messageQueue.Add(applicationMessage);
             }
         }
@@ -129,6 +132,7 @@ namespace MQTTnet.ManagedClient
                 {
                     if (_subscriptions.Remove(topic))
                     {
+                        _unsubscriptions.Add(topic);
                         _subscriptionsNotPushed = true;
                     }
                 }
@@ -153,7 +157,7 @@ namespace MQTTnet.ManagedClient
             {
                 while (!cancellationToken.IsCancellationRequested)
                 {
-                    await TryMaintainConnectionAsync(cancellationToken);
+                    await TryMaintainConnectionAsync(cancellationToken).ConfigureAwait(false);
                 }
             }
             catch (OperationCanceledException)
@@ -184,11 +188,11 @@ namespace MQTTnet.ManagedClient
 
                 if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
                 {
-                    await PushSubscriptionsAsync().ConfigureAwait(false);
+                    await SynchronizeSubscriptionsAsync().ConfigureAwait(false);
 
                     StartPublishing();
 
-                    
+
                     return;
                 }
 
@@ -209,7 +213,7 @@ namespace MQTTnet.ManagedClient
                 _logger.Error<ManagedMqttClient>(exception, "Unhandled exception while maintaining connection.");
             }
         }
-        
+
         private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
         {
             try
@@ -227,7 +231,7 @@ namespace MQTTnet.ManagedClient
                         continue;
                     }
 
-                    await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);                 
+                    await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
                 }
             }
             catch (OperationCanceledException)
@@ -245,6 +249,7 @@ namespace MQTTnet.ManagedClient
 
         private async Task TryPublishQueuedMessageAsync(MqttApplicationMessage message)
         {
+            Exception transmitException = null;
             try
             {
                 await _mqttClient.PublishAsync(message).ConfigureAwait(false);
@@ -256,6 +261,8 @@ namespace MQTTnet.ManagedClient
             }
             catch (MqttCommunicationException exception)
             {
+                transmitException = exception;
+
                 _logger.Warning<ManagedMqttClient>(exception, "Publishing application message failed.");
 
                 if (message.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
@@ -265,39 +272,60 @@ namespace MQTTnet.ManagedClient
             }
             catch (Exception exception)
             {
+                transmitException = exception;
                 _logger.Error<ManagedMqttClient>(exception, "Unhandled exception while publishing queued application message.");
             }
+            finally
+            {
+                ApplicationMessageProcessed?.Invoke(this, new ApplicationMessageProcessedEventArgs(message, transmitException));
+            }
         }
 
-        private async Task PushSubscriptionsAsync()
+        private async Task SynchronizeSubscriptionsAsync()
         {
             _logger.Info<ManagedMqttClient>(nameof(ManagedMqttClient), "Synchronizing subscriptions");
 
             List<TopicFilter> subscriptions;
+            List<string> unsubscriptions;
+
             await _subscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
             try
             {
                 subscriptions = _subscriptions.Select(i => new TopicFilter(i.Key, i.Value)).ToList();
+
+                unsubscriptions = new List<string>(_unsubscriptions);
+                _unsubscriptions.Clear();
+
                 _subscriptionsNotPushed = false;
             }
             finally
             {
                 _subscriptionsSemaphore.Release();
             }
-
-            if (!subscriptions.Any())
+            
+            if (!subscriptions.Any() && !unsubscriptions.Any())
             {
                 return;
             }
 
             try
             {
-                await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false);
+                if (subscriptions.Any())
+                {
+                    await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false);
+                }
+
+                if (unsubscriptions.Any())
+                {
+                    await _mqttClient.UnsubscribeAsync(unsubscriptions).ConfigureAwait(false);
+                }
             }
             catch (Exception exception)
             {
-                _logger.Warning<ManagedMqttClient>(exception, "Synchronizing subscriptions failed");
+                _logger.Warning<ManagedMqttClient>(exception, "Synchronizing subscriptions failed.");
                 _subscriptionsNotPushed = true;
+
+                SynchronizingSubscriptionsFailed?.Invoke(this, EventArgs.Empty);
             }
         }
 

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/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/IMqttPacketSerializer.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Serializer
@@ -10,6 +11,6 @@ namespace MQTTnet.Serializer
 
         ICollection<ArraySegment<byte>> Serialize(MqttBasePacket mqttPacket);
 
-        MqttBasePacket Deserialize(MqttPacketHeader header, byte[] body);
+        MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body);
     }
 }

Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs

@@ -22,17 +22,21 @@ namespace MQTTnet.Serializer
 
         public bool EndOfRemainingData => BaseStream.Position == _header.BodyLength;
 
-        public static MqttPacketHeader ReadHeaderFromSource(Stream stream, CancellationToken cancellationToken)
+        public static async Task<MqttPacketHeader> ReadHeaderAsync(Stream stream, CancellationToken cancellationToken)
         {
-            var buffer = stream.ReadByte();
-            if (buffer == -1)
+            // 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.
+            var buffer = new byte[1];
+            var readCount = await stream.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
+            if (readCount <= 0)
             {
                 return null;
             }
-            
-            var fixedHeader = (byte)buffer;
+
+            var fixedHeader = buffer[0];
             var controlPacketType = (MqttControlPacketType)(fixedHeader >> 4);
-            var bodyLength = ReadBodyLengthFromSource(stream, cancellationToken);
+            var bodyLength = await ReadBodyLengthAsync(stream, cancellationToken).ConfigureAwait(false);
 
             return new MqttPacketHeader
             {
@@ -80,13 +84,14 @@ namespace MQTTnet.Serializer
             return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
         }
 
-        private static int ReadBodyLengthFromSource(Stream stream, CancellationToken cancellationToken)
+        private static async Task<int> ReadBodyLengthAsync(Stream stream, CancellationToken cancellationToken)
         {
-            // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
+            // Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
             var multiplier = 1;
             var value = 0;
             byte encodedByte;
 
+            var buffer = new byte[1];
             var readBytes = new List<byte>();
             do
             {
@@ -95,21 +100,22 @@ namespace MQTTnet.Serializer
                     throw new TaskCanceledException();
                 }
 
-                var buffer = stream.ReadByte();
-                if (buffer == -1)
+                var readCount = await stream.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
+                if (readCount <= 0)
                 {
                     throw new MqttCommunicationException("Connection closed while reading remaining length data.");
                 }
 
-                encodedByte = (byte)buffer;
+                encodedByte = buffer[0];
                 readBytes.Add(encodedByte);
 
                 value += (byte)(encodedByte & 127) * multiplier;
-                multiplier *= 128;
                 if (multiplier > 128 * 128 * 128)
                 {
                     throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
                 }
+
+                multiplier *= 128;
             } while ((encodedByte & 128) != 0);
 
             return value;

Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs

@@ -42,13 +42,12 @@ namespace MQTTnet.Serializer
             }
         }
 
-        public MqttBasePacket Deserialize(MqttPacketHeader header, byte[] body)
+        public MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body)
         {
             if (header == null) throw new ArgumentNullException(nameof(header));
             if (body == null) throw new ArgumentNullException(nameof(body));
 
-            using (var bodyStream = new MemoryStream(body))
-            using (var reader = new MqttPacketReader(header, bodyStream))
+            using (var reader = new MqttPacketReader(header, body))
             {
                 return Deserialize(header, reader);
             }
@@ -76,7 +75,7 @@ namespace MQTTnet.Serializer
             }
         }
 
-        private static MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader)
+        private MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader)
         {
             switch (header.ControlPacketType)
             {
@@ -279,20 +278,22 @@ namespace MQTTnet.Serializer
             return packet;
         }
 
-        private static MqttBasePacket DeserializeConnAck(MqttPacketReader reader)
+        private MqttBasePacket DeserializeConnAck(MqttPacketReader reader)
         {
-            var variableHeader1 = reader.ReadByte();
-            var variableHeader2 = reader.ReadByte();
+            var packet = new MqttConnAckPacket();
 
-            var packet = new MqttConnAckPacket
+            var firstByteReader = new ByteReader(reader.ReadByte());
+
+            if (ProtocolVersion == MqttProtocolVersion.V311)
             {
-                IsSessionPresent = new ByteReader(variableHeader1).Read(),
-                ConnectReturnCode = (MqttConnectReturnCode)variableHeader2
-            };
+                packet.IsSessionPresent = firstByteReader.Read();
+            }
+
+            packet.ConnectReturnCode = (MqttConnectReturnCode)reader.ReadByte();
 
             return packet;
         }
-
+        
         private static void ValidateConnectPacket(MqttConnectPacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
@@ -374,14 +375,21 @@ namespace MQTTnet.Serializer
 
         private byte Serialize(MqttConnAckPacket packet, MqttPacketWriter writer)
         {
-            var connectAcknowledgeFlags = new ByteWriter();
-
-            if (ProtocolVersion == MqttProtocolVersion.V311)
+            if (ProtocolVersion == MqttProtocolVersion.V310)
             {
+                writer.Write(0);
+            }
+            else if (ProtocolVersion == MqttProtocolVersion.V311)
+            {
+                var connectAcknowledgeFlags = new ByteWriter();
                 connectAcknowledgeFlags.Write(packet.IsSessionPresent);
+                writer.Write(connectAcknowledgeFlags);
+            }
+            else
+            {
+                throw new MqttProtocolViolationException("Protocol version not supported.");
             }
 
-            writer.Write(connectAcknowledgeFlags);
             writer.Write((byte)packet.ConnectReturnCode);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
@@ -389,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);
         }
@@ -402,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
             {
@@ -436,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);
         }
@@ -459,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)
             {
@@ -475,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)
             {
@@ -492,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)
             {
@@ -505,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/ConnectedMqttClient.cs

@@ -12,5 +12,7 @@ namespace MQTTnet.Server
         public TimeSpan LastPacketReceived { get; set; }
 
         public TimeSpan LastNonKeepAlivePacketReceived { get; set; }
+
+        public int PendingApplicationMessages { get; set; }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/IMqttServer.cs

@@ -6,12 +6,20 @@ namespace MQTTnet.Server
 {
     public interface IMqttServer : IApplicationMessageReceiver, IApplicationMessagePublisher
     {
-        event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
-        event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
         event EventHandler<MqttServerStartedEventArgs> Started;
 
+        event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
+        event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
+        event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
+        
+        IMqttServerOptions Options { get; }
+
         Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync();
 
+        Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters);
+        Task UnsubscribeAsync(string clientId, IList<string> topicFilters);
+
         Task StartAsync(IMqttServerOptions options);
         Task StopAsync();
     }

Frameworks/MQTTnet.NetStandard/Server/IMqttServerOptions.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Net;
 
 namespace MQTTnet.Server
 {

Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs

@@ -0,0 +1,88 @@
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Diagnostics;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttClientKeepAliveMonitor
+    {
+        private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch();
+        private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch();
+
+        private readonly string _clientId;
+        private readonly Func<Task> _timeoutCallback;
+        private readonly IMqttNetLogger _logger;
+
+        public MqttClientKeepAliveMonitor(string clientId, Func<Task> timeoutCallback, IMqttNetLogger logger)
+        {
+            _clientId = clientId;
+            _timeoutCallback = timeoutCallback;
+            _logger = logger;
+        }
+
+        public TimeSpan LastPacketReceived => _lastPacketReceivedTracker.Elapsed;
+
+        public TimeSpan LastNonKeepAlivePacketReceived => _lastNonKeepAlivePacketReceivedTracker.Elapsed;
+
+        public void Start(int keepAlivePeriod, CancellationToken cancellationToken)
+        {
+            if (keepAlivePeriod == 0)
+            {
+                return;
+            }
+
+            Task.Run(async () => await RunAsync(keepAlivePeriod, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
+        }
+
+        private async Task RunAsync(int keepAlivePeriod, CancellationToken cancellationToken)
+        {
+            try
+            {
+                _lastPacketReceivedTracker.Restart();
+                _lastNonKeepAlivePacketReceivedTracker.Restart();
+
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    // Values described here: [MQTT-3.1.2-24].
+                    if (_lastPacketReceivedTracker.Elapsed.TotalSeconds > keepAlivePeriod * 1.5D)
+                    {
+                        _logger.Warning<MqttClientSession>("Client '{0}': Did not receive any packet or keep alive signal.", _clientId);
+
+                        if (_timeoutCallback != null)
+                        {
+                            await _timeoutCallback().ConfigureAwait(false);
+                        }
+
+                        return;
+                    }
+
+                    await Task.Delay(keepAlivePeriod, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            catch (Exception exception)
+            {
+                _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while checking keep alive timeouts.", _clientId);
+            }
+            finally
+            {
+                _logger.Trace<MqttClientSession>("Client {0}: Stopped checking keep alive timeout.", _clientId);
+            }
+        }
+
+        public void PacketReceived(MqttBasePacket packet)
+        {
+            _lastPacketReceivedTracker.Restart();
+
+            if (!(packet is MqttPingReqPacket))
+            {
+                _lastNonKeepAlivePacketReceivedTracker.Restart();
+            }
+        }
+    }
+}

Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs

@@ -12,18 +12,21 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClientPendingMessagesQueue : IDisposable
     {
-        private readonly BlockingCollection<MqttBasePacket> _queue = new BlockingCollection<MqttBasePacket>();
+        private readonly ConcurrentQueue<MqttBasePacket> _queue = new ConcurrentQueue<MqttBasePacket>();
+        private readonly SemaphoreSlim _queueWaitSemaphore = new SemaphoreSlim(0);
         private readonly IMqttServerOptions _options;
-        private readonly MqttClientSession _session;
+        private readonly MqttClientSession _clientSession;
         private readonly IMqttNetLogger _logger;
 
-        public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession session, IMqttNetLogger logger)
+        public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetLogger logger)
         {
-            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
-            _session = session ?? throw new ArgumentNullException(nameof(session));
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
+        public int Count => _queue.Count;
+
         public void Start(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -40,8 +43,10 @@ namespace MQTTnet.Server
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
-            _queue.Add(packet);
-            _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _session.ClientId);
+            _queue.Enqueue(packet);
+            _queueWaitSemaphore.Release();
+
+            _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
         }
 
         private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -58,7 +63,7 @@ namespace MQTTnet.Server
             }
             catch (Exception exception)
             {
-                _logger.Error<MqttClientPendingMessagesQueue>(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _session.ClientId);
+                _logger.Error<MqttClientPendingMessagesQueue>(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _clientSession.ClientId);
             }
         }
 
@@ -67,27 +72,32 @@ namespace MQTTnet.Server
             MqttBasePacket packet = null;
             try
             {
-                packet = _queue.Take(cancellationToken);
-                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
+                await _queueWaitSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+                if (!_queue.TryDequeue(out packet))
+                {
+                    throw new InvalidOperationException(); // should not happen
+                }
 
-                _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _session.ClientId);
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false);
+
+                _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
             }
             catch (Exception exception)
             {
                 if (exception is MqttCommunicationTimedOutException)
                 {
-                    _logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _session.ClientId);
+                    _logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _clientSession.ClientId);
                 }
                 else if (exception is MqttCommunicationException)
                 {
-                    _logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _session.ClientId);
+                    _logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _clientSession.ClientId);
                 }
                 else if (exception is OperationCanceledException)
                 {
                 }
                 else
                 {
-                    _logger.Error<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed (ClientId: {0}).", _session.ClientId);
+                    _logger.Error<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed (ClientId: {0}).", _clientSession.ClientId);
                 }
 
                 if (packet is MqttPublishPacket publishPacket)
@@ -95,17 +105,18 @@ namespace MQTTnet.Server
                     if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
                     {
                         publishPacket.Dup = true;
-                        _queue.Add(packet, CancellationToken.None);
+                        _queue.Enqueue(packet);
+                        _queueWaitSemaphore.Release();
                     }
                 }
 
-                await _session.StopAsync();
+                await _clientSession.StopAsync();
             }
         }
 
         public void Dispose()
         {
-            _queue?.Dispose();
+            _queueWaitSemaphore?.Dispose();
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs

@@ -1,8 +1,9 @@
 using System;
-using System.Diagnostics;
+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,16 +15,10 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClientSession : IDisposable
     {
-        private readonly Stopwatch _lastPacketReceivedTracker = Stopwatch.StartNew();
-        private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = Stopwatch.StartNew();
-
+        private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         private readonly IMqttServerOptions _options;
         private readonly IMqttNetLogger _logger;
-
-        private readonly MqttClientSessionsManager _sessionsManager;
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
-        private readonly MqttClientSubscriptionsManager _subscriptionsManager;
-        private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
 
         private IMqttChannelAdapter _adapter;
         private CancellationTokenSource _cancellationTokenSource;
@@ -33,28 +28,31 @@ namespace MQTTnet.Server
             string clientId,
             IMqttServerOptions options,
             MqttRetainedMessagesManager retainedMessagesManager,
-            MqttClientSessionsManager sessionsManager,
             IMqttNetLogger logger)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
-            _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
             ClientId = clientId;
-            
-            _subscriptionsManager = new MqttClientSubscriptionsManager(_options);
-            _pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
+
+            KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeoutAsync, _logger);
+            SubscriptionsManager = new MqttClientSubscriptionsManager(_options, clientId);
+            PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
         }
 
+        public Func<MqttClientSession, MqttApplicationMessage, Task> ApplicationMessageReceivedCallback { get; set; }
+
+        public MqttClientSubscriptionsManager SubscriptionsManager { get; }
+
+        public MqttClientPendingMessagesQueue PendingMessagesQueue { get; }
+
+        public MqttClientKeepAliveMonitor KeepAliveMonitor { get; }
+
         public string ClientId { get; }
 
         public MqttProtocolVersion? ProtocolVersion => _adapter?.PacketSerializer.ProtocolVersion;
 
-        public TimeSpan LastPacketReceived => _lastPacketReceivedTracker.Elapsed;
-
-        public TimeSpan LastNonKeepAlivePacketReceived => _lastNonKeepAlivePacketReceivedTracker.Elapsed;
-
         public bool IsConnected => _adapter != null;
 
         public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
@@ -70,15 +68,8 @@ namespace MQTTnet.Server
                 _adapter = adapter;
                 _cancellationTokenSource = cancellationTokenSource;
 
-                _pendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
-
-                _lastPacketReceivedTracker.Restart();
-                _lastNonKeepAlivePacketReceivedTracker.Restart();
-
-                if (connectPacket.KeepAlivePeriod > 0)
-                {
-                    StartCheckingKeepAliveTimeout(TimeSpan.FromSeconds(connectPacket.KeepAlivePeriod), cancellationTokenSource.Token);
-                }
+                PendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
+                KeepAliveMonitor.Start(connectPacket.KeepAlivePeriod, cancellationTokenSource.Token);
 
                 await ReceivePacketsAsync(adapter, cancellationTokenSource.Token).ConfigureAwait(false);
             }
@@ -121,8 +112,8 @@ namespace MQTTnet.Server
                 var willMessage = _willMessage;
                 if (willMessage != null)
                 {
-                    _willMessage = null; //clear willmessage so it is send just once
-                    await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage).ConfigureAwait(false);
+                    _willMessage = null; // clear willmessage so it is send just once
+                    await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
                 }
             }
         }
@@ -131,7 +122,7 @@ namespace MQTTnet.Server
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            var result = await _subscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
+            var result = await SubscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
             if (!result.IsSubscribed)
             {
                 return;
@@ -140,15 +131,52 @@ namespace MQTTnet.Server
             var publishPacket = applicationMessage.ToPublishPacket();
             publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
 
-            _pendingMessagesQueue.Enqueue(publishPacket);
+            if (publishPacket.QualityOfServiceLevel > 0)
+            {
+                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
+            }
+
+            PendingMessagesQueue.Enqueue(publishPacket);
+        }
+
+        public async Task SubscribeAsync(IList<TopicFilter> topicFilters)
+        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            await SubscriptionsManager.SubscribeAsync(new MqttSubscribePacket
+            {
+                TopicFilters = topicFilters
+            }).ConfigureAwait(false);
+
+            await EnqueueSubscribedRetainedMessagesAsync(topicFilters).ConfigureAwait(false);
+        }
+
+        public Task UnsubscribeAsync(IList<string> topicFilters)
+        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            return SubscriptionsManager.UnsubscribeAsync(new MqttUnsubscribePacket
+            {
+                TopicFilters = topicFilters
+            });
         }
 
         public void Dispose()
         {
-            _pendingMessagesQueue?.Dispose();
+            ApplicationMessageReceivedCallback = null;
+
+            SubscriptionsManager?.Dispose();
+            PendingMessagesQueue?.Dispose();
+
             _cancellationTokenSource?.Dispose();
         }
 
+        private Task StopDueToKeepAliveTimeoutAsync()
+        {
+            _logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId);
+            return StopAsync();
+        }
+
         private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
         {
             try
@@ -156,14 +184,7 @@ namespace MQTTnet.Server
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
-
-                    _lastPacketReceivedTracker.Restart();
-
-                    if (!(packet is MqttPingReqPacket))
-                    {
-                        _lastNonKeepAlivePacketReceivedTracker.Restart();
-                    }
-
+                    KeepAliveMonitor.PacketReceived(packet);
                     await ProcessReceivedPacketAsync(adapter, packet, cancellationToken).ConfigureAwait(false);
                 }
             }
@@ -191,7 +212,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)
@@ -201,7 +222,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)
@@ -231,31 +257,30 @@ namespace MQTTnet.Server
 
         private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
         {
-            var subscribeResult = await _subscriptionsManager.SubscribeAsync(subscribePacket, ClientId);
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket).ConfigureAwait(false);
+            var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).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 adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttDisconnectPacket() }).ConfigureAwait(false);
                 await StopAsync().ConfigureAwait(false);
             }
 
-            await EnqueueSubscribedRetainedMessagesAsync(subscribePacket).ConfigureAwait(false);
+            await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
         }
 
         private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
         {
-            var unsubscribeResult = await _subscriptionsManager.UnsubscribeAsync(unsubscribePacket);
-
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, unsubscribeResult);
+            var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { unsubscribeResult });
         }
 
-        private async Task EnqueueSubscribedRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
+        private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
         {
-            var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(subscribePacket);
-            foreach (var publishPacket in retainedMessages)
+            var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters);
+            foreach (var applicationMessage in retainedMessages)
             {
-                await EnqueueApplicationMessageAsync(publishPacket);
+                await EnqueueApplicationMessageAsync(applicationMessage).ConfigureAwait(false);
             }
         }
 
@@ -267,7 +292,7 @@ namespace MQTTnet.Server
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
                     {
-                        return _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
+                        return ApplicationMessageReceivedCallback?.Invoke(this, applicationMessage);
                     }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                     {
@@ -286,62 +311,25 @@ namespace MQTTnet.Server
 
         private async Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
-            await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
+            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)
         {
             // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
-            await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
+            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);
-        }
-
-        private void StartCheckingKeepAliveTimeout(TimeSpan keepAlivePeriod, CancellationToken cancellationToken)
-        {
-            Task.Run(
-                async () => await CheckKeepAliveTimeoutAsync(keepAlivePeriod, cancellationToken).ConfigureAwait(false)
-                , cancellationToken);
-        }
-
-        private async Task CheckKeepAliveTimeoutAsync(TimeSpan keepAlivePeriod, CancellationToken cancellationToken)
-        {
-            try
-            {
-                while (!cancellationToken.IsCancellationRequested)
-                {
-                    // Values described here: [MQTT-3.1.2-24].
-                    if (_lastPacketReceivedTracker.Elapsed.TotalSeconds > keepAlivePeriod.TotalSeconds * 1.5D)
-                    {
-                        _logger.Warning<MqttClientSession>("Client '{0}': Did not receive any packet or keep alive signal.", ClientId);
-                        await StopAsync();
-                        return;
-                    }
-
-                    await Task.Delay(keepAlivePeriod, cancellationToken);
-                }
-            }
-            catch (OperationCanceledException)
-            {
-            }
-            catch (Exception exception)
-            {
-                _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while checking keep alive timeouts.", ClientId);
-            }
-            finally
-            {
-                _logger.Trace<MqttClientSession>("Client {0}: Stopped checking keep alive timeout.", ClientId);
-            }
+            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response });
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs

@@ -12,27 +12,32 @@ using MQTTnet.Serializer;
 
 namespace MQTTnet.Server
 {
-    public sealed class MqttClientSessionsManager
+    public sealed class MqttClientSessionsManager : IDisposable
     {
         private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
 
         private readonly IMqttServerOptions _options;
-        private readonly MqttServer _server;
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
         private readonly IMqttNetLogger _logger;
 
-        public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, MqttServer server, IMqttNetLogger logger)
+        public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
         {
-            _server = server ?? throw new ArgumentNullException(nameof(server));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
             _options = options ?? throw new ArgumentNullException(nameof(options));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
         }
 
-        public async Task RunClientSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
+        public Action<ConnectedMqttClient> ClientConnectedCallback { get; set; }
+        public Action<ConnectedMqttClient> 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;
+            MqttClientSession clientSession = null;
             try
             {
                 if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false) is MqttConnectPacket connectPacket))
@@ -48,29 +53,30 @@ 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[] { new MqttConnAckPacket
                     {
                         ConnectReturnCode = connectReturnCode
-                    }).ConfigureAwait(false);
+                    }}).ConfigureAwait(false);
 
                     return;
                 }
 
-                var clientSession = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
+                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[] { new MqttConnAckPacket
                 {
                     ConnectReturnCode = connectReturnCode,
-                    IsSessionPresent = clientSession.IsExistingSession
-                }).ConfigureAwait(false);
+                    IsSessionPresent = result.IsExistingSession
+                }}).ConfigureAwait(false);
 
-                _server.OnClientConnected(new ConnectedMqttClient
+                ClientConnectedCallback?.Invoke(new ConnectedMqttClient
                 {
                     ClientId = clientId,
                     ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
                 });
 
-                await clientSession.Session.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
+                await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -87,10 +93,11 @@ namespace MQTTnet.Server
                     // ignored
                 }
 
-                _server.OnClientDisconnected(new ConnectedMqttClient
+                ClientDisconnectedCallback?.Invoke(new ConnectedMqttClient
                 {
                     ClientId = clientId,
-                    ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
+                    ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion,
+                    PendingApplicationMessages = clientSession?.PendingMessagesQueue.Count ?? 0
                 });
             }
         }
@@ -122,8 +129,9 @@ namespace MQTTnet.Server
                 {
                     ClientId = s.Value.ClientId,
                     ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311,
-                    LastPacketReceived = s.Value.LastPacketReceived,
-                    LastNonKeepAlivePacketReceived = s.Value.LastNonKeepAlivePacketReceived
+                    LastPacketReceived = s.Value.KeepAliveMonitor.LastPacketReceived,
+                    LastNonKeepAlivePacketReceived = s.Value.KeepAliveMonitor.LastNonKeepAlivePacketReceived,
+                    PendingApplicationMessages = s.Value.PendingMessagesQueue.Count
                 }).ToList();
             }
             finally
@@ -147,7 +155,7 @@ namespace MQTTnet.Server
                     await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
                 }
 
-                _server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage);
+                ApplicationMessageReceivedCallback?.Invoke(senderClientSession?.ClientId, applicationMessage);
             }
             catch (Exception exception)
             {
@@ -168,18 +176,61 @@ namespace MQTTnet.Server
             }
         }
 
+        public async Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            await _semaphore.WaitAsync().ConfigureAwait(false);
+            try
+            {
+                if (!_sessions.TryGetValue(clientId, out var session))
+                {
+                    throw new InvalidOperationException($"Client session {clientId} is unknown.");
+                }
+
+                await session.SubscribeAsync(topicFilters);
+            }
+            finally
+            {
+                _semaphore.Release();
+            }
+        }
+
+        public async Task UnsubscribeAsync(string clientId, IList<string> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            await _semaphore.WaitAsync().ConfigureAwait(false);
+            try
+            {
+                if (!_sessions.TryGetValue(clientId, out var session))
+                {
+                    throw new InvalidOperationException($"Client session {clientId} is unknown.");
+                }
+
+                await session.UnsubscribeAsync(topicFilters);
+            }
+            finally
+            {
+                _semaphore.Release();
+            }
+        }
+
         private MqttApplicationMessage InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
         {
-            if (_options.ApplicationMessageInterceptor == null)
+            var interceptor = _options.ApplicationMessageInterceptor;
+            if (interceptor == null)
             {
                 return applicationMessage;
             }
 
             var interceptorContext = new MqttApplicationMessageInterceptorContext(
-                senderClientSession.ClientId,
+                senderClientSession?.ClientId,
                 applicationMessage);
 
-            _options.ApplicationMessageInterceptor(interceptorContext);
+            interceptor(interceptorContext);
             return interceptorContext.ApplicationMessage;
         }
 
@@ -229,7 +280,14 @@ namespace MQTTnet.Server
                 {
                     isExistingSession = false;
 
-                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, this, _logger);
+                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, _logger)
+                    {
+                        ApplicationMessageReceivedCallback = DispatchApplicationMessageAsync
+                    };
+
+                    clientSession.SubscriptionsManager.TopicSubscribedCallback = ClientSubscribedTopicCallback;
+                    clientSession.SubscriptionsManager.TopicUnsubscribedCallback = ClientUnsubscribedTopicCallback;
+
                     _sessions[connectPacket.ClientId] = clientSession;
 
                     _logger.Trace<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
@@ -242,5 +300,16 @@ namespace MQTTnet.Server
                 _semaphore.Release();
             }
         }
+
+        public void Dispose()
+        {
+            ClientConnectedCallback = null;
+            ClientDisconnectedCallback = null;
+            ClientSubscribedTopicCallback = null;
+            ClientUnsubscribedTopicCallback = null;
+            ApplicationMessageReceivedCallback = null;
+            
+            _semaphore?.Dispose();
+        }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscribedTopicEventArgs.cs

@@ -0,0 +1,17 @@
+using System;
+
+namespace MQTTnet.Server
+{
+    public class MqttClientSubscribedTopicEventArgs : EventArgs
+    {
+        public MqttClientSubscribedTopicEventArgs(string clientId, TopicFilter topicFilter)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
+        }
+
+        public string ClientId { get; }
+
+        public TopicFilter TopicFilter { get; }
+    }
+}

Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs

@@ -8,24 +8,33 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
-    public sealed class MqttClientSubscriptionsManager
+    public sealed class MqttClientSubscriptionsManager : IDisposable
     {
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
         private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly IMqttServerOptions _options;
+        private readonly string _clientId;
 
-        public MqttClientSubscriptionsManager(IMqttServerOptions options)
+        public MqttClientSubscriptionsManager(IMqttServerOptions options, string clientId)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
         }
 
-        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket, string clientId)
+        public Action<string, TopicFilter> TopicSubscribedCallback { get; set; }
+        public Action<string, string> TopicUnsubscribedCallback { get; set; }
+
+        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket)
         {
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
 
             var result = new MqttClientSubscribeResult
             {
-                ResponsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>(),
+                ResponsePacket = new MqttSubAckPacket
+                {
+                    PacketIdentifier = subscribePacket.PacketIdentifier
+                },
+
                 CloseConnection = false
             };
 
@@ -34,7 +43,7 @@ namespace MQTTnet.Server
             {
                 foreach (var topicFilter in subscribePacket.TopicFilters)
                 {
-                    var interceptorContext = InterceptSubscribe(clientId, topicFilter);
+                    var interceptorContext = InterceptSubscribe(topicFilter);
                     if (!interceptorContext.AcceptSubscription)
                     {
                         result.ResponsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.Failure);
@@ -52,6 +61,7 @@ namespace MQTTnet.Server
                     if (interceptorContext.AcceptSubscription)
                     {
                         _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                        TopicSubscribedCallback?.Invoke(_clientId, topicFilter);
                     }
                 }
             }
@@ -73,6 +83,7 @@ namespace MQTTnet.Server
                 foreach (var topicFilter in unsubscribePacket.TopicFilters)
                 {
                     _subscriptions.Remove(topicFilter);
+                    TopicUnsubscribedCallback?.Invoke(_clientId, topicFilter);
                 }
             }
             finally
@@ -80,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)
@@ -117,9 +131,25 @@ namespace MQTTnet.Server
             }
         }
 
-        private MqttSubscriptionInterceptorContext InterceptSubscribe(string clientId, TopicFilter topicFilter)
+        public void Dispose()
         {
-            var interceptorContext = new MqttSubscriptionInterceptorContext(clientId, topicFilter);
+            _semaphore?.Dispose();
+        }
+
+        private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel)
+        {
+            switch (qualityOfServiceLevel)
+            {
+                case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
+                case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
+                case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
+                default: return MqttSubscribeReturnCode.Failure;
+            }
+        }
+
+        private MqttSubscriptionInterceptorContext InterceptSubscribe(TopicFilter topicFilter)
+        {
+            var interceptorContext = new MqttSubscriptionInterceptorContext(_clientId, topicFilter);
             _options.SubscriptionInterceptor?.Invoke(interceptorContext);
             return interceptorContext;
         }
@@ -146,16 +176,5 @@ namespace MQTTnet.Server
                 QualityOfServiceLevel = effectiveQoS
             };
         }
-
-        private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel)
-        {
-            switch (qualityOfServiceLevel)
-            {
-                case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
-                case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
-                case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
-                default: return MqttSubscribeReturnCode.Failure;
-            }
-        }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientUnSubscribedTopicEventArgs.cs

@@ -0,0 +1,17 @@
+using System;
+
+namespace MQTTnet.Server
+{
+    public class MqttClientUnsubscribedTopicEventArgs : EventArgs
+    {
+        public MqttClientUnsubscribedTopicEventArgs(string clientId, string topicFilter)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
+        }
+
+        public string ClientId { get; }
+
+        public string TopicFilter { get; }
+    }
+}

Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs

@@ -4,11 +4,10 @@ using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
-using MQTTnet.Packets;
 
 namespace MQTTnet.Server
 {
-    public sealed class MqttRetainedMessagesManager
+    public sealed class MqttRetainedMessagesManager : IDisposable
     {
         private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>();
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
@@ -68,7 +67,7 @@ namespace MQTTnet.Server
             }
         }
 
-        public async Task<List<MqttApplicationMessage>> GetSubscribedMessagesAsync(MqttSubscribePacket subscribePacket)
+        public async Task<List<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<TopicFilter> topicFilters)
         {
             var retainedMessages = new List<MqttApplicationMessage>();
 
@@ -77,13 +76,8 @@ namespace MQTTnet.Server
             {
                 foreach (var retainedMessage in _retainedMessages.Values)
                 {
-                    foreach (var topicFilter in subscribePacket.TopicFilters)
+                    foreach (var topicFilter in topicFilters)
                     {
-                        if (retainedMessage.QualityOfServiceLevel < topicFilter.QualityOfServiceLevel)
-                        {
-                            continue;
-                        }
-
                         if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic))
                         {
                             continue;
@@ -102,6 +96,11 @@ namespace MQTTnet.Server
             return retainedMessages;
         }
 
+        public void Dispose()
+        {
+            _semaphore?.Dispose();
+        }
+
         private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage)
         {
             var saveIsRequired = false;

Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs

@@ -16,30 +16,47 @@ namespace MQTTnet.Server
         private MqttClientSessionsManager _clientSessionsManager;
         private MqttRetainedMessagesManager _retainedMessagesManager;
         private CancellationTokenSource _cancellationTokenSource;
-        private IMqttServerOptions _options;
 
         public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
         {
+            if (adapters == null) throw new ArgumentNullException(nameof(adapters));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
-            if (adapters == null)
-            {
-                throw new ArgumentNullException(nameof(adapters));
-            }
-
             _adapters = adapters.ToList();
         }
 
         public event EventHandler<MqttServerStartedEventArgs> Started;
+
         public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
         public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
+        public event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
+        public event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
+
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
+        public IMqttServerOptions Options { get; private set; }
+
         public Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
         {
             return _clientSessionsManager.GetConnectedClientsAsync();
         }
 
+        public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            return _clientSessionsManager.SubscribeAsync(clientId, topicFilters);
+        }
+
+        public Task UnsubscribeAsync(string clientId, IList<string> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters);
+        }
+
         public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
         {
             if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
@@ -54,20 +71,28 @@ namespace MQTTnet.Server
 
         public async Task StartAsync(IMqttServerOptions options)
         {
-            _options = options ?? throw new ArgumentNullException(nameof(options));
+            Options = options ?? throw new ArgumentNullException(nameof(options));
 
             if (_cancellationTokenSource != null) throw new InvalidOperationException("The server is already started.");
 
             _cancellationTokenSource = new CancellationTokenSource();
-            _retainedMessagesManager = new MqttRetainedMessagesManager(_options, _logger);
-            _clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, this, _logger);
 
+            _retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
             await _retainedMessagesManager.LoadMessagesAsync();
 
+            _clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _logger)
+            {
+                ClientConnectedCallback = OnClientConnected,
+                ClientDisconnectedCallback = OnClientDisconnected,
+                ClientSubscribedTopicCallback = OnClientSubscribedTopic,
+                ClientUnsubscribedTopicCallback = OnClientUnsubscribedTopic,
+                ApplicationMessageReceivedCallback = OnApplicationMessageReceived
+            };
+
             foreach (var adapter in _adapters)
             {
                 adapter.ClientAccepted += OnClientAccepted;
-                await adapter.StartAsync(_options);
+                await adapter.StartAsync(Options);
             }
 
             _logger.Info<MqttServer>("Started.");
@@ -98,39 +123,46 @@ namespace MQTTnet.Server
             }
             finally
             {
+                _clientSessionsManager?.Dispose();
+                _retainedMessagesManager?.Dispose();
+
                 _cancellationTokenSource = null;
                 _retainedMessagesManager = null;
                 _clientSessionsManager = null;
             }
         }
 
-        internal void OnClientConnected(ConnectedMqttClient client)
+        private void OnClientConnected(ConnectedMqttClient client)
         {
-            if (client == null) throw new ArgumentNullException(nameof(client));
-
             _logger.Info<MqttServer>("Client '{0}': Connected.", client.ClientId);
             ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client));
         }
 
-        internal void OnClientDisconnected(ConnectedMqttClient client)
+        private void OnClientDisconnected(ConnectedMqttClient client)
         {
-            if (client == null) throw new ArgumentNullException(nameof(client));
-
             _logger.Info<MqttServer>("Client '{0}': Disconnected.", client.ClientId);
             ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client));
         }
 
-        internal void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
+        private void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
         {
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+            ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
+        }
 
+        private void OnClientUnsubscribedTopic(string clientId, string topicFilter)
+        {
+            ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
+        }
+
+        private void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
+        {
             ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(clientId, applicationMessage));
         }
 
         private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
         {
             eventArgs.SessionTask = Task.Run(
-                async () => await _clientSessionsManager.RunClientSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
+                async () => await _clientSessionsManager.RunSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
                 _cancellationTokenSource.Token);
         }
     }

Frameworks/MQTTnet.Netstandard/MQTTnet.NetStandard.csproj

@@ -0,0 +1,65 @@
+<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.DotSettings

@@ -1,7 +1,13 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
+	<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_ACCESSORHOLDER_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String>
 	<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_CALL_PARENTHESES/@EntryValue">False</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_PARENTHESES/@EntryValue">False</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_PARENTHESES/@EntryValue">False</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpRenamePlacementToArrangementMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EAddAccessorOwnerDeclarationBracesMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

README.md

@@ -18,12 +18,12 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
 * TLS 1.2 support for client and server (but not UWP servers)
 * Extensible communication channels (i.e. In-Memory, TCP, TCP+TLS, WS)
 * Lightweight (only the low level implementation of MQTT, no overhead)
-* Performance optimized (processing ~30.000 messages / second)*
+* Performance optimized (processing ~40.000 messages / second)*
 * Interfaces included for mocking and testing
 * Access to internal trace messages
 * Unit tested (70+ tests)
 
-\* Tested on local machine with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
+\* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
 
 ### Client
 
@@ -43,6 +43,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
 * Retained messages are supported including persisting via interface methods (own implementation required)
 * WebSockets supported (via ASP.NET Core 2.0, separate nuget)
 * A custom message interceptor can be added which allows transforming or extending every received application message
+* Validate subscriptions and deny subscribing of certain topics depending on requesting clients
 
 ## Supported frameworks
 

Tests/MQTTnet.Core.Tests/MqttClientTests.cs

@@ -0,0 +1,36 @@
+using System.Net.Sockets;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.Core.Tests
+{
+    [TestClass]
+    public class MqttClientTests
+    {
+
+        [TestMethod]
+        public async Task ClientDisconnectException()
+        {
+            var factory = new MqttFactory();
+            var client = factory.CreateMqttClient();
+
+            var exceptionIsCorrect = false;
+            client.Disconnected += (s, e) =>
+            {
+                exceptionIsCorrect = e.Exception is MqttCommunicationException c && c.InnerException is SocketException;
+            };
+
+            try
+            {
+                await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("wrong-server").Build());
+            }
+            catch
+            {
+            }
+            
+            Assert.IsTrue(exceptionIsCorrect);
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs

@@ -0,0 +1,66 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Diagnostics;
+using MQTTnet.Packets;
+using MQTTnet.Server;
+
+namespace MQTTnet.Core.Tests
+{
+    [TestClass]
+    public class MqttKeepAliveMonitorTests
+    {
+        [TestMethod]
+        public void KeepAlive_Timeout()
+        {
+            var timeoutCalledCount = 0;
+
+            var monitor = new MqttClientKeepAliveMonitor(string.Empty, delegate
+            {
+                timeoutCalledCount++;
+                return Task.FromResult(0);
+            }, new MqttNetLogger());
+
+            Assert.AreEqual(0, timeoutCalledCount);
+
+            monitor.Start(1, CancellationToken.None);
+
+            Assert.AreEqual(0, timeoutCalledCount);
+
+            Thread.Sleep(2000); // Internally the keep alive timeout is multiplied with 1.5 as per protocol specification.
+
+            Assert.AreEqual(1, timeoutCalledCount);
+        }
+
+        [TestMethod]
+        public void KeepAlive_NoTimeout()
+        {
+            var timeoutCalledCount = 0;
+
+            var monitor = new MqttClientKeepAliveMonitor(string.Empty, delegate
+            {
+                timeoutCalledCount++;
+                return Task.FromResult(0);
+            }, new MqttNetLogger());
+
+            Assert.AreEqual(0, timeoutCalledCount);
+
+            monitor.Start(1, CancellationToken.None);
+
+            Assert.AreEqual(0, timeoutCalledCount);
+
+            // Simulate traffic.
+            Thread.Sleep(1000); // Internally the keep alive timeout is multiplied with 1.5 as per protocol specification.
+            monitor.PacketReceived(new MqttPublishPacket());
+            Thread.Sleep(1000);
+            monitor.PacketReceived(new MqttPublishPacket());
+            Thread.Sleep(1000);
+
+            Assert.AreEqual(0, timeoutCalledCount);
+
+            Thread.Sleep(2000);
+
+            Assert.AreEqual(1, timeoutCalledCount);
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/MqttPacketIdentifierProviderTests.cs

@@ -0,0 +1,32 @@
+using Microsoft.VisualStudio.TestTools .UnitTesting;
+using MQTTnet.Client;
+
+namespace MQTTnet.Core.Tests
+{
+    [TestClass]
+    public class MqttPacketIdentifierProviderTests
+    {
+        [TestMethod]
+        public void Reset()
+        {
+            var p = new MqttPacketIdentifierProvider();
+            Assert.AreEqual(1, p.GetNewPacketIdentifier());
+            Assert.AreEqual(2, p.GetNewPacketIdentifier());
+            p.Reset();
+            Assert.AreEqual(1, p.GetNewPacketIdentifier());
+        }
+
+        [TestMethod]
+        public void ReachBoundaries()
+        {
+            var p = new MqttPacketIdentifierProvider();
+
+            for (ushort i = 0; i < ushort.MaxValue; i++)
+            {
+                Assert.AreEqual(i + 1, p.GetNewPacketIdentifier());
+            }
+
+            Assert.AreEqual(1, p.GetNewPacketIdentifier());
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs

@@ -12,7 +12,7 @@ namespace MQTTnet.Core.Tests
         public void MqttPacketReader_EmptyStream()
         {
             var memStream = new MemoryStream();
-            var header = MqttPacketReader.ReadHeaderFromSource(memStream, CancellationToken.None);
+            var header = MqttPacketReader.ReadHeaderAsync(memStream, CancellationToken.None).GetAwaiter().GetResult();
 
             Assert.IsNull(header);
         }

Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs

@@ -119,10 +119,10 @@ namespace MQTTnet.Core.Tests
         {
             var p = new MqttConnAckPacket
             {
-                ConnectReturnCode = MqttConnectReturnCode.ConnectionAccepted
+                ConnectReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized
             };
 
-            SerializeAndCompare(p, "IAIAAA==", MqttProtocolVersion.V310);
+            SerializeAndCompare(p, "IAIABQ==", MqttProtocolVersion.V310);
         }
 
         [TestMethod]
@@ -137,6 +137,17 @@ namespace MQTTnet.Core.Tests
             DeserializeAndCompare(p, "IAIBBQ==");
         }
 
+        [TestMethod]
+        public void DeserializeV310_MqttConnAckPacket()
+        {
+            var p = new MqttConnAckPacket
+            {
+                ConnectReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized
+            };
+
+            DeserializeAndCompare(p, "IAIABQ==", MqttProtocolVersion.V310);
+        }
+
         [TestMethod]
         public void SerializeV311_MqttDisconnectPacket()
         {
@@ -397,19 +408,19 @@ namespace MQTTnet.Core.Tests
             Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(chunks)));
         }
 
-        private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
+        private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
         {
-            var serializer = new MqttPacketSerializer();
+            var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion };
 
             var buffer1 = serializer.Serialize(packet);
 
             using (var headerStream = new MemoryStream(Join(buffer1)))
             {
-                var header = MqttPacketReader.ReadHeaderFromSource(headerStream, CancellationToken.None);
+                var header = MqttPacketReader.ReadHeaderAsync(headerStream, CancellationToken.None).GetAwaiter().GetResult();
 
                 using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.BodyLength))
                 {
-                    var deserializedPacket = serializer.Deserialize(header, bodyStream.ToArray());
+                    var deserializedPacket = serializer.Deserialize(header, bodyStream);
                     var buffer2 = serializer.Serialize(deserializedPacket);
 
                     Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2)));

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -79,7 +79,7 @@ namespace MQTTnet.Core.Tests
         }
 
         [TestMethod]
-        public async Task MqttServer_Unsubscribe()
+        public async Task MqttServer_SubscribeUnsubscribe()
         {
             var serverAdapter = new TestMqttServerAdapter();
             var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
@@ -99,13 +99,31 @@ namespace MQTTnet.Core.Tests
                 await c2.PublishAsync(message);
                 Assert.AreEqual(0, receivedMessagesCount);
 
+                var subscribeEventCalled = false;
+                s.ClientSubscribedTopic += (_, e) =>
+                    {
+                        subscribeEventCalled = e.TopicFilter.Topic == "a" && e.ClientId == "c1";
+                    };
+
                 await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
+                await Task.Delay(100);
+                Assert.IsTrue(subscribeEventCalled, "Subscribe event not called.");
+
                 await c2.PublishAsync(message);
 
                 await Task.Delay(500);
                 Assert.AreEqual(1, receivedMessagesCount);
 
+                var unsubscribeEventCalled = false;
+                s.ClientUnsubscribedTopic += (_, e) =>
+                {
+                    unsubscribeEventCalled = e.TopicFilter == "a" && e.ClientId == "c1";
+                };
+
                 await c1.UnsubscribeAsync("a");
+                await Task.Delay(100);
+                Assert.IsTrue(unsubscribeEventCalled, "Unsubscribe event not called.");
+
                 await c2.PublishAsync(message);
 
                 await Task.Delay(500);
@@ -299,7 +317,7 @@ namespace MQTTnet.Core.Tests
 
             var serverAdapter = new TestMqttServerAdapter();
             var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
-            
+
             try
             {
                 var options = new MqttServerOptions { ApplicationMessageInterceptor = Interceptor };
@@ -330,6 +348,41 @@ namespace MQTTnet.Core.Tests
             }
         }
 
+        [TestMethod]
+        public async Task MqttServer_Body()
+        {
+            var serverAdapter = new TestMqttServerAdapter();
+            var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
+
+            var bodyIsMatching = false;
+            try
+            {
+                await s.StartAsync(new MqttServerOptions());
+
+                var c1 = await serverAdapter.ConnectTestClient(s, "c1");
+                var c2 = await serverAdapter.ConnectTestClient(s, "c2");
+
+                c1.ApplicationMessageReceived += (_, e) =>
+                {
+                    if (Encoding.UTF8.GetString(e.ApplicationMessage.Payload) == "The body")
+                    {
+                        bodyIsMatching = true;
+                    }
+                };
+
+                await c1.SubscribeAsync("A", MqttQualityOfServiceLevel.AtMostOnce);
+                await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("A").WithPayload(Encoding.UTF8.GetBytes("The body")).Build());
+
+                await Task.Delay(500);
+            }
+            finally
+            {
+                await s.StopAsync();
+            }
+
+            Assert.IsTrue(bodyIsMatching);
+        }
+
         private class TestStorage : IMqttServerStorage
         {
             private IList<MqttApplicationMessage> _messages = new List<MqttApplicationMessage>();

Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs

@@ -11,12 +11,12 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeSingleSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {
@@ -32,12 +32,12 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeDifferentQoSSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce));
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {
@@ -53,13 +53,13 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeTwoTimesSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce));
             sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtLeastOnce));
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {
@@ -75,12 +75,12 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeSingleNoSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {
@@ -94,12 +94,12 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {

Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj

@@ -10,7 +10,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.3" />
+    <PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.5" />
   </ItemGroup>
 
   <ItemGroup>

Tests/MQTTnet.TestApp.AspNetCore2/package.json

@@ -6,7 +6,7 @@
   },
 
   "dependencies": {
-    "mqtt": "2.13.1",
+    "mqtt": "2.15.1",
     "@types/node": "8.0.46",
     "systemjs": "0.20.19",
     "typescript": "2.3.4"

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>
@@ -138,10 +138,10 @@
   </ItemGroup>
   <ItemGroup>
     <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform">
-      <Version>5.4.0</Version>
+      <Version>6.0.8</Version>
     </PackageReference>
     <PackageReference Include="Microsoft.Toolkit.Uwp.UI.Controls">
-      <Version>2.0.0</Version>
+      <Version>2.2.0</Version>
     </PackageReference>
   </ItemGroup>
   <PropertyGroup Condition=" '$(VisualStudioVersion)' == '' or '$(VisualStudioVersion)' &lt; '14.0' ">

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

@@ -42,7 +42,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,
@@ -197,8 +198,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;
         }