Refactor trace messages

Reference to MQTT Tester - updated

Build/MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.5.1</version>
+    <version>2.5.2</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,18 +10,17 @@
     <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>* Fixed several library reference issues.
-* [Client] Fixed issue in _MqttClientOptionsBuilder_ when using no certificates (Thanks to 1iveowl).
+    <releaseNotes>* [Core] Refactored trace messages.
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2017</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
     <dependencies>
 
       <group targetFramework="netstandard1.3">
-        <dependency id="NETStandard.Library" version="1.6.1" />
-        <dependency id="Microsoft.Extensions.DependencyInjection" version="1.1.1" />
-        <dependency id="Microsoft.Extensions.Logging" version="1.1.2" />
-        <dependency id="Microsoft.Extensions.Options" version="1.1.2" />
+        <dependency id="NETStandard.Library" version="1.3.0" />
+        <dependency id="Microsoft.Extensions.DependencyInjection" version="1.0.2" />
+        <dependency id="Microsoft.Extensions.Logging" version="1.0.2" />
+        <dependency id="Microsoft.Extensions.Options" version="1.0.2" />
         <dependency id="System.Net.Security" version="4.3.2" />
         <dependency id="System.Net.WebSockets" version="4.3.0" />
         <dependency id="System.Net.WebSockets.Client" version="4.3.1" />
@@ -47,9 +46,15 @@
       </group>
         
       <group targetFramework="net452">
-        <dependency id="Microsoft.Extensions.DependencyInjection" version="1.1.1" />
-        <dependency id="Microsoft.Extensions.Logging" version="1.1.2" />
-        <dependency id="Microsoft.Extensions.Options" version="1.1.2" />
+        <dependency id="Microsoft.Extensions.DependencyInjection" version="1.0.2" />
+        <dependency id="Microsoft.Extensions.Logging" version="1.0.2" />
+        <dependency id="Microsoft.Extensions.Options" version="1.0.2" />
+      </group>
+
+      <group targetFramework="net461">
+        <dependency id="Microsoft.Extensions.DependencyInjection" version="2.0.0" />
+        <dependency id="Microsoft.Extensions.Logging" version="2.0.0" />
+        <dependency id="Microsoft.Extensions.Options" version="2.0.0" />
       </group>
 
     </dependencies>
@@ -71,6 +76,9 @@
     <!-- .NET Framework -->
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net452\MQTTnet.Core.*" target="lib\net452\"/>
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net452\MQTTnet.*" target="lib\net452\"/>
+
+    <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net461\MQTTnet.Core.*" target="lib\net461\"/>
+    <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net461\MQTTnet.*" target="lib\net461\"/>
     
   </files>
 </package>

Build/build.ps1

@@ -1,4 +1,4 @@
-param([string]$version)
+param([string]$assemblyVersion, [string]$nugetVersion)
 
 if ([string]::IsNullOrEmpty($version)) {$version = "0.0.1"}
 
@@ -8,16 +8,16 @@ $path = &$vswhere -latest -products * -requires Microsoft.Component.MSBuild -pro
 if ($path) {
     $msbuild = join-path $path 'MSBuild\15.0\Bin\MSBuild.exe'
 
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m
-    &$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$version /p:AssemblyVersion=$version /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
+    &$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.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$version /p:AssemblyVersion=$version /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
 
     Remove-Item .\NuGet -Force -Recurse
     New-Item -ItemType Directory -Force -Path .\NuGet
-    .\NuGet.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $version
-    .\NuGet.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $version
-}
-
+    .\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
+}

Frameworks/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs

@@ -2,6 +2,7 @@
 using Microsoft.AspNetCore.Builder;
 using Microsoft.Extensions.DependencyInjection;
 using MQTTnet.Core.Server;
+using System.Linq;
 
 namespace MQTTnet.AspNetCore
 {
@@ -14,8 +15,18 @@ namespace MQTTnet.AspNetCore
             {
                 if (context.Request.Path == path && context.WebSockets.IsWebSocketRequest)
                 {
+                    string subprotocol = null;
+
+                    if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues)
+                     && requestedSubProtocolValues.Count > 0
+                     && requestedSubProtocolValues.Any(v => v.ToLower() == "mqtt")
+                     )
+                    {
+                        subprotocol = "mqtt";
+                    }
+
                     var adapter = app.ApplicationServices.GetRequiredService<MqttWebSocketServerAdapter>();
-                    using (var webSocket = await context.WebSockets.AcceptWebSocketAsync("mqtt"))
+                    using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subprotocol))
                     {
                         await adapter.AcceptWebSocketAsync(webSocket);
                     }

Frameworks/MQTTnet.AspnetCore/MQTTnet.AspnetCore.csproj

@@ -2,6 +2,8 @@
 
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
+    <AssemblyVersion>2.5.2.0</AssemblyVersion>
+    <FileVersion>2.5.2.0</FileVersion>
   </PropertyGroup>
 
   <ItemGroup>

Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs

@@ -1,12 +1,8 @@
 using System;
-using System.IO;
 using System.Net.WebSockets;
-using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
-using MQTTnet.Core.Channel;
 using MQTTnet.Core.Server;
-using MQTTnet.Implementations;
 
 namespace MQTTnet.AspNetCore
 {
@@ -47,47 +43,5 @@ namespace MQTTnet.AspNetCore
         {
             StopAsync();
         }
-
-        private class MqttWebSocketServerChannel : IMqttCommunicationChannel, IDisposable
-        {
-            private readonly WebSocket _webSocket;
-
-            public MqttWebSocketServerChannel(WebSocket webSocket)
-            {
-                _webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket));
-
-                RawReceiveStream = new WebSocketStream(_webSocket);
-            }
-
-            public Stream SendStream => RawReceiveStream;
-            public Stream ReceiveStream => RawReceiveStream;
-            public Stream RawReceiveStream { get; }
-
-            public Task ConnectAsync()
-            {
-                return Task.CompletedTask;
-            }
-
-            public Task DisconnectAsync()
-            {
-                RawReceiveStream?.Dispose();
-
-                if (_webSocket == null)
-                {
-                    return Task.CompletedTask;
-                }
-
-                return _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
-            }
-
-            public void Dispose()
-            {
-                RawReceiveStream?.Dispose();
-                SendStream?.Dispose();
-                ReceiveStream?.Dispose();
-
-                _webSocket?.Dispose();
-            }
-        }
     }
 }

Frameworks/MQTTnet.AspnetCore/MqttWebSocketServerChannel.cs

@@ -0,0 +1,60 @@
+using System;
+using System.IO;
+using System.Net.WebSockets;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Core.Channel;
+using MQTTnet.Implementations;
+
+namespace MQTTnet.AspNetCore
+{
+    public class MqttWebSocketServerChannel : IMqttCommunicationChannel, IDisposable
+    {
+        private WebSocket _webSocket;
+
+        public MqttWebSocketServerChannel(WebSocket webSocket)
+        {
+            _webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket));
+
+            SendStream = new WebSocketStream(_webSocket);
+            ReceiveStream = SendStream;
+        }
+
+        public Stream SendStream { get; private set; }
+        public Stream ReceiveStream { get; private set; }
+
+        public Task ConnectAsync()
+        {
+            return Task.CompletedTask;
+        }
+
+        public async Task DisconnectAsync()
+        {
+            if (_webSocket == null)
+            {
+                return;
+            }
+
+            try
+            {
+                await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
+            }
+            finally
+            {
+                Dispose();
+            }
+        }
+
+        public void Dispose()
+        {
+            SendStream?.Dispose();
+            ReceiveStream?.Dispose();
+
+            _webSocket?.Dispose();
+
+            SendStream = null;
+            ReceiveStream = null;
+            _webSocket = null;
+        }
+    }
+}

Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs

@@ -1,4 +1,4 @@
-#if NET452 || NETSTANDARD1_3 || NETSTANDARD2_0
+#if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
 using System;
 using System.Net;
 using System.Net.Security;
@@ -98,7 +98,7 @@ namespace MQTTnet.Implementations
                 try
                 {
                     //todo: else branch can be used with min dependency NET46
-#if NET452
+#if NET452 || NET461
                     var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false);
 #else
                     var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false);
