Merge branch 'master' of https://github.com/chkr1011/MQTTnet

Reference to MQTT Tester - updated

Build/MQTTnet.AspNetCore.nuspec

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

Build/MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.5.0</version>
+    <version>2.5.1.1</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,32 +10,34 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
-    <releaseNotes>* [Core] Merged the platform specific projects (Thanks to @JanEggers)
-* [Core] Migrated the trace to a non-static approach (Breaking Change!)
-* [Core] Added a builder for application messages using a fluent API
-* [Core] Introduced CI and DI (Thanks to @JanEggers)
-* [Core] Added interfaces for publishing and receiving which applies to every client and server (Thanks to @ChristianRiedl)
-* [Core] Fixed an issue when reading from closed streams
-* [Client] Added a first version of a managed client which will manage the connection, subscription etc. automatically (Thanks to @JTrotta)
-* [Client] The session state response from the server is now returned in the _ConnectAsync_ method and also part of the _Connected_ event args
-* [Client] Added a _TopicFilterBuilder_ using a fluent API (Namespace Changes!)
-* [Client] Added several new options for the WebSocket channel (Thanks to @ChristianRiedl)
-* [Client] Refactored the options and added a builder using a fluent API (Breaking Change!)
-* [Client] Added more options for WebSocket connections like RequestHeaders, SubProtocol etc.
-* [Server] Added support for WebSockets via ASP.NET Core 2.0 (additional nuget) (Thanks to @ChristianRiedl, @JanEggers)
-* [Server] Added support for a custom application message interceptor
-* [Server] Fixed an issue with dropped connections on UWP (Thanks to @haeberle)
-* [Server] Added support for a custom subscription interceptor which allows denying of subscriptions and closing the connection
+    <releaseNotes>* [Core] Fixed library reference issues for .NET 4.6 and netstandard 2.0 (Thanks to @JanEggers).
+* [Core] Several COM exceptions are now wrapped properly resulting in less warnings in the trace.
+* [Core] Removed application message payload from trace to reduce trace size and increase performance.
+* [Client] Fixed WebSocket sub protocol negotiation for ASP.NET Core 2 servers (Thanks to @JanEggers).
+* [Client] Fixed broken connection after 30 seconds then using WebSocket protocol (Thanks to @ChristianRiedl).
+* [Server] Client connections are now closed when the server is stopped (Thanks to @zhudanfei).
+* [Server] Published messages from the server are now retained (if set) (Thanks to @ChristianRiedl). BREAKING CHANGE!
     </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" />
+        <dependency id="System.Threading.Thread" version="4.3.0" />
+      </group>
+
+      <group targetFramework="netstandard2.0">
+        <dependency id="NETStandard.Library" version="2.0.0" />
+        <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" />
         <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" />
@@ -49,10 +51,16 @@
         <dependency id="Microsoft.Extensions.Options" version="1.1.2" />
       </group>
         
-      <group targetFramework="net451">
-        <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" />
+      <group targetFramework="net452">
+        <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>
@@ -62,6 +70,10 @@
     <!-- .NET Standard 1.3 -->
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard1.3\MQTTnet.Core.*" target="lib\netstandard1.3\"/>
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>
+    
+    <!-- .NET Standard 2.0 -->
+    <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard2.0\MQTTnet.Core.*" target="lib\netstandard2.0\"/>
+    <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard2.0\MQTTnet.*" target="lib\netstandard2.0\"/>
 
     <!-- Universal Windows -->
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\uap10.0\MQTTnet.Core.*" target="lib\uap10.0\"/>
@@ -70,6 +82,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

@@ -9,7 +9,9 @@ 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="net461" /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.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$version /p:AssemblyVersion=$version /verbosity:m

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/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
+#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
+#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,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard1.3;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>
@@ -32,23 +32,46 @@
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'" />
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="1.1.1" />
-    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
   </ItemGroup>
 
   <ItemGroup>
     <ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj" />
   </ItemGroup>
 
