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.7.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.7.0.
+    <releaseNotes>* Updated to MQTTnet 2.7.1.
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2018</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
     <dependencies>
       <group targetFramework="netstandard2.0">
-        <dependency id="MQTTnet" version="2.7.0" />
+        <dependency id="MQTTnet" version="2.7.1" />
       </group>
     </dependencies>
   </metadata>

Build/MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.7.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,16 +10,11 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
-    <releaseNotes>* [Core] Fixed some still thread blocking parts in the code (thanks to @kpreisser).
-* [Core] Updated 3rd-Party packages.
-* [Core] Fixed wrong packet identifier calculation (thanks to @benpittoors).
-* [Core] Fixed an issue when reading the body of a package from a disconnected sender (thanks to @kpreisser).
-* [Core] Fixed wrong parsing of the body length (thanks to @kpreisser).
-* [Client] The client interfaces are now implementing _IDisposable_.
-* [Client] The disconnected event now contains the exception which was thrown and causing the disconnect.
-* [Server] Fixed an issue which lets the server block 1 second after accepting a new connection (thanks to @kpreisser).
-* [Server] The server now allows managing client subscriptions.
-* [Server] Added events for topic subscriptions.
+    <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-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> 

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

@@ -3,4 +3,7 @@ In order to import the key for code signing on a new developer machine use the f
 
 > sn –i codeSigningKey.pfx VS_KEY_EFCA4C5B6DFD4B4F
 
-The container name may be different.
+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.NetStandard/Adapter/MqttChannelAdapter.cs

@@ -138,7 +138,7 @@ namespace MQTTnet.Adapter
             }
 
             var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(header.BodyLength) : new MemoryStream();
-            
+
             var buffer = new byte[ReadBufferSize];
             while (body.Length < header.BodyLength)
             {
@@ -149,7 +149,7 @@ namespace MQTTnet.Adapter
                 }
 
                 var readBytesCount = await stream.ReadAsync(buffer, 0, bytesLeft, cancellationToken).ConfigureAwait(false);
-                
+
                 // Check if the client closed the connection before sending the full body.
                 if (readBytesCount == 0)
                 {
@@ -162,7 +162,7 @@ namespace MQTTnet.Adapter
             }
 
             body.Seek(0L, SeekOrigin.Begin);
-            
+
             return new ReceivedMqttPacket(header, body);
         }
 
@@ -190,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/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/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;
@@ -15,6 +16,7 @@ namespace MQTTnet.Client
     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;
@@ -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);
@@ -149,7 +153,7 @@ namespace MQTTnet.Client
                     case MqttQualityOfServiceLevel.AtMostOnce:
                         {
                             // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
-                            await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosGroup).ConfigureAwait(false);
+                            await SendAsync(qosGroup.ToArray()).ConfigureAwait(false);
                             break;
                         }
                     case MqttQualityOfServiceLevel.AtLeastOnce:
@@ -167,8 +171,14 @@ namespace MQTTnet.Client
                             foreach (var publishPacket in qosGroup)
                             {
                                 publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
+
                                 var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
-                                await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).ConfigureAwait(false);
+                                var pubRelPacket = new MqttPubRelPacket
+                                {
+                                    PacketIdentifier = pubRecPacket.PacketIdentifier
+                                };
+
+                                await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket).ConfigureAwait(false);
                             }
 
                             break;
@@ -325,18 +335,31 @@ 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);
+            ushort? identifier = null;
+            if (requestPacket is IMqttPacketWithIdentifier requestPacketWithIdentifier)
+            {
+                identifier = requestPacketWithIdentifier.PacketIdentifier;
+            }
+
+            var packetAwaiter = _packetDispatcher.WaitForPacketAsync(typeof(TResponsePacket), identifier, _options.CommunicationTimeout);
+            await SendAsync(requestPacket).ConfigureAwait(false);
+
             return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
         }
 
@@ -348,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)

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