@@ -122,7 +122,7 @@ namespace MQTTnet.Implementations
             {
                 try
                 {
-#if NET452
+#if NET452 || NET461
                     var clientSocket = await Task.Factory.FromAsync(_tlsEndpointSocket.BeginAccept, _tlsEndpointSocket.EndAccept, null).ConfigureAwait(false);
 #else
                     var clientSocket = await _tlsEndpointSocket.AcceptAsync().ConfigureAwait(false);

Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs

@@ -1,4 +1,4 @@
-#if NET452 || NETSTANDARD1_3 || NETSTANDARD2_0
+#if NET452 || NET461 || NETSTANDARD1_3 || NETSTANDARD2_0
 using System;
 using System.Net.Security;
 using System.Net.Sockets;
@@ -17,7 +17,7 @@ namespace MQTTnet.Implementations
         private readonly MqttClientTcpOptions _options;
 
         //todo: this can be used with min dependency NetStandard1.6
-#if NET452
+#if NET452 || NET461
         // ReSharper disable once MemberCanBePrivate.Global
         // ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
         public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user.
@@ -59,7 +59,7 @@ namespace MQTTnet.Implementations
             }
 
             //todo: else brach can be used with min dependency NET46
-#if NET452
+#if NET452 || NET461
             await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
 #else
             await _socket.ConnectAsync(_options.Server, _options.GetPort()).ConfigureAwait(false);
@@ -144,7 +144,7 @@ namespace MQTTnet.Implementations
             //need two streams otherwise read and write have to be synchronized
 
             //todo: if branch can be used with min dependency NetStandard1.6
-#if NET452
+#if NET452 || NET461
             SendStream = new BufferedStream(stream, BufferSize);
             ReceiveStream = new BufferedStream(stream, BufferSize);
 #else

Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs

@@ -11,6 +11,10 @@ namespace MQTTnet.Implementations
 {
     public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable
     {
+        // ReSharper disable once MemberCanBePrivate.Global
+        // ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
+        public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user.
+
         private readonly MqttClientWebSocketOptions _options;
         private ClientWebSocket _webSocket;
 
@@ -19,9 +23,8 @@ namespace MQTTnet.Implementations
             _options = options ?? throw new ArgumentNullException(nameof(options));
         }
 
-        public Stream SendStream => RawReceiveStream;
-        public Stream ReceiveStream => RawReceiveStream;
-        public Stream RawReceiveStream { get; private set; }
+        public Stream SendStream { get; private set; }
+        public Stream ReceiveStream { get; private set; }
 
         public async Task ConnectAsync()
         {
@@ -32,7 +35,7 @@ namespace MQTTnet.Implementations
             }
 
             _webSocket = new ClientWebSocket();
-
+            
             if (_options.RequestHeaders != null)
             {
                 foreach (var requestHeader in _options.RequestHeaders)
@@ -64,13 +67,13 @@ namespace MQTTnet.Implementations
             }
 
             await _webSocket.ConnectAsync(new Uri(uri), CancellationToken.None).ConfigureAwait(false);
-            RawReceiveStream = new WebSocketStream(_webSocket);
+
+            SendStream = new WebSocketStream(_webSocket);
+            ReceiveStream = SendStream;
         }
 
         public async Task DisconnectAsync()
         {
-            RawReceiveStream = null;
-
             if (_webSocket == null)
             {
                 return;
@@ -80,6 +83,8 @@ namespace MQTTnet.Implementations
             {
                 await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
             }
+
+            Dispose();
         }
 
         public void Dispose()

Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs

@@ -1,5 +1,7 @@
 using System;
+using System.Collections.Generic;
 using System.IO;
+using System.Linq;
 using System.Net.WebSockets;
 using System.Threading;
 using System.Threading.Tasks;
@@ -10,7 +12,9 @@ namespace MQTTnet.Implementations
     public class WebSocketStream : Stream
     {
         private readonly WebSocket _webSocket;
-        
+        private readonly byte[] _chunkBuffer = new byte[MqttWebSocketChannel.BufferSize];
+        private readonly Queue<byte> _buffer = new Queue<byte>(MqttWebSocketChannel.BufferSize);
+
         public WebSocketStream(WebSocket webSocket)
         {
             _webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket));
@@ -34,33 +38,9 @@ namespace MQTTnet.Implementations
         {
         }
 
-        public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        public override Task FlushAsync(CancellationToken cancellationToken)
         {
-            var currentOffset = offset;
-            var targetOffset = offset + count;
-            while (_webSocket.State == WebSocketState.Open && currentOffset < targetOffset)
-            {
-                var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, currentOffset, count), cancellationToken).ConfigureAwait(false);
-                currentOffset += response.Count;
-                count -= response.Count;
-
-                if (response.MessageType == WebSocketMessageType.Close)
-                {
-                    await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
-                }
-            }
-
-            if (_webSocket.State == WebSocketState.Closed)
-            {
-                throw new MqttCommunicationException( "connection closed" );
-            }
-
-            return currentOffset - offset;
-        }
-
-        public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
-        {
-            return _webSocket.SendAsync(new ArraySegment<byte>(buffer, offset, count), WebSocketMessageType.Binary, true, cancellationToken);
+            return Task.FromResult(0);
         }
 
         public override int Read(byte[] buffer, int offset, int count)
@@ -68,11 +48,58 @@ namespace MQTTnet.Implementations
             return ReadAsync(buffer, offset, count).GetAwaiter().GetResult();
         }
 
+        public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            var bytesRead = 0;
+
+            // Use existing date from buffer.
+            while (count > 0 && _buffer.Any())
+            {
+                buffer[offset++] = _buffer.Dequeue();
+                count--;
+                bytesRead++;
+            }
+
+            if (count == 0)
+            {
+                return bytesRead;
+            }
+            
+            while (_webSocket.State == WebSocketState.Open)
+            {
+                await FetchChunkAsync(cancellationToken);
+
+                while (count > 0 && _buffer.Any())
+                {
+                    buffer[offset++] = _buffer.Dequeue();
+                    count--;
+                    bytesRead++;
+                }
+
+                if (count == 0)
+                {
+                    return bytesRead;
+                }
+            }
+
+            if (_webSocket.State == WebSocketState.Closed)
+            {
+                throw new MqttCommunicationException("WebSocket connection closed.");
+            }
+
+            return bytesRead;
+        }
+
         public override void Write(byte[] buffer, int offset, int count)
         {
             WriteAsync(buffer, offset, count).GetAwaiter().GetResult();
         }
 
+        public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            return _webSocket.SendAsync(new ArraySegment<byte>(buffer, offset, count), WebSocketMessageType.Binary, true, cancellationToken);
+        }
+
         public override long Seek(long offset, SeekOrigin origin)
         {
             throw new NotSupportedException();
@@ -82,5 +109,21 @@ namespace MQTTnet.Implementations
         {
             throw new NotSupportedException();
         }
+
+        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);
+            }
+
+            if (response.MessageType == WebSocketMessageType.Close)
+            {
+                await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
+            }
+        }
     }
 }

Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj

@@ -1,11 +1,11 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard1.3;netstandard2.0;net452;uap10.0</TargetFrameworks>
+    <TargetFrameworks>netstandard1.3;netstandard2.0;net452;net461;uap10.0</TargetFrameworks>
     <AssemblyName>MQTTnet</AssemblyName>
     <RootNamespace>MQTTnet</RootNamespace>
-    <AssemblyVersion>2.5.0.0</AssemblyVersion>
-    <FileVersion>2.5.0.0</FileVersion>
+    <AssemblyVersion>2.5.2.0</AssemblyVersion>
+    <FileVersion>2.5.2.0</FileVersion>
     <Version>0.0.0.0</Version>
     <Company />
     <Product />
@@ -53,8 +53,8 @@
     <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
     <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
     <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="1.1.1" />
-    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="1.0.2" />
+    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.0.2" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
@@ -64,8 +64,14 @@
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)'=='net452'">
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="1.1.1" />
-    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="1.0.2" />
+    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.0.2" />
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)'=='net461'">
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="2.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging" Version="2.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Options" Version="2.0.0" />
   </ItemGroup>
 
 </Project>

Frameworks/MQTTnet.NetStandard/MqttFactory.cs

@@ -23,7 +23,7 @@ namespace MQTTnet
 
         public MqttFactory(IServiceProvider serviceProvider)
         {
-            _serviceProvider = serviceProvider;
+            _serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
         }
 
         public ILoggerFactory GetLoggerFactory()
@@ -85,9 +85,7 @@ namespace MQTTnet
                 clientSessionsManager,
                 _serviceProvider.GetRequiredService<MqttClientSubscriptionsManager>(),
                 _serviceProvider.GetRequiredService<ILogger<MqttClientSession>>(),