+  <ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
+    <PackageReference Include="System.Net.Security" Version="4.3.2" />
+    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
+    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
+    <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
+    <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>
+
   <ItemGroup Condition="'$(TargetFramework)'=='netstandard1.3'">
     <PackageReference Include="System.Net.Security" Version="4.3.2" />
     <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
     <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
     <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="1.0.2" />
+    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.0.2" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
     <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="5.4.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="1.1.1" />
+    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)'=='net452'">
+    <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

@@ -16,20 +16,6 @@ namespace MQTTnet
     {
         private readonly IServiceProvider _serviceProvider;
 
-        private static IServiceProvider BuildServiceProvider()
-        {
-            var serviceProvider = new ServiceCollection()
-                .AddMqttClient()
-                .AddMqttServer()
-                .AddLogging()
-                .BuildServiceProvider();
-
-            serviceProvider.GetRequiredService<ILoggerFactory>()
-                .AddMqttTrace();
-
-            return serviceProvider;
-        }
-
         public MqttFactory()
             : this(BuildServiceProvider())
         {
@@ -37,7 +23,7 @@ namespace MQTTnet
 
         public MqttFactory(IServiceProvider serviceProvider)
         {
-            _serviceProvider = serviceProvider;
+            _serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
         }
 
         public ILoggerFactory GetLoggerFactory()
@@ -99,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()
@@ -128,5 +112,19 @@ namespace MQTTnet
 
             return _serviceProvider.GetRequiredService<IMqttServer>();
         }
+
+        private static IServiceProvider BuildServiceProvider()
+        {
+            var serviceProvider = new ServiceCollection()
+                .AddMqttClient()
+                .AddMqttServer()
+                .AddLogging()
+                .BuildServiceProvider();
+
+            serviceProvider.GetRequiredService<ILoggerFactory>()
+                .AddMqttTrace();
+
+            return serviceProvider;
+        }
     }
 }

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,87 @@ 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);
-            }
+            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);
-            }
+            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 +139,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/Client/MqttClientOptionsBuilder.cs

@@ -98,7 +98,7 @@ namespace MQTTnet.Core.Client
                 AllowUntrustedCertificates = allowUntrustedCertificates,
                 IgnoreCertificateChainErrors = ignoreCertificateChainErrors,
                 IgnoreCertificateRevocationErrors = ignoreCertificateRevocationErrors,
-                Certificates = certificates.ToList()
+                Certificates = certificates?.ToList()
             };
 
             return this;

MQTTnet.Core/MQTTnet.Core.csproj

@@ -23,8 +23,8 @@
   </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/MqttPublishPacket.cs

@@ -1,5 +1,4 @@
-using System;
-using MQTTnet.Core.Protocol;
+using MQTTnet.Core.Protocol;
 
 namespace MQTTnet.Core.Packets
 {
@@ -19,7 +18,7 @@ namespace MQTTnet.Core.Packets
         {
             return nameof(MqttPublishPacket) +
                 ": [Topic=" + Topic + "]" +
-                " [Payload=" + Convert.ToBase64String(Payload) + "]" +
+                " [PayloadLength=" + Payload?.Length + "]" +
                 " [QoSLevel=" + QualityOfServiceLevel + "]" +
                 " [Dup=" + Dup + "]" +
                 " [Retain=" + Retain + "]" +

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

@@ -8,9 +8,9 @@ namespace MQTTnet.Core.Server
     {
         event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
         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

@@ -28,6 +28,11 @@ namespace MQTTnet.Core.Server
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
 
+            if (cancellationToken.IsCancellationRequested)
+            {
+                return;
+            }
+
             Task.Run(async () => await SendPendingPublishPacketsAsync(adapter, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
@@ -91,7 +96,7 @@ namespace MQTTnet.Core.Server
                     _pendingPublishPackets.Add(packet, CancellationToken.None);
                 }
 
-                _session.Stop();
+                await _session.StopAsync();
             }
         }
     }

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));
@@ -60,12 +57,14 @@ namespace MQTTnet.Core.Server
 
             try
             {
+                var cancellationTokenSource = new CancellationTokenSource();
+
                 _willMessage = willMessage;
                 _adapter = adapter;
-                _cancellationTokenSource = new CancellationTokenSource();
+                _cancellationTokenSource = cancellationTokenSource;
 
-                _pendingMessagesQueue.Start(adapter, _cancellationTokenSource.Token);
-                await ReceivePacketsAsync(adapter, _cancellationTokenSource.Token).ConfigureAwait(false);
+                _pendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
+                await ReceivePacketsAsync(adapter, cancellationTokenSource.Token).ConfigureAwait(false);
             }
             catch (OperationCanceledException)
             {
@@ -80,7 +79,7 @@ namespace MQTTnet.Core.Server
             }
         }
 