@@ -12,6 +12,8 @@ namespace MQTTnet.ManagedClient
         event EventHandler<MqttClientConnectedEventArgs> Connected;
         event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
 
+        event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
+
         Task StartAsync(IManagedMqttClientOptions options);
         Task StopAsync();
 

Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClient.cs

@@ -16,6 +16,7 @@ namespace MQTTnet.ManagedClient
         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/MqttPacketSerializer.cs

@@ -75,7 +75,7 @@ namespace MQTTnet.Serializer
             }
         }
 
-        private static MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader)
+        private MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader)
         {
             switch (header.ControlPacketType)
             {
@@ -278,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));
@@ -373,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);
@@ -388,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);
         }
@@ -401,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
             {
@@ -435,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);
         }
@@ -458,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)
             {
@@ -474,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)
             {
@@ -491,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)
             {
@@ -504,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/IMqttServer.cs

@@ -13,6 +13,8 @@ namespace MQTTnet.Server
         event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
         event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
         
+        IMqttServerOptions Options { get; }
+
         Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync();
 
         Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters);

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

@@ -78,7 +78,7 @@ namespace MQTTnet.Server
                     throw new InvalidOperationException(); // should not happen
                 }
 
-                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false);
 
                 _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
             }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs

@@ -1,9 +1,9 @@
 using System;
 using System.Collections.Generic;
-using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
+using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Internal;
@@ -15,12 +15,9 @@ 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 MqttRetainedMessagesManager _retainedMessagesManager;
 
         private IMqttChannelAdapter _adapter;
@@ -38,7 +35,8 @@ namespace MQTTnet.Server
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
             ClientId = clientId;
-            
+
+            KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeoutAsync, _logger);
             SubscriptionsManager = new MqttClientSubscriptionsManager(_options, clientId);
             PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
         }
@@ -49,14 +47,12 @@ namespace MQTTnet.Server
 
         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)
@@ -73,14 +69,7 @@ namespace MQTTnet.Server
                 _cancellationTokenSource = cancellationTokenSource;
 
                 PendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
-
-                _lastPacketReceivedTracker.Restart();
-                _lastNonKeepAlivePacketReceivedTracker.Restart();
-
-                if (connectPacket.KeepAlivePeriod > 0)
-                {
-                    StartCheckingKeepAliveTimeout(TimeSpan.FromSeconds(connectPacket.KeepAlivePeriod), cancellationTokenSource.Token);
-                }
+                KeepAliveMonitor.Start(connectPacket.KeepAlivePeriod, cancellationTokenSource.Token);
 
                 await ReceivePacketsAsync(adapter, cancellationTokenSource.Token).ConfigureAwait(false);
             }
@@ -123,7 +112,7 @@ namespace MQTTnet.Server
                 var willMessage = _willMessage;
                 if (willMessage != null)
                 {
-                    _willMessage = null; //clear willmessage so it is send just once
+                    _willMessage = null; // clear willmessage so it is send just once
                     await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
                 }
             }
@@ -142,31 +131,34 @@ namespace MQTTnet.Server
             var publishPacket = applicationMessage.ToPublishPacket();
             publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
 
+            if (publishPacket.QualityOfServiceLevel > 0)
+            {
+                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
+            }
+
             PendingMessagesQueue.Enqueue(publishPacket);
         }
 
-        public Task SubscribeAsync(IList<TopicFilter> topicFilters)
+        public async Task SubscribeAsync(IList<TopicFilter> topicFilters)
         {
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
-            var response = SubscriptionsManager.SubscribeAsync(new MqttSubscribePacket
+            await SubscriptionsManager.SubscribeAsync(new MqttSubscribePacket
             {
                 TopicFilters = topicFilters
-            });
+            }).ConfigureAwait(false);
 