-                _serviceProvider.GetRequiredService<ILogger<MqttClientPendingMessagesQueue>>(),
-                _serviceProvider.GetRequiredService<IMqttClientRetainedMessageManager>()
-                );
+                _serviceProvider.GetRequiredService<ILogger<MqttClientPendingMessagesQueue>>());
         }
 
         public IMqttClient CreateMqttClient()

Frameworks/MQTTnet.NetStandard/TargetFrameworkInfoProvider.cs

@@ -0,0 +1,23 @@
+namespace MQTTnet
+{
+    public static class TargetFrameworkInfoProvider
+    {
+        public static string TargetFramework
+        {
+            get
+            {
+#if NET452
+                return "net452";
+#elif NET461
+                return "net461";
+#elif NETSTANDARD1_3
+                return "netstandard1.3";
+#elif NETSTANDARD2_0
+                return "netstandard2.0";
+#elif WINDOWS_UWP
+                return "uap10.0";
+#endif
+            }
+        }
+    }
+}

MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Channel;
@@ -14,10 +15,12 @@ namespace MQTTnet.Core.Adapter
 {
     public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
     {
+        private const uint ErrorOperationAborted = 0x800703E3;
+
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
         private readonly ILogger<MqttChannelCommunicationAdapter> _logger;
         private readonly IMqttCommunicationChannel _channel;
-        
+
         public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer, ILogger<MqttChannelCommunicationAdapter> logger)
         {
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -29,166 +32,91 @@ namespace MQTTnet.Core.Adapter
 
         public async Task ConnectAsync(TimeSpan timeout)
         {
-            try
-            {
-                await _channel.ConnectAsync().TimeoutAfter(timeout).ConfigureAwait(false);
-            }
-            catch (TaskCanceledException)
-            {
-                throw;
-            }
-            catch (OperationCanceledException)
-            {
-                throw;
-            }
-            catch (MqttCommunicationTimedOutException)
-            {
-                throw;
-            }
-            catch (MqttCommunicationException)
-            {
-                throw;
-            }
-            catch (Exception exception)
-            {
-                throw new MqttCommunicationException(exception);
-            }
+            _logger.LogInformation("Connecting [Timeout={0}]", timeout);
+
+            await ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout));
         }
 
         public async Task DisconnectAsync(TimeSpan timeout)
         {
-            try
-            {
-                await _channel.DisconnectAsync().TimeoutAfter(timeout).ConfigureAwait(false);
-            }
-            catch (TaskCanceledException)
-            {
-                throw;
-            }
-            catch (OperationCanceledException)
-            {
-                throw;
-            }
-            catch (MqttCommunicationTimedOutException)
-            {
-                throw;
-            }
-            catch (MqttCommunicationException)
-            {
-                throw;
-            }
-            catch (Exception exception)
-            {
-                throw new MqttCommunicationException(exception);
-            }
+            _logger.LogInformation("Disconnecting [Timeout={0}]", timeout);
+
+            await ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout));
         }
 
         public async Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable<MqttBasePacket> packets)
         {
-            try
+            await ExecuteAndWrapExceptionAsync(async () =>
             {
                 await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
-
-                foreach (var packet in packets)
+                try
                 {
-                    if (packet == null)
+                    foreach (var packet in packets)
                     {
-                        continue;
+                        if (packet == null)
+                        {
+                            continue;
+                        }
+
+                        _logger.LogInformation("TX >>> {0} [Timeout={1}]", packet, timeout);
+
+                        var writeBuffer = PacketSerializer.Serialize(packet);
+                        await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false);
                     }
 
-                    _logger.LogInformation("TX >>> {0} [Timeout={1}]", packet, timeout);
-
-                    var writeBuffer = PacketSerializer.Serialize(packet);
-                    await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false);
+                    if (timeout > TimeSpan.Zero)
+                    {
+                        await _channel.SendStream.FlushAsync(cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        await _channel.SendStream.FlushAsync(cancellationToken).ConfigureAwait(false);
+                    }
                 }
-
-                if (timeout > TimeSpan.Zero)
+                finally
                 {
-                    await _channel.SendStream.FlushAsync(cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
+                    _semaphore.Release();
                 }
-                else
-                {
-                    await _channel.SendStream.FlushAsync(cancellationToken).ConfigureAwait(false);
-                }
-            }
-            catch (TaskCanceledException)
-            {
-                throw;
-            }
-            catch (OperationCanceledException)
-            {
-                throw;
-            }
-            catch (MqttCommunicationTimedOutException)
-            {
-                throw;
-            }
-            catch (MqttCommunicationException)
-            {
-                throw;
-            }
-            catch (Exception exception)
-            {
-                throw new MqttCommunicationException(exception);
-            }
-            finally
-            {
-                _semaphore.Release();
-            }
+            });
         }
 
         public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
-            ReceivedMqttPacket receivedMqttPacket = null;
-            try
+            MqttBasePacket packet = null;
+            await ExecuteAndWrapExceptionAsync(async () =>
             {
-                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();
+                    }
 
-                var packet = PacketSerializer.Deserialize(receivedMqttPacket);
-                if (packet == null)
-                {
-                    throw new MqttProtocolViolationException("Received malformed packet.");
-                }
+                    packet = PacketSerializer.Deserialize(receivedMqttPacket);
+                    if (packet == null)
+                    {
+                        throw new MqttProtocolViolationException("Received malformed packet.");
+                    }
 
-                _logger.LogInformation("RX <<< {0}", packet);
-                return packet;
-            }
-            catch (TaskCanceledException)
-            {
-                throw;
-            }
-            catch (OperationCanceledException)
-            {
-                throw;
-            }
-            catch (MqttCommunicationTimedOutException)
-            {
-                throw;
-            }
-            catch (MqttCommunicationException)
-            {
-                throw;
-            }
-            catch (Exception exception)
-            {
-                throw new MqttCommunicationException(exception);
-            }
-            finally
-            {
-                receivedMqttPacket?.Dispose();
-            }
+                    _logger.LogInformation("RX <<< {0}", packet);
+                }
+                finally
+                {
+                    receivedMqttPacket?.Dispose();
+                }
+            });
+
+            return packet;
         }
 
         private static async Task<ReceivedMqttPacket> ReceiveAsync(Stream stream, CancellationToken cancellationToken)
@@ -215,5 +143,42 @@ namespace MQTTnet.Core.Adapter
 
             return new ReceivedMqttPacket(header, new MemoryStream(body, 0, body.Length));
         }
+
+        private static async Task ExecuteAndWrapExceptionAsync(Func<Task> action)
+        {
+            try
+            {
+                await action().ConfigureAwait(false);
+            }
+            catch (TaskCanceledException)
+            {
+                throw;
+            }
+            catch (OperationCanceledException)
+            {
+                throw;
+            }
+            catch (MqttCommunicationTimedOutException)
+            {
+                throw;
+            }
+            catch (MqttCommunicationException)
+            {
+                throw;
+            }
+            catch (COMException comException)
+            {
+                if ((uint)comException.HResult == ErrorOperationAborted)
+                {
+                    throw new OperationCanceledException();
+                }
+
+                throw new MqttCommunicationException(comException);
+            }
+            catch (Exception exception)
+            {
+                throw new MqttCommunicationException(exception);
+            }
+        }
     }
 }

MQTTnet.Core/Client/MqttClient.cs

@@ -60,14 +60,14 @@ namespace MQTTnet.Core.Client
                 await _adapter.ConnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
                 _logger.LogTrace("Connection with server established.");
 
-                await SetupIncomingPacketProcessingAsync().ConfigureAwait(false);
+                await StartReceivingPacketsAsync(_cancellationTokenSource.Token).ConfigureAwait(false);
                 var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false);
 
                 _logger.LogTrace("MQTT connection with server established.");
 
                 if (_options.KeepAlivePeriod != TimeSpan.Zero)
                 {
-                    StartSendKeepAliveMessages(_cancellationTokenSource.Token);
+                    StartSendingKeepAliveMessages(_cancellationTokenSource.Token);
                 }
 
                 IsConnected = true;
@@ -96,7 +96,6 @@ namespace MQTTnet.Core.Client
             finally
             {
                 await DisconnectInternalAsync().ConfigureAwait(false);
-                _scopeHandle.Dispose();
             }
         }
 
@@ -149,36 +148,36 @@ namespace MQTTnet.Core.Client
                 switch (qosGroup.Key)
                 {
                     case MqttQualityOfServiceLevel.AtMostOnce:
-                    {
-                        // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
-                        await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets).ConfigureAwait(false);
-                        break;
-                    }
+                        {
+                            // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
+                            await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets).ConfigureAwait(false);
+                            break;
+                        }
                     case MqttQualityOfServiceLevel.AtLeastOnce:
-                    {
-                        foreach (var publishPacket in qosPackets)
                         {
-                            publishPacket.PacketIdentifier = GetNewPacketIdentifier();
-                            await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket).ConfigureAwait(false);
-                        }
+                            foreach (var publishPacket in qosPackets)
+                            {
+                                publishPacket.PacketIdentifier = GetNewPacketIdentifier();
+                                await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket).ConfigureAwait(false);
+                            }
 
-                        break;
-                    }
+                            break;
+                        }
                     case MqttQualityOfServiceLevel.ExactlyOnce:
-                    {
-                        foreach (var publishPacket in qosPackets)
                         {
-                            publishPacket.PacketIdentifier = GetNewPacketIdentifier();
-                            var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
-                            await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).ConfigureAwait(false);
-                        }
+                            foreach (var publishPacket in qosPackets)
+                            {
+                                publishPacket.PacketIdentifier = GetNewPacketIdentifier();
+                                var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
+                                await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).ConfigureAwait(false);
+                            }
 
-                        break;
-                    }
+                            break;
+                        }
                     default:
-                    {
-                        throw new InvalidOperationException();
-                    }
+                        {
+                            throw new InvalidOperationException();
+                        }
                 }
             }
         }
@@ -191,7 +190,7 @@ namespace MQTTnet.Core.Client
                 Username = _options.Credentials?.Username,
                 Password = _options.Credentials?.Password,
                 CleanSession = _options.CleanSession,
-                KeepAlivePeriod = (ushort) _options.KeepAlivePeriod.TotalSeconds,
+                KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds,
                 WillMessage = willApplicationMessage
             };
 
@@ -204,24 +203,6 @@ namespace MQTTnet.Core.Client
             return response;
         }
 
-        private async Task SetupIncomingPacketProcessingAsync()
-        {
-            _isReceivingPackets = false;
-
-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-            Task.Factory.StartNew(
-                () => ReceivePackets(_cancellationTokenSource.Token),
-                _cancellationTokenSource.Token,
-                TaskCreationOptions.LongRunning,
-                TaskScheduler.Default).ConfigureAwait(false);
-#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-
-            while (!_isReceivingPackets && _cancellationTokenSource != null && !_cancellationTokenSource.IsCancellationRequested)
-            {
-                await Task.Delay(TimeSpan.FromMilliseconds(100)).ConfigureAwait(false);
-            }
-        }
-
         private void ThrowIfNotConnected()
         {
             if (!IsConnected) throw new MqttCommunicationException("The client is not connected.");
@@ -234,6 +215,8 @@ namespace MQTTnet.Core.Client
 
         private async Task DisconnectInternalAsync()
         {
+            _scopeHandle?.Dispose();
+
             var clientWasConnected = IsConnected;
             IsConnected = false;
 
@@ -258,6 +241,7 @@ namespace MQTTnet.Core.Client
             }
             finally
             {
+                _logger.LogInformation("Disconnected.");
                 Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected));
             }
         }
@@ -324,7 +308,7 @@ namespace MQTTnet.Core.Client
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
                 FireApplicationMessageReceivedEvent(publishPacket);
-                await SendAsync(new MqttPubAckPacket {PacketIdentifier = publishPacket.PacketIdentifier});
+                await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
                 return;
             }
 
@@ -337,7 +321,7 @@ namespace MQTTnet.Core.Client
                 }
 
                 FireApplicationMessageReceivedEvent(publishPacket);
-                await SendAsync(new MqttPubRecPacket {PacketIdentifier = publishPacket.PacketIdentifier});
+                await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
                 return;
             }
 
@@ -363,12 +347,12 @@ namespace MQTTnet.Core.Client
         {
             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);
+            return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
         }
 
         private ushort GetNewPacketIdentifier()
         {
-            return (ushort) Interlocked.Increment(ref _latestPacketIdentifier);
+            return (ushort)Interlocked.Increment(ref _latestPacketIdentifier);
         }
 
         private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
@@ -390,6 +374,12 @@ namespace MQTTnet.Core.Client
             }
             catch (OperationCanceledException)
             {
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return;
+                }
+
+                await DisconnectInternalAsync().ConfigureAwait(false);
             }
             catch (MqttCommunicationException exception)
             {
@@ -412,7 +402,7 @@ namespace MQTTnet.Core.Client
             }
         }
 
-        private async Task ReceivePackets(CancellationToken cancellationToken)
+        private async Task ReceivePacketsAsync(CancellationToken cancellationToken)
         {
             _logger.LogInformation("Start receiving packets.");
 
@@ -423,6 +413,7 @@ namespace MQTTnet.Core.Client
                     _isReceivingPackets = true;
 
                     var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
+
                     if (cancellationToken.IsCancellationRequested)
                     {
                         return;
@@ -433,6 +424,12 @@ namespace MQTTnet.Core.Client
             }
             catch (OperationCanceledException)
             {
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return;
+                }
+
+                await DisconnectInternalAsync().ConfigureAwait(false);
             }
             catch (MqttCommunicationException exception)
             {
@@ -458,15 +455,34 @@ namespace MQTTnet.Core.Client
         private void StartProcessReceivedPacket(MqttBasePacket packet, CancellationToken cancellationToken)
         {
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-            Task.Run(() => ProcessReceivedPacketAsync(packet), cancellationToken).ConfigureAwait(false);
+            Task.Run(
+                () => ProcessReceivedPacketAsync(packet),
+                cancellationToken).ConfigureAwait(false);
 #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
         }
 
-        private void StartSendKeepAliveMessages(CancellationToken cancellationToken)
+        private async Task StartReceivingPacketsAsync(CancellationToken cancellationToken)
+        {
+            _isReceivingPackets = false;
+
+#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            Task.Run(
+                async () => await ReceivePacketsAsync(cancellationToken),
+                cancellationToken).ConfigureAwait(false);
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+
+            while (!_isReceivingPackets && !cancellationToken.IsCancellationRequested)
+            {
+                await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken).ConfigureAwait(false);
+            }
+        }
+
+        private void StartSendingKeepAliveMessages(CancellationToken cancellationToken)
         {
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-            Task.Factory.StartNew(() => SendKeepAliveMessagesAsync(cancellationToken), cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default)
-                .ConfigureAwait(false);
+            Task.Run(
+                async () => await SendKeepAliveMessagesAsync(cancellationToken), 
+                cancellationToken).ConfigureAwait(false);
 #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
         }
     }

MQTTnet.Core/MQTTnet.Core.csproj

@@ -17,14 +17,14 @@
     <PackageIconUrl></PackageIconUrl>
     <RepositoryUrl></RepositoryUrl>
     <PackageTags></PackageTags>
-    <FileVersion>2.5.0.0</FileVersion>
-    <AssemblyVersion>2.5.0.0</AssemblyVersion>
+    <FileVersion>2.5.2.0</FileVersion>
+    <AssemblyVersion>2.5.2.0</AssemblyVersion>
     <PackageLicenseUrl></PackageLicenseUrl>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="1.1.2" />
-    <PackageReference Include="Microsoft.Extensions.Options" Version="1.1.2" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="1.0.2" />
+    <PackageReference Include="Microsoft.Extensions.Options" Version="1.0.2" />
     <PackageReference Include="System.Threading" Version="4.3.0" />
   </ItemGroup>
 

MQTTnet.Core/Packets/MqttConnAckPacket.cs

@@ -10,7 +10,7 @@ namespace MQTTnet.Core.Packets
 
         public override string ToString()
         {
-            return nameof(MqttConnAckPacket) + ": [ConnectReturnCode=" + ConnectReturnCode + "] [IsSessionPresent=" + IsSessionPresent + "]";
+            return "ConnAck: [ConnectReturnCode=" + ConnectReturnCode + "] [IsSessionPresent=" + IsSessionPresent + "]";
         }
     }
 }

MQTTnet.Core/Packets/MqttConnectPacket.cs

@@ -20,7 +20,7 @@ namespace MQTTnet.Core.Packets
 
         public override string ToString()
         {
-            return nameof(MqttConnectPacket) + ": [ClientId=" + ClientId + "] [Username=" + Username + "] [Password=" + Password + "] [KeepAlivePeriod=" + KeepAlivePeriod + "] [CleanSession=" + CleanSession + "]";
+            return "Connect: [ClientId=" + ClientId + "] [Username=" + Username + "] [Password=" + Password + "] [KeepAlivePeriod=" + KeepAlivePeriod + "] [CleanSession=" + CleanSession + "]";
         }
     }
 }