-        public void Stop()
+        public async Task StopAsync()
         {
             try
             {
@@ -88,15 +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
             {
-                if (_willMessage != null)
+                var willMessage = _willMessage;
+                if (willMessage != null)
                 {
-                    _sessionsManager.DispatchApplicationMessage(this, _willMessage);
+                    _willMessage = null; //clear willmessage so it is send just once
+                    await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage);
                 }
             }
         }
@@ -105,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);
         }
 
@@ -129,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();
             }
         }
 
@@ -178,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)
@@ -198,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());
@@ -215,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 });
@@ -252,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,16 +43,17 @@ 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;
         public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
         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));
 
@@ -64,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.");
@@ -96,6 +83,8 @@ namespace MQTTnet.Core.Server
             }
 
             _logger.LogInformation("Started.");
+
+            Started?.Invoke(this, new MqttServerStartedEventArgs());
         }
 
         public async Task StopAsync()
@@ -110,7 +99,7 @@ namespace MQTTnet.Core.Server
                 await adapter.StopAsync();
             }
 
-            _clientSessionsManager.Clear();
+            await _clientSessionsManager.StopAsync();
 
             _logger.LogInformation("Stopped.");
         }

MQTTnet.Core/Server/MqttServerStartedEventArgs.cs

@@ -0,0 +1,8 @@
+using System;
+
+namespace MQTTnet.Core.Server
+{
+    public class MqttServerStartedEventArgs : EventArgs
+    {
+    }
+}

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/MQTTnet.Core.Tests.csproj

@@ -7,7 +7,6 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
     <PackageReference Include="MSTest.TestAdapter" Version="1.2.0" />
     <PackageReference Include="MSTest.TestFramework" Version="1.2.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.3.0" />
@@ -15,7 +14,6 @@
 
   <ItemGroup>
     <ProjectReference Include="..\..\Frameworks\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj" />
-    <ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj" />
   </ItemGroup>
 
 </Project>

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -9,6 +9,7 @@ using MQTTnet.Core.Client;
 using MQTTnet.Core.Protocol;
 using MQTTnet.Core.Server;
 using Microsoft.Extensions.DependencyInjection;
+using MQTTnet.Core.Internal;
 using MQTTnet.Core.Packets;
 
 namespace MQTTnet.Core.Tests
@@ -53,6 +54,7 @@ namespace MQTTnet.Core.Tests
         public async Task MqttServer_WillMessage()
         {
             var serverAdapter = new TestMqttServerAdapter();
+
             var services = new ServiceCollection()
                 .AddLogging()
                 .AddMqttServer()
@@ -70,16 +72,19 @@ namespace MQTTnet.Core.Tests
                 var c2 = await serverAdapter.ConnectTestClient(s, "c2", willMessage);
 
                 c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
-                await c1.SubscribeAsync(new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce));
+                await c1.SubscribeAsync(new TopicFilter("#"));
 
                 await c2.DisconnectAsync();
 
                 await Task.Delay(1000);
+
+                await c1.DisconnectAsync();
             }
             finally
             {
                 await s.StopAsync();
             }
+
             Assert.AreEqual(1, receivedMessagesCount);
         }
 
@@ -212,7 +217,6 @@ namespace MQTTnet.Core.Tests
                 .BuildServiceProvider();
 
             var s = new MqttFactory(services).CreateMqttServer();
-            var retainMessagemanager = services.GetRequiredService<IMqttClientRetainedMessageManager>();
 
             var receivedMessagesCount = 0;
             try
@@ -223,19 +227,7 @@ namespace MQTTnet.Core.Tests
                 await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
                 await c1.DisconnectAsync();
 
