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>
@@ -72,5 +77,8 @@
     <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()
         {
@@ -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,6 +12,8 @@ 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)
         {
@@ -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,6 +15,8 @@ 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;
@@ -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,9 +87,13 @@ 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
             {
@@ -100,7 +101,7 @@ namespace MQTTnet.Core.Server
                 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,
             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

@@ -24,8 +24,7 @@ namespace MQTTnet.Core.Server
             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));
@@ -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

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)