MQTTnet.Core/Packets/MqttDisconnectPacket.cs

@@ -2,5 +2,9 @@
 {
     public sealed class MqttDisconnectPacket : MqttBasePacket
     {
+        public override string ToString()
+        {
+            return "Disconnect";
+        }
     }
 }

MQTTnet.Core/Packets/MqttPingReqPacket.cs

@@ -4,7 +4,7 @@
     {
         public override string ToString()
         {
-            return nameof(MqttPingReqPacket);
+            return "PingReq";
         }
     }
 }

MQTTnet.Core/Packets/MqttPingRespPacket.cs

@@ -4,7 +4,7 @@
     {
         public override string ToString()
         {
-            return nameof(MqttPingRespPacket);
+            return "PingResp";
         }
     }
 }

MQTTnet.Core/Packets/MqttPubAckPacket.cs

@@ -2,5 +2,9 @@
 {
     public sealed class MqttPubAckPacket : MqttBasePublishPacket
     {
+        public override string ToString()
+        {
+            return "PubAck";
+        }
     }
 }

MQTTnet.Core/Packets/MqttPubCompPacket.cs

@@ -2,5 +2,9 @@
 {
     public sealed class MqttPubCompPacket : MqttBasePublishPacket
     {
+        public override string ToString()
+        {
+            return "PubComp";
+        }
     }
 }

MQTTnet.Core/Packets/MqttPubRecPacket.cs

@@ -2,5 +2,9 @@
 {
     public sealed class MqttPubRecPacket : MqttBasePublishPacket
     {
+        public override string ToString()
+        {
+            return "PubRec";
+        }
     }
 }

MQTTnet.Core/Packets/MqttPubRelPacket.cs

@@ -2,5 +2,9 @@
 {
     public sealed class MqttPubRelPacket : MqttBasePublishPacket
     {
+        public override string ToString()
+        {
+            return "PubRel";
+        }
     }
 }

MQTTnet.Core/Packets/MqttPublishPacket.cs

@@ -1,5 +1,4 @@
-using System;
-using MQTTnet.Core.Protocol;
+using MQTTnet.Core.Protocol;
 
 namespace MQTTnet.Core.Packets
 {
@@ -17,9 +16,8 @@ namespace MQTTnet.Core.Packets
 
         public override string ToString()
         {
-            return nameof(MqttPublishPacket) +
-                ": [Topic=" + Topic + "]" +
-                " [Payload=" + Convert.ToBase64String(Payload) + "]" +
+            return "Publish: [Topic=" + Topic + "]" +
+                " [Payload.Length=" + Payload?.Length + "]" +
                 " [QoSLevel=" + QualityOfServiceLevel + "]" +
                 " [Dup=" + Dup + "]" +
                 " [Retain=" + Retain + "]" +

MQTTnet.Core/Packets/MqttSubAckPacket.cs

@@ -13,7 +13,7 @@ namespace MQTTnet.Core.Packets
         public override string ToString()
         {
             var subscribeReturnCodesText = string.Join(",", SubscribeReturnCodes.Select(f => f.ToString()));
-            return nameof(MqttSubAckPacket) + ": [PacketIdentifier=" + PacketIdentifier + "] [SubscribeReturnCodes=" + subscribeReturnCodesText + "]";
+            return "SubAck: [PacketIdentifier=" + PacketIdentifier + "] [SubscribeReturnCodes=" + subscribeReturnCodesText + "]";
         }
     }
 }

MQTTnet.Core/Packets/MqttSubscribePacket.cs

@@ -12,7 +12,7 @@ namespace MQTTnet.Core.Packets
         public override string ToString()
         {
             var topicFiltersText = string.Join(",", TopicFilters.Select(f => f.Topic + "@" + f.QualityOfServiceLevel));
-            return nameof(MqttSubscribePacket) + ": [PacketIdentifier=" + PacketIdentifier + "] [TopicFilters=" + topicFiltersText + "]";
+            return "Subscribe: [PacketIdentifier=" + PacketIdentifier + "] [TopicFilters=" + topicFiltersText + "]";
         }
     }
 }

MQTTnet.Core/Packets/MqttUnsubAckPacket.cs

@@ -3,5 +3,10 @@
     public sealed class MqttUnsubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
+
+        public override string ToString()
+        {
+            return "UnsubAck: [PacketIdentifier=" + PacketIdentifier + "]";
+        }
     }
 }

MQTTnet.Core/Packets/MqttUnsubscribe.cs

@@ -7,5 +7,11 @@ namespace MQTTnet.Core.Packets
         public ushort PacketIdentifier { get; set; }
 
         public IList<string> TopicFilters { get; set; } = new List<string>();
+
+        public override string ToString()
+        {
+            var topicFiltersText = string.Join(",", TopicFilters);
+            return "Subscribe: [PacketIdentifier=" + PacketIdentifier + "] [TopicFilters=" + topicFiltersText + "]";
+        }
     }
 }

MQTTnet.Core/Serializer/MqttPacketReader.cs

@@ -91,7 +91,7 @@ namespace MQTTnet.Core.Serializer
 
                 if (buffer == -1)
                 {
-                    break;
+                    throw new MqttCommunicationException("Connection closed while reading remaining length data.");
                 }
 
                 encodedByte = (byte)buffer;

MQTTnet.Core/Server/CheckSubscriptionsResult.cs

@@ -0,0 +1,11 @@
+using MQTTnet.Core.Protocol;
+
+namespace MQTTnet.Core.Server
+{
+    public class CheckSubscriptionsResult
+    {
+        public bool IsSubscribed { get; set; }
+
+        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
+    }
+}

MQTTnet.Core/Server/IMqttServer.cs

@@ -10,8 +10,7 @@ namespace MQTTnet.Core.Server
         event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
         event EventHandler<MqttServerStartedEventArgs> Started;
 
-        IList<ConnectedMqttClient> GetConnectedClients();
-        void Publish(IEnumerable<MqttApplicationMessage> applicationMessages);
+        Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync();
 
         Task StartAsync();
         Task StopAsync();

MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs

@@ -96,7 +96,7 @@ namespace MQTTnet.Core.Server
                     _pendingPublishPackets.Add(packet, CancellationToken.None);
                 }
 
-                _session.Stop();
+                await _session.StopAsync();
             }
         }
     }

MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs

@@ -57,42 +57,7 @@ namespace MQTTnet.Core.Server
             await _gate.WaitAsync().ConfigureAwait(false);
             try
             {
-                var saveIsRequired = false;
-
-                if (applicationMessage.Payload?.Any() == false)
-                {
-                    saveIsRequired = _retainedMessages.Remove(applicationMessage.Topic);
-                    _logger.LogInformation("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
-                }
-                else
-                {
-                    if (!_retainedMessages.ContainsKey(applicationMessage.Topic))
-                    {
-                        _retainedMessages[applicationMessage.Topic] = applicationMessage;
-                        saveIsRequired = true;
-                    }
-                    else
-                    {
-                        var existingMessage = _retainedMessages[applicationMessage.Topic];
-                        if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
-                        {
-                            _retainedMessages[applicationMessage.Topic] = applicationMessage;
-                            saveIsRequired = true;
-                        }
-                    }
-                    
-                    _logger.LogInformation("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
-                }
-
-                if (!saveIsRequired)
-                {
-                    _logger.LogTrace("Skipped saving retained messages because no changes were detected.");
-                }
-
-                if (saveIsRequired && _options.Storage != null)
-                {
-                    await _options.Storage.SaveRetainedMessagesAsync(_retainedMessages.Values.ToList());
-                }
+                await HandleMessageInternalAsync(clientId, applicationMessage);
             }
             catch (Exception exception)
             {
@@ -137,5 +102,45 @@ namespace MQTTnet.Core.Server
 
             return retainedMessages;
         }
+
+        private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage)
+        {
+            var saveIsRequired = false;
+
+            if (applicationMessage.Payload?.Any() == false)
+            {
+                saveIsRequired = _retainedMessages.Remove(applicationMessage.Topic);
+                _logger.LogInformation("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
+            }
+            else
+            {
+                if (!_retainedMessages.ContainsKey(applicationMessage.Topic))
+                {
+                    _retainedMessages[applicationMessage.Topic] = applicationMessage;
+                    saveIsRequired = true;
+                }
+                else
+                {
+                    var existingMessage = _retainedMessages[applicationMessage.Topic];
+                    if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
+                    {
+                        _retainedMessages[applicationMessage.Topic] = applicationMessage;
+                        saveIsRequired = true;
+                    }
+                }
+
+                _logger.LogInformation("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
+            }
+
+            if (!saveIsRequired)
+            {
+                _logger.LogTrace("Skipped saving retained messages because no changes were detected.");
+            }
+
+            if (saveIsRequired && _options.Storage != null)
+            {
+                await _options.Storage.SaveRetainedMessagesAsync(_retainedMessages.Values.ToList());
+            }
+        }
     }
 }

MQTTnet.Core/Server/MqttClientSession.cs

@@ -17,7 +17,6 @@ namespace MQTTnet.Core.Server
     {
         private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
 
-        private readonly IMqttClientRetainedMessageManager _clientRetainedMessageManager;
         private readonly MqttClientSubscriptionsManager _subscriptionsManager;
         private readonly MqttClientSessionsManager _sessionsManager;
         private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
@@ -34,10 +33,8 @@ namespace MQTTnet.Core.Server
             MqttClientSessionsManager sessionsManager,
             MqttClientSubscriptionsManager subscriptionsManager,
             ILogger<MqttClientSession> logger,
-            ILogger<MqttClientPendingMessagesQueue> messageQueueLogger,
-            IMqttClientRetainedMessageManager clientRetainedMessageManager)
+            ILogger<MqttClientPendingMessagesQueue> messageQueueLogger)
         {
-            _clientRetainedMessageManager = clientRetainedMessageManager ?? throw new ArgumentNullException(nameof(clientRetainedMessageManager));
             _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
             _subscriptionsManager = subscriptionsManager ?? throw new ArgumentNullException(nameof(subscriptionsManager));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -82,7 +79,7 @@ namespace MQTTnet.Core.Server
             }
         }
 