-                var subscribe = new MqttSubscribePacket()
-                {
-                    TopicFilters = new List<TopicFilter>()
-                    {
-                        new TopicFilter("retained", MqttQualityOfServiceLevel.AtMostOnce)
-                    }
-                };
-
-                //make shure the retainedMessageManagerreceived the package
-                while (!(await retainMessagemanager.GetSubscribedMessagesAsync(subscribe)).Any())
-                {
-                    await Task.Delay(TimeSpan.FromMilliseconds(10));
-                }
+                await services.WaitForRetainedMessage("retained").TimeoutAfter(TimeSpan.FromSeconds(5));
 
                 var c2 = await serverAdapter.ConnectTestClient(s, "c2");
                 c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
@@ -314,6 +306,8 @@ namespace MQTTnet.Core.Tests
                 await s.StopAsync();
             }
 
+            await services.WaitForRetainedMessage("retained").TimeoutAfter(TimeSpan.FromSeconds(5));
+
             s = new MqttFactory(services).CreateMqttServer(options => options.Storage = storage);
 
             var receivedMessagesCount = 0;
@@ -436,4 +430,25 @@ namespace MQTTnet.Core.Tests
             Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount);
         }
     }
+
+    public static class TestExtensions
+    {
+        public static async Task WaitForRetainedMessage(this IServiceProvider services, string topic)
+        {
+            var retainMessagemanager = services.GetRequiredService<IMqttClientRetainedMessageManager>();
+            
+            var subscribe = new MqttSubscribePacket()
+            {
+                TopicFilters = new List<TopicFilter>()
+                {
+                    new TopicFilter(topic, MqttQualityOfServiceLevel.AtMostOnce)
+                }
+            };
+
+            while (!(await retainMessagemanager.GetSubscribedMessagesAsync(subscribe)).Any())
+            {
+                await Task.Delay(TimeSpan.FromMilliseconds(10));
+            }
+        }
+    }
 }

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.Core.Tests/TestMqttCommunicationAdapter.cs

@@ -33,7 +33,7 @@ namespace MQTTnet.Core.Tests
 
             foreach (var packet in packets)
             {
-                Partner.SendPacketInternal(packet);
+                Partner.EnqueuePacketInternal(packet);
             }
 
             return Task.FromResult(0);
@@ -46,7 +46,7 @@ namespace MQTTnet.Core.Tests
             return Task.Run(() => _incomingPackets.Take(), cancellationToken);
         }
 