-            return response;
+            await EnqueueSubscribedRetainedMessagesAsync(topicFilters).ConfigureAwait(false);
         }
 
         public Task UnsubscribeAsync(IList<string> topicFilters)
         {
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
 
-            var response = SubscriptionsManager.UnsubscribeAsync(new MqttUnsubscribePacket
+            return SubscriptionsManager.UnsubscribeAsync(new MqttUnsubscribePacket
             {
                 TopicFilters = topicFilters
             });
-
-            return response;
         }
 
         public void Dispose()
@@ -179,6 +171,12 @@ namespace MQTTnet.Server
             _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
@@ -186,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);
                 }
             }
@@ -221,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)
@@ -231,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)
@@ -262,29 +258,29 @@ namespace MQTTnet.Server
         private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
         {
             var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { subscribeResult.ResponsePacket }).ConfigureAwait(false);
 
             if (subscribeResult.CloseConnection)
             {
-                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false);
+                await 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).ConfigureAwait(false);
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, unsubscribeResult);
+            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);
             }
         }
 
@@ -318,7 +314,7 @@ namespace MQTTnet.Server
             await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
 
             var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier };
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false);
         }
 
         private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
@@ -327,50 +323,13 @@ namespace MQTTnet.Server
             await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
 
             var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier };
-            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { response }).ConfigureAwait(false);
         }
 
         private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
         {
             var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier };
-            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response);
-        }
-
-        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

@@ -53,10 +53,10 @@ 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;
                 }
@@ -64,11 +64,11 @@ namespace MQTTnet.Server
                 var result = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
                 clientSession = result.Session;
 
-                await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
+                await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttConnAckPacket
                 {
                     ConnectReturnCode = connectReturnCode,
                     IsSessionPresent = result.IsExistingSession
-                }).ConfigureAwait(false);
+                }}).ConfigureAwait(false);
 
                 ClientConnectedCallback?.Invoke(new ConnectedMqttClient
                 {
@@ -129,8 +129,8 @@ 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();
             }
@@ -220,16 +220,17 @@ namespace MQTTnet.Server
 
         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;
         }
 

Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs

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

Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs

@@ -4,7 +4,6 @@ using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
-using MQTTnet.Packets;
 
 namespace MQTTnet.Server
 {
@@ -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;

Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs

@@ -16,7 +16,6 @@ namespace MQTTnet.Server
         private MqttClientSessionsManager _clientSessionsManager;
         private MqttRetainedMessagesManager _retainedMessagesManager;
         private CancellationTokenSource _cancellationTokenSource;
-        private IMqttServerOptions _options;
 
         public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
         {
@@ -35,6 +34,8 @@ namespace MQTTnet.Server
 
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
+        public IMqttServerOptions Options { get; private set; }
+
         public Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
         {
             return _clientSessionsManager.GetConnectedClientsAsync();
@@ -70,16 +71,16 @@ 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);
+            _retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
             await _retainedMessagesManager.LoadMessagesAsync();
 
-            _clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, _logger)
+            _clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _logger)
             {
                 ClientConnectedCallback = OnClientConnected,
                 ClientDisconnectedCallback = OnClientDisconnected,
@@ -91,7 +92,7 @@ namespace MQTTnet.Server
             foreach (var adapter in _adapters)
             {
                 adapter.ClientAccepted += OnClientAccepted;
-                await adapter.StartAsync(_options);
+                await adapter.StartAsync(Options);
             }
 
             _logger.Info<MqttServer>("Started.");
@@ -142,7 +143,7 @@ namespace MQTTnet.Server
             _logger.Info<MqttServer>("Client '{0}': Disconnected.", client.ClientId);
             ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client));
         }
-        
+
         private void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
         {
             ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
@@ -152,7 +153,7 @@ namespace MQTTnet.Server
         {
             ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
         }
-        
+
         private void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
         {
             ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(clientId, applicationMessage));

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>
 

README.md

@@ -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/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/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,9 +408,9 @@ 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);
 

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.1</Version>
+      <Version>6.0.8</Version>
     </PackageReference>
     <PackageReference Include="Microsoft.Toolkit.Uwp.UI.Controls">
-      <Version>2.1.1</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;
         }