-        public void Stop()
+        public async Task StopAsync()
         {
             try
             {
@@ -90,17 +87,21 @@ namespace MQTTnet.Core.Server
                 _cancellationTokenSource?.Dispose();
                 _cancellationTokenSource = null;
 
-                _adapter = null;
+                if (_adapter != null)
+                {
+                    await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
+                    _adapter = null;
+                }
 
-                _logger.LogInformation("Client '{0}': Disconnected.", ClientId);
+                _logger.LogInformation("Client '{0}': Session stopped.", ClientId);
             }
             finally
             {
-                var willMessage = _willMessage; 
+                var willMessage = _willMessage;
                 if (willMessage != null)
                 {
                     _willMessage = null; //clear willmessage so it is send just once
-                    _sessionsManager.DispatchApplicationMessage(this, willMessage);
+                    await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage);
                 }
             }
         }
@@ -109,11 +110,13 @@ namespace MQTTnet.Core.Server
         {
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
-            if (!_subscriptionsManager.IsSubscribed(publishPacket))
+            var result = _subscriptionsManager.CheckSubscriptions(publishPacket);
+            if (!result.IsSubscribed)
             {
                 return;
             }
 
+            publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
             _pendingMessagesQueue.Enqueue(publishPacket);
         }
 
@@ -133,12 +136,12 @@ namespace MQTTnet.Core.Server
             catch (MqttCommunicationException exception)
             {
                 _logger.LogWarning(new EventId(), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
-                Stop();
+                await StopAsync();
             }
             catch (Exception exception)
             {
                 _logger.LogError(new EventId(), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
-                Stop();
+                await StopAsync();
             }
         }
 
@@ -182,14 +185,11 @@ namespace MQTTnet.Core.Server
 
             if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
             {
-                Stop();
-                return Task.FromResult(0);
+                return StopAsync();
             }
 
             _logger.LogWarning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
-            Stop();
-
-            return Task.FromResult(0);
+            return StopAsync();            
         }
 
         private async Task HandleIncomingSubscribePacketAsync(IMqttCommunicationAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
@@ -202,13 +202,13 @@ namespace MQTTnet.Core.Server
             if (subscribeResult.CloseConnection)
             {
                 await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false);
-                Stop();
+                await StopAsync();
             }
         }
 
         private async Task EnqueueSubscribedRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
         {
-            var retainedMessages = await _clientRetainedMessageManager.GetSubscribedMessagesAsync(subscribePacket).ConfigureAwait(false);
+            var retainedMessages = await _sessionsManager.GetRetainedMessagesAsync(subscribePacket).ConfigureAwait(false);
             foreach (var publishPacket in retainedMessages)
             {
                 EnqueuePublishPacket(publishPacket.ToPublishPacket());
@@ -219,29 +219,16 @@ namespace MQTTnet.Core.Server
         {
             var applicationMessage = publishPacket.ToApplicationMessage();
 
-            var interceptorContext = new MqttApplicationMessageInterceptorContext
-            {
-                ApplicationMessage = applicationMessage
-            };
-
-            _options.ApplicationMessageInterceptor?.Invoke(interceptorContext);
-            applicationMessage = interceptorContext.ApplicationMessage;
-
-            if (applicationMessage.Retain)
-            {
-                await _clientRetainedMessageManager.HandleMessageAsync(ClientId, applicationMessage).ConfigureAwait(false);
-            }
-
             switch (applicationMessage.QualityOfServiceLevel)
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
                     {
-                        _sessionsManager.DispatchApplicationMessage(this, applicationMessage);
+                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
                         return;
                     }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                     {
-                        _sessionsManager.DispatchApplicationMessage(this, applicationMessage);
+                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
 
                         await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken,
                             new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
@@ -256,7 +243,7 @@ namespace MQTTnet.Core.Server
                             _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
                         }
 
-                        _sessionsManager.DispatchApplicationMessage(this, applicationMessage);
+                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
 
                         await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken,
                             new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false);

MQTTnet.Core/Server/MqttClientSessionsManager.cs

@@ -16,19 +16,24 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSessionsManager
     {
-        private readonly Dictionary<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>();
+        private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
+        private readonly SemaphoreSlim _sessionsSemaphore = new SemaphoreSlim(1, 1);
+
+        private readonly MqttServerOptions _options;
         private readonly ILogger<MqttClientSessionsManager> _logger;
         private readonly IMqttClientSesssionFactory _clientSesssionFactory;
-        private readonly MqttServerOptions _options;
+        private readonly IMqttClientRetainedMessageManager _clientRetainedMessageManager;
 
         public MqttClientSessionsManager(
-            IOptions<MqttServerOptions> options, 
+            IOptions<MqttServerOptions> options,
             ILogger<MqttClientSessionsManager> logger,
-            IMqttClientSesssionFactory clientSesssionFactory)
+            IMqttClientSesssionFactory clientSesssionFactory,
+            IMqttClientRetainedMessageManager clientRetainedMessageManager)
         {
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
             _options = options.Value ?? throw new ArgumentNullException(nameof(options));
             _clientSesssionFactory = clientSesssionFactory ?? throw new ArgumentNullException(nameof(clientSesssionFactory));
+            _clientRetainedMessageManager = clientRetainedMessageManager ?? throw new ArgumentNullException(nameof(clientRetainedMessageManager));
         }
 
         public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
@@ -61,7 +66,7 @@ namespace MQTTnet.Core.Server
                     return;
                 }
 
-                var clientSession = GetOrCreateClientSession(connectPacket);
+                var clientSession = await GetOrCreateClientSessionAsync(connectPacket);
 
                 await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
                 {
@@ -92,7 +97,7 @@ namespace MQTTnet.Core.Server
                 }
                 catch (Exception)
                 {
-                    //ignored
+                    // ignored
                 }
 
                 ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(new ConnectedMqttClient
@@ -103,30 +108,58 @@ namespace MQTTnet.Core.Server
             }
         }
 
-        public void Clear()
+        public async Task StopAsync()
         {
-            lock (_clientSessions)
+            await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
+            try
             {
-                _clientSessions.Clear();
+                foreach (var session in _sessions)
+                {
+                    await session.Value.StopAsync();
+                }
+
+                _sessions.Clear();
+            }
+            finally
+            {
+                _sessionsSemaphore.Release();
             }
         }
 
-        public IList<ConnectedMqttClient> GetConnectedClients()
+        public async Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
         {
-            lock (_clientSessions)
+            await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
+            try
             {
-                return _clientSessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
+                return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
                 {
                     ClientId = s.Value.ClientId,
                     ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311
                 }).ToList();
             }