-        private void SendPacketInternal(MqttBasePacket packet)
+        private void EnqueuePacketInternal(MqttBasePacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 

Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapterFactory.cs

@@ -4,11 +4,11 @@ using MQTTnet.Core.Client;
 
 namespace MQTTnet.Core.Tests
 {
-    public class MqttCommunicationAdapterFactory : IMqttCommunicationAdapterFactory
+    public class TestMqttCommunicationAdapterFactory : IMqttCommunicationAdapterFactory
     {
         private readonly IMqttCommunicationAdapter _adapter;
 
-        public MqttCommunicationAdapterFactory(IMqttCommunicationAdapter adapter)
+        public TestMqttCommunicationAdapterFactory(IMqttCommunicationAdapter adapter)
         {
             _adapter = adapter;
         }

Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs

@@ -17,7 +17,11 @@ namespace MQTTnet.Core.Tests
             adapterA.Partner = adapterB;
             adapterB.Partner = adapterA;
 
-            var client = new MqttClient(new MqttCommunicationAdapterFactory(adapterA), new TestLogger<MqttClient>(), new MqttPacketDispatcher(new TestLogger<MqttPacketDispatcher>()));
+            var client = new MqttClient(
+                new TestMqttCommunicationAdapterFactory(adapterA),
+                new TestLogger<MqttClient>(), 
+                new MqttPacketDispatcher(new TestLogger<MqttPacketDispatcher>()));
+
             var connected = WaitForClientToConnect(server, clientId);
 
             FireClientAcceptedEvent(adapterB);
@@ -29,8 +33,6 @@ namespace MQTTnet.Core.Tests
                 ChannelOptions = new MqttClientTcpOptions()
             };
 
-            options.ChannelOptions = new MqttClientTcpOptions();
-
             await client.ConnectAsync(options);
             await connected;
 

Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs

@@ -5,7 +5,6 @@ using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.FileProviders;
-using Microsoft.Extensions.Logging;
 using MQTTnet.AspNetCore;
 using MQTTnet.Core;
 
@@ -21,21 +20,34 @@ namespace MQTTnet.TestApp.AspNetCore2
         }
 
         // In class _Startup_ of the ASP.NET Core 2.0 project.
-        public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory loggerFactory)
+        public void Configure(IApplicationBuilder app, IHostingEnvironment env)
         {
             app.UseMqttEndpoint();
-            app.UseMqttServer(async server =>
+            app.UseMqttServer(server =>
             {
-                var msg = new MqttApplicationMessageBuilder()
-                    .WithPayload("Mqtt is awesome")
-                    .WithTopic("message");
-
-                while (true)
+                server.Started += async (sender, args) =>
                 {
-                    server.PublishAsync(msg.Build()).Wait();
-                    await Task.Delay(TimeSpan.FromSeconds(2));
-                    msg.WithPayload("Mqtt is still awesome at " + DateTime.Now);
-                }
+                    var msg = new MqttApplicationMessageBuilder()
+                        .WithPayload("Mqtt is awesome")
+                        .WithTopic("message");
+
+                    while (true)
+                    {
+                        try
+                        {
+                            await server.PublishAsync(msg.Build());
+                            msg.WithPayload("Mqtt is still awesome at " + DateTime.Now);
+                        }
+                        catch (Exception e)
+                        {
+                            Console.WriteLine(e);
+                        }
+                        finally
+                        {
+                            await Task.Delay(TimeSpan.FromSeconds(2));
+                        }
+                    }
+                };
             });
 
             app.Use((context, next) =>

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,17 +3,15 @@
   <PropertyGroup>
     <OutputType>Exe</OutputType>
     <DebugType>Full</DebugType>
-    <TargetFrameworks>netcoreapp2.0;net452</TargetFrameworks>
+    <TargetFrameworks>netcoreapp2.0;net452;net461</TargetFrameworks>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
     <PackageReference Include="Newtonsoft.Json" Version="10.0.3" />
   </ItemGroup>
 
   <ItemGroup>
     <ProjectReference Include="..\..\Frameworks\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj" />
-    <ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj" />
   </ItemGroup>
 
   <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

@@ -9,6 +9,7 @@ using Microsoft.Extensions.Logging;
 using MQTTnet.Core;
 using MQTTnet.Core.Client;
 using MQTTnet.Core.Diagnostics;
+using MQTTnet.Core.ManagedClient;
 using MQTTnet.Core.Protocol;
 using MQTTnet.Core.Server;
 using MQTTnet.Implementations;
@@ -29,14 +30,14 @@ namespace MQTTnet.TestApp.UniversalWindows
 
         private async void OnTraceMessagePublished(object sender, MqttNetTraceMessagePublishedEventArgs e)
         {
-            await Trace.Dispatcher.RunAsync(CoreDispatcherPriority.High, () =>
+            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}";
+            }
+            
+            await Trace.Dispatcher.RunAsync(CoreDispatcherPriority.Low, () =>
             {
-                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}";
-                }
-
                 Trace.Text += text;
             });
         }
@@ -108,11 +109,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)
@@ -454,6 +457,21 @@ namespace MQTTnet.TestApp.UniversalWindows
                 var mqttServer = new MqttFactory().CreateMqttServer();
             }
 
+            {
+                // Setup and start a managed MQTT client.
+                var options = new ManagedMqttClientOptionsBuilder()
+                    .WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
+                    .WithClientOptions(new MqttClientOptionsBuilder()
+                        .WithClientId("Client1")
+                        .WithTcpServer("broker.hivemq.com")
+                        .WithTls().Build())
+                    .Build();
+
+                var mqttClient = new MqttFactory().CreateManagedMqttClient();
+                await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build());
+                await mqttClient.StartAsync(options);
+            }
+
         }
 
         private async void StartServer(object sender, RoutedEventArgs e)