+            finally
+            {
+                _sessionsSemaphore.Release();
+            }
         }
 
-        public void DispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        public async Task DispatchApplicationMessageAsync(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
         {
             try
             {
+                var interceptorContext = new MqttApplicationMessageInterceptorContext
+                {
+                    ApplicationMessage = applicationMessage
+                };
+
+                _options.ApplicationMessageInterceptor?.Invoke(interceptorContext);
+                applicationMessage = interceptorContext.ApplicationMessage;
+
+                if (applicationMessage.Retain)
+                {
+                    await _clientRetainedMessageManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
+                }
+
                 var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, applicationMessage);
                 ApplicationMessageReceived?.Invoke(this, eventArgs);
             }
@@ -135,15 +168,20 @@ namespace MQTTnet.Core.Server
                 _logger.LogError(new EventId(), exception, "Error while processing application message");
             }
 
-            lock (_clientSessions)
+            lock (_sessions)
             {
-                foreach (var clientSession in _clientSessions.Values.ToList())
+                foreach (var clientSession in _sessions.Values.ToList())
                 {
                     clientSession.EnqueuePublishPacket(applicationMessage.ToPublishPacket());
                 }
             }
         }
 
+        public Task<List<MqttApplicationMessage>> GetRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
+        {
+            return _clientRetainedMessageManager.GetSubscribedMessagesAsync(subscribePacket);
+        }
+
         private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
         {
             if (_options.ConnectionValidator != null)
@@ -154,17 +192,18 @@ namespace MQTTnet.Core.Server
             return MqttConnectReturnCode.ConnectionAccepted;
         }
 
-        private GetOrCreateClientSessionResult GetOrCreateClientSession(MqttConnectPacket connectPacket)
+        private async Task<GetOrCreateClientSessionResult> GetOrCreateClientSessionAsync(MqttConnectPacket connectPacket)
         {
-            lock (_clientSessions)
+            await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
+            try
             {
-                var isSessionPresent = _clientSessions.TryGetValue(connectPacket.ClientId, out var clientSession);
+                var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
                 if (isSessionPresent)
                 {
                     if (connectPacket.CleanSession)
                     {
-                        _clientSessions.Remove(connectPacket.ClientId);
-                        clientSession.Stop();
+                        _sessions.Remove(connectPacket.ClientId);
+                        await clientSession.StopAsync();
                         clientSession = null;
 
                         _logger.LogTrace("Stopped existing session of client '{0}'.", connectPacket.ClientId);
@@ -181,13 +220,17 @@ namespace MQTTnet.Core.Server
                     isExistingSession = false;
 
                     clientSession = _clientSesssionFactory.CreateClientSession(connectPacket.ClientId, this);
-                    _clientSessions[connectPacket.ClientId] = clientSession;
+                    _sessions[connectPacket.ClientId] = clientSession;
 
                     _logger.LogTrace("Created a new session for client '{0}'.", connectPacket.ClientId);
                 }
 
                 return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
             }
+            finally
+            {
+                _sessionsSemaphore.Release();
+            }
         }
     }
 }

MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs

@@ -8,7 +8,7 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSubscriptionsManager
     {
-        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscribedTopics = new Dictionary<string, MqttQualityOfServiceLevel>();
+        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly MqttServerOptions _options;
 
         public MqttClientSubscriptionsManager(IOptions<MqttServerOptions> options)
@@ -23,7 +23,7 @@ namespace MQTTnet.Core.Server
             var responsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>();
             var closeConnection = false;
 
-            lock (_subscribedTopics)
+            lock (_subscriptions)
             {
                 foreach (var topicFilter in subscribePacket.TopicFilters)
                 {
@@ -38,7 +38,7 @@ namespace MQTTnet.Core.Server
 
                     if (interceptorContext.AcceptSubscription)
                     {
-                        _subscribedTopics[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                        _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
                     }
                 }
             }
@@ -54,40 +54,48 @@ namespace MQTTnet.Core.Server
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
 
-            lock (_subscribedTopics)
+            lock (_subscriptions)
             {
                 foreach (var topicFilter in unsubscribePacket.TopicFilters)
                 {
-                    _subscribedTopics.Remove(topicFilter);
+                    _subscriptions.Remove(topicFilter);
                 }
             }
 
             return unsubscribePacket.CreateResponse<MqttUnsubAckPacket>();
         }
 
-        public bool IsSubscribed(MqttPublishPacket publishPacket)
+        public CheckSubscriptionsResult CheckSubscriptions(MqttPublishPacket publishPacket)
         {
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
-            lock (_subscribedTopics)
+            lock (_subscriptions)
             {
-                foreach (var subscribedTopic in _subscribedTopics)
+                foreach (var subscription in _subscriptions)
                 {
-                    if (publishPacket.QualityOfServiceLevel > subscribedTopic.Value)
+                    if (!MqttTopicFilterComparer.IsMatch(publishPacket.Topic, subscription.Key))
                     {
                         continue;
                     }
 
-                    if (!MqttTopicFilterComparer.IsMatch(publishPacket.Topic, subscribedTopic.Key))
+                    var effectiveQos = subscription.Value;
+                    if (publishPacket.QualityOfServiceLevel < effectiveQos)
                     {
-                        continue;
+                        effectiveQos = publishPacket.QualityOfServiceLevel;
                     }
 
-                    return true;
+                    return new CheckSubscriptionsResult
+                    {
+                        IsSubscribed = true,
+                        QualityOfServiceLevel = effectiveQos
+                    };
                 }
             }
 
-            return false;
+            return new CheckSubscriptionsResult
+            {
+                IsSubscribed = false
+            };
         }
     }
 }

MQTTnet.Core/Server/MqttServer.cs

@@ -20,12 +20,11 @@ namespace MQTTnet.Core.Server
         private CancellationTokenSource _cancellationTokenSource;
 
         public MqttServer(
-            IOptions<MqttServerOptions> options, 
+            IOptions<MqttServerOptions> options,
             IEnumerable<IMqttServerAdapter> adapters,
             ILogger<MqttServer> logger,
             MqttClientSessionsManager clientSessionsManager,
-            IMqttClientRetainedMessageManager clientRetainedMessageManager
-            )
+            IMqttClientRetainedMessageManager clientRetainedMessageManager)
         {
             _options = options.Value ?? throw new ArgumentNullException(nameof(options));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -35,7 +34,7 @@ namespace MQTTnet.Core.Server
             if (adapters == null)
             {
                 throw new ArgumentNullException(nameof(adapters));
-            }            
+            }
 
             _adapters = adapters.ToList();
 
@@ -44,9 +43,9 @@ namespace MQTTnet.Core.Server
             _clientSessionsManager.ClientDisconnected += OnClientDisconnected;
         }
 
-        public IList<ConnectedMqttClient> GetConnectedClients()
+        public Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
         {
-            return _clientSessionsManager.GetConnectedClients();
+            return _clientSessionsManager.GetConnectedClientsAsync();
         }
 
         public event EventHandler<MqttServerStartedEventArgs> Started;
@@ -54,7 +53,7 @@ namespace MQTTnet.Core.Server
         public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
-        public void Publish(IEnumerable<MqttApplicationMessage> applicationMessages)
+        public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
         {
             if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
 
@@ -65,23 +64,10 @@ namespace MQTTnet.Core.Server
 
             foreach (var applicationMessage in applicationMessages)
             {
-                var interceptorContext = new MqttApplicationMessageInterceptorContext
-                {
-                    ApplicationMessage = applicationMessage
-                };
-
-                _options.ApplicationMessageInterceptor?.Invoke(interceptorContext);
-                
-                _clientSessionsManager.DispatchApplicationMessage(null, interceptorContext.ApplicationMessage);
+                await _clientSessionsManager.DispatchApplicationMessageAsync(null, applicationMessage);
             }
         }
 
-        public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
-        {
-            Publish(applicationMessages);
-            return Task.FromResult(0);
-        }
-
         public async Task StartAsync()
         {
             if (_cancellationTokenSource != null) throw new InvalidOperationException("The MQTT server is already started.");
@@ -113,7 +99,7 @@ namespace MQTTnet.Core.Server
                 await adapter.StopAsync();
             }
 
-            _clientSessionsManager.Clear();
+            await _clientSessionsManager.StopAsync();
 
             _logger.LogInformation("Stopped.");
         }

MQTTnet.sln

@@ -1,7 +1,7 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio 15
-VisualStudioVersion = 15.0.27004.2006
+VisualStudioVersion = 15.0.27004.2008
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Core.Tests", "Tests\MQTTnet.Core.Tests\MQTTnet.Core.Tests.csproj", "{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}"
 EndProject

README.md

@@ -77,8 +77,10 @@ If you want to contribute to this project just create a pull request. But only p
 This library is used in the following projects:
 
 * MQTT Client Rx (Wrapper for Reactive Extensions, <https://github.com/1iveowl/MQTTClient.rx>)
+* MQTT Tester (MQTT client test app for [Android](https://play.google.com/store/apps/details?id=com.liveowl.mqtttester), [iOS](https://itunes.apple.com/us/app/mqtt-tester/id1278621826?mt=8), and [Windows 10](https://www.microsoft.com/en-us/store/p/mqtt-tester/9wzdncrd272c))
 * Wirehome (Open Source Home Automation system for .NET, <https://github.com/chkr1011/Wirehome>)
 
+
 If you use this library and want to see your project here please let me know.
 
 ## MIT License
@@ -101,4 +103,4 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
+SOFTWARE.

Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs

@@ -15,7 +15,7 @@ namespace MQTTnet.Core.Tests
             var sm = new MqttClientSubscriptionsManager(new OptionsWrapper<MqttServerOptions>(new MqttServerOptions()));
 
             var sp = new MqttSubscribePacket();
-            sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce));
+            sp.TopicFilters.Add(new TopicFilter("A/B/C"));
 
             sm.Subscribe(sp, "");
 
@@ -25,7 +25,7 @@ namespace MQTTnet.Core.Tests
                 QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
             };
 
-            Assert.IsTrue(sm.IsSubscribed(pp));
+            Assert.IsTrue(sm.CheckSubscriptions(pp).IsSubscribed);
         }
 
         [TestMethod]
@@ -34,7 +34,7 @@ namespace MQTTnet.Core.Tests
             var sm = new MqttClientSubscriptionsManager(new OptionsWrapper<MqttServerOptions>(new MqttServerOptions()));
 
             var sp = new MqttSubscribePacket();
-            sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce));
+            sp.TopicFilters.Add(new TopicFilter("A/B/C"));
 
             sm.Subscribe(sp, "");
 
@@ -44,7 +44,7 @@ namespace MQTTnet.Core.Tests
                 QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
             };
 
-            Assert.IsFalse(sm.IsSubscribed(pp));
+            Assert.IsFalse(sm.CheckSubscriptions(pp).IsSubscribed);
         }
 
         [TestMethod]
@@ -53,7 +53,7 @@ namespace MQTTnet.Core.Tests
             var sm = new MqttClientSubscriptionsManager(new OptionsWrapper<MqttServerOptions>(new MqttServerOptions()));
 
             var sp = new MqttSubscribePacket();
-            sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce));
+            sp.TopicFilters.Add(new TopicFilter("A/B/C"));
 
             sm.Subscribe(sp, "");
 
@@ -63,13 +63,13 @@ namespace MQTTnet.Core.Tests
                 QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
             };
 
-            Assert.IsTrue(sm.IsSubscribed(pp));
+            Assert.IsTrue(sm.CheckSubscriptions(pp).IsSubscribed);
 
             var up = new MqttUnsubscribePacket();
             up.TopicFilters.Add("A/B/C");
             sm.Unsubscribe(up);
 
-            Assert.IsFalse(sm.IsSubscribed(pp));
+            Assert.IsFalse(sm.CheckSubscriptions(pp).IsSubscribed);
         }
     }
 }

Tests/MQTTnet.TestApp.NetCore/ClientTest.cs

@@ -21,10 +21,10 @@ namespace MQTTnet.TestApp.NetCore
                     {
                         Server = "localhost"
                     },
-                    ////ChannelOptions = new MqttClientWebSocketOptions
-                    ////{
-                    ////    Uri = "localhost"
-                    ////}
+                    //ChannelOptions = new MqttClientWebSocketOptions
+                    //{
+                    //    Uri = "ws://localhost:59690/mqtt"
+                    //}
                 };
 
                 var factory = new MqttFactory();

Tests/MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj

@@ -3,7 +3,7 @@
   <PropertyGroup>
     <OutputType>Exe</OutputType>
     <DebugType>Full</DebugType>
-    <TargetFrameworks>netcoreapp2.0;net452</TargetFrameworks>
+    <TargetFrameworks>netcoreapp2.0;net452;net461</TargetFrameworks>
   </PropertyGroup>
 
   <ItemGroup>

Tests/MQTTnet.TestApp.NetCore/Program.cs

@@ -15,7 +15,7 @@ namespace MQTTnet.TestApp.NetCore
     {
         public static void Main()
         {
-            Console.WriteLine("MQTTnet - TestApp.NetFramework");
+            Console.WriteLine($"MQTTnet - TestApp.{TargetFrameworkInfoProvider.TargetFramework}");
             Console.WriteLine("1 = Start client");
             Console.WriteLine("2 = Start server");
             Console.WriteLine("3 = Start performance test");

Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml

@@ -78,6 +78,7 @@
                     </StackPanel>
 
                     <TextBlock>Received messages:</TextBlock>
+                    <CheckBox x:Name="AddReceivedMessagesToList" IsChecked="True">Add received messages to list</CheckBox>
                     <ListBox MinHeight="50" MaxHeight="250" x:Name="ReceivedMessages" Margin="0,0,0,10">
                         <ListBox.ItemTemplate>
                             <DataTemplate>

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

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Text;
 using System.Threading.Tasks;
 using Windows.Security.Cryptography.Certificates;
@@ -18,6 +19,8 @@ namespace MQTTnet.TestApp.UniversalWindows
 {
     public sealed partial class MainPage
     {
+        private readonly ConcurrentQueue<MqttNetTraceMessage> _traceMessages = new ConcurrentQueue<MqttNetTraceMessage>();
+
         private IMqttClient _mqttClient;
         private IMqttServer _mqttServer;
 
@@ -30,15 +33,32 @@ namespace MQTTnet.TestApp.UniversalWindows
 
         private async void OnTraceMessagePublished(object sender, MqttNetTraceMessagePublishedEventArgs e)
         {
-            await Trace.Dispatcher.RunAsync(CoreDispatcherPriority.High, () =>
+            _traceMessages.Enqueue(e.TraceMessage);
+            while (_traceMessages.Count > 100)
             {
-                var text = $"[{e.TraceMessage.Timestamp:yyyy-MM-dd HH:mm:ss.fff}] [{e.TraceMessage.Level}] [{e.TraceMessage.Source}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Message}]{Environment.NewLine}";
-                if (e.TraceMessage.Exception != null)
-                {
-                    text += $"{e.TraceMessage.Exception}{Environment.NewLine}";
-                }
+                _traceMessages.TryDequeue(out _);
+            }
 
-                Trace.Text += text;
+            var logText = new StringBuilder();
+            foreach (var traceMessage in _traceMessages)
+            {
+                logText.AppendFormat(
+                    "[{0:yyyy-MM-dd HH:mm:ss.fff}] [{1}] [{2}] [{3}] [{4}]{5}", traceMessage.Timestamp,
+                    traceMessage.Level,
+                    traceMessage.Source,
+                    traceMessage.ThreadId,
+                    traceMessage.Message,
+                    Environment.NewLine);
+
+                if (traceMessage.Exception != null)
+                {
+                    logText.AppendLine(traceMessage.Exception.ToString());
+                }
+            }
+
+            await Trace.Dispatcher.RunAsync(CoreDispatcherPriority.Low, () =>
+            {
+                Trace.Text = logText.ToString();
             });
         }
 
@@ -109,11 +129,13 @@ namespace MQTTnet.TestApp.UniversalWindows
         {
             var item = $"Timestamp: {DateTime.Now:O} | Topic: {eventArgs.ApplicationMessage.Topic} | Payload: {Encoding.UTF8.GetString(eventArgs.ApplicationMessage.Payload)} | QoS: {eventArgs.ApplicationMessage.QualityOfServiceLevel}";
 
-            await Dispatcher.RunAsync(CoreDispatcherPriority.Normal, () =>
+            await Dispatcher.RunAsync(CoreDispatcherPriority.Low, () =>
             {
-                ReceivedMessages.Items.Add(item);
+                if (AddReceivedMessagesToList.IsChecked == true)
+                {
+                    ReceivedMessages.Items.Add(item);
+                }
             });
-
         }
 
         private async void Publish(object sender, RoutedEventArgs e)