Update solution files.

.gitignore

@@ -288,8 +288,6 @@ __pycache__/
 *.odx.cs
 *.xsd.cs
 
-
-Build/nuget.exe
 *.js
 *.map
 

Build/MQTTnet.AspNetCore.nuspec

@@ -5,7 +5,7 @@
     <version>0.0.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
-    <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
+    <license type="file">LICENSE</license>
     <projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
@@ -23,6 +23,9 @@
   </metadata>
 
   <files>
+    <!-- License -->
+    <file src="..\LICENSE" />
+
     <!-- .NET Standard 2.0 -->
     <file src="..\Source\MQTTnet.AspNetCore\bin\Release\netstandard2.0\MQTTnet.AspNetCore.*" target="lib\netstandard2.0\"/>
   </files>

Build/MQTTnet.Extensions.ManagedClient.nuspec

@@ -5,7 +5,7 @@
     <version>0.0.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
-    <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
+    <license type="file">LICENSE</license>
     <projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
@@ -19,6 +19,9 @@
   </metadata>
 
   <files>
+    <!-- License -->
+    <file src="..\LICENSE" />
+
     <!-- .NET Standard 1.3 -->
     <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard1.3\MQTTnet.Extensions.ManagedClient.*" target="lib\netstandard1.3\"/>
 

Build/MQTTnet.Extensions.Rpc.nuspec

@@ -5,7 +5,7 @@
     <version>0.0.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
-    <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
+    <license type="file">LICENSE</license>
     <projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
@@ -19,6 +19,9 @@
   </metadata>
 
   <files>
+    <!-- License -->
+    <file src="..\LICENSE" />
+
     <!-- .NET Standard 1.3 -->
     <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard1.3\MQTTnet.Extensions.Rpc.*" target="lib\netstandard1.3\"/>
 

Build/MQTTnet.NETStandard.nuspec

@@ -5,7 +5,7 @@
     <version>0.0.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
-    <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
+    <license type="file">LICENSE</license>
     <projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
@@ -30,7 +30,13 @@
   </metadata>
 
   <files>
-    <file src="..\Source\MQTTnet\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>   
+    <!-- License -->
+    <file src="..\LICENSE" />
+
+    <!-- .NET Standard 1.3 -->
+    <file src="..\Source\MQTTnet\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>
+
+    <!-- .NET Standard 2.0 -->
     <file src="..\Source\MQTTnet\bin\Release\netstandard2.0\MQTTnet.*" target="lib\netstandard2.0\"/>
   </files>
 </package>

Build/MQTTnet.nuspec

@@ -5,23 +5,26 @@
     <version>0.0.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
-    <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
+    <license type="file">LICENSE</license>
     <projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
     <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] Added support for MQTTv5 packages.
-* [Core] Performance improvements (removed several exceptions).
+    <releaseNotes>
+* [Core] Added support for MQTTv5 packages.
+* [Core] Performance improvements.
 * [Core] Removed obsolete methods.
 * [Core] Fixed a memory leak when processing lots of messages (thanks to @tschanko)
 * [Core] Added more overloads for MQTT factory.
 * [Core] The client password is now hidden from the logs (replaced with **** if set).
+* [Core] Fixed a memory leak when using SSL connections (thanks to @biovoid). 
 * [Client] Added validation of topics before publishing.
 * [Client] Added new MQTTv5 features to options builder.
 * [Client] Added uniform API across all supported MQTT versions (BREAKING CHANGE!)
 * [Client] The client will now avoid sending an ACK if an exception has been thrown in message handler (thanks to @ramonsmits).
 * [Client] Fixed issues in QoS 2 handling which leads to message loss.
 * [Client] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
+* [Client] Added more configuration values to TCP endpoint options.
 * [ManagedClient] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
 * [ManagedClient] The log ID is now propagated to the internal client (thanks to @vbBerni).
 * [ManagedClient] Added validation of topics before publishing.
@@ -33,6 +36,7 @@
 * [Server] Added the flag "IsSecureConnection" which is set to true when the connection is encrypted.
 * [Server] Fixed wrong will message behavior when stopping server (thanks to @JohBa)
 * [Server] Added validation of topics before publishing.
+* [Server] Added more configuration values to TCP endpoint options.
 * [MQTTnet Server] Added as first Alpha version of standalone cross platform MQTT server.
 
 * [Note] Due to MQTTv5 a lot of new classes were introduced. This required adding new namespaces as well. Most classes are backward compatible but new namespaces must be added.
@@ -53,7 +57,7 @@
         <dependency id="System.Net.WebSockets.Client" version="4.3.2" />
       </group>
       <group targetFramework="uap10.0">
-        <dependency id="Microsoft.NETCore.UniversalWindowsPlatform" version="6.1.9" />
+        <dependency id="Microsoft.NETCore.UniversalWindowsPlatform" version="6.2.8" />
       </group>
       <group targetFramework="net452">
       </group>
@@ -63,6 +67,9 @@
   </metadata>
 
   <files>
+    <!-- License -->
+    <file src="..\LICENSE" />
+
     <!-- .NET Standard 1.3 -->
     <file src="..\Source\MQTTnet\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>
     

Build/build.ps1

@@ -3,57 +3,55 @@ param([string]$assemblyVersion, [string]$nugetVersion)
 if ([string]::IsNullOrEmpty($assemblyVersion)) {$assemblyVersion = "0.0.1"}
 if ([string]::IsNullOrEmpty($nugetVersion)) {$nugetVersion = "0.0.1"}
 
+$vswhere = ${Env:\ProgramFiles(x86)} + '\Microsoft Visual Studio\Installer\vswhere'
+$msbuild = &$vswhere -products * -requires Microsoft.Component.MSBuild -latest -find MSBuild\**\Bin\MSBuild.exe
+
+Write-Host
 Write-Host "Assembly version = $assemblyVersion"
 Write-Host "Nuget version    = $nugetVersion"
+Write-Host "MSBuild path     = $msbuild"
 Write-Host
 
-$vswhere = ${Env:\ProgramFiles(x86)} + '\Microsoft Visual Studio\Installer\vswhere'
+# Build the core library
+&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 
-$path = &$vswhere -latest -products * -requires Microsoft.Component.MSBuild -property installationPath
-if ($path) {
-    $msbuild = join-path $path 'MSBuild\15.0\Bin\MSBuild.exe'
+# Build the ASP.NET Core 2.0 extension
+&$msbuild ..\Source\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 
-	# Build the core library
-    &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    
-	# Build the ASP.NET Core 2.0 extension
-	&$msbuild ..\Source\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+# Build the RPC extension
+&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 
-	# Build the RPC extension
-	&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+# Build the Managed Client extension
+&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 
-	# Build the Managed Client extension
-	&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net461" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard1.3" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
-    &$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+Remove-Item .\NuGet -Force -Recurse -ErrorAction SilentlyContinue
 
-	Remove-Item .\NuGet -Force -Recurse -ErrorAction SilentlyContinue
+Copy-Item MQTTnet.AspNetCore.nuspec -Destination MQTTnet.AspNetCore.nuspec.old -Force
+(Get-Content MQTTnet.AspNetCore.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.AspNetCore.nuspec
+Copy-Item MQTTnet.Extensions.Rpc.nuspec -Destination MQTTnet.Extensions.Rpc.nuspec.old -Force
+(Get-Content MQTTnet.Extensions.Rpc.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.Rpc.nuspec
+Copy-Item MQTTnet.Extensions.ManagedClient.nuspec -Destination MQTTnet.Extensions.ManagedClient.nuspec.old -Force
+(Get-Content MQTTnet.Extensions.ManagedClient.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.ManagedClient.nuspec
 
-    Copy-Item MQTTnet.AspNetCore.nuspec -Destination MQTTnet.AspNetCore.nuspec.old -Force
-	(Get-Content MQTTnet.AspNetCore.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.AspNetCore.nuspec
-    Copy-Item MQTTnet.Extensions.Rpc.nuspec -Destination MQTTnet.Extensions.Rpc.nuspec.old -Force
-    (Get-Content MQTTnet.Extensions.Rpc.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.Rpc.nuspec
-    Copy-Item MQTTnet.Extensions.ManagedClient.nuspec -Destination MQTTnet.Extensions.ManagedClient.nuspec.old -Force
-    (Get-Content MQTTnet.Extensions.ManagedClient.nuspec) -replace '\$nugetVersion', $nugetVersion | Set-Content MQTTnet.Extensions.ManagedClient.nuspec
-	
-	New-Item -ItemType Directory -Force -Path .\NuGet
-    .\nuget.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-    .\nuget.exe pack MQTTnet.NETStandard.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-    .\nuget.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-	.\nuget.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-    .\nuget.exe pack MQTTnet.Extensions.ManagedClient.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
-    
-    Move-Item MQTTnet.AspNetCore.nuspec.old -Destination MQTTnet.AspNetCore.nuspec -Force
-    Move-Item MQTTnet.Extensions.Rpc.nuspec.old -Destination MQTTnet.Extensions.Rpc.nuspec -Force
-    Move-Item MQTTnet.Extensions.ManagedClient.nuspec.old -Destination MQTTnet.Extensions.ManagedClient.nuspec -Force
-}
+New-Item -ItemType Directory -Force -Path .\NuGet
+.\nuget.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.NETStandard.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.AspNetCore.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.Extensions.Rpc.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
+.\nuget.exe pack MQTTnet.Extensions.ManagedClient.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $nugetVersion
+
+Move-Item MQTTnet.AspNetCore.nuspec.old -Destination MQTTnet.AspNetCore.nuspec -Force
+Move-Item MQTTnet.Extensions.Rpc.nuspec.old -Destination MQTTnet.Extensions.Rpc.nuspec -Force
+Move-Item MQTTnet.Extensions.ManagedClient.nuspec.old -Destination MQTTnet.Extensions.ManagedClient.nuspec -Force

MQTTnet.sln

@@ -1,7 +1,7 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio 15
-VisualStudioVersion = 15.0.27004.2010
+# Visual Studio Version 16
+VisualStudioVersion = 16.0.28729.10
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Tests", "Tests\MQTTnet.Core.Tests\MQTTnet.Tests.csproj", "{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}"
 EndProject
@@ -19,6 +19,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{67C28AC1
 		Build\MQTTnet.AspNetCore.nuspec = Build\MQTTnet.AspNetCore.nuspec
 		Build\MQTTnet.Extensions.ManagedClient.nuspec = Build\MQTTnet.Extensions.ManagedClient.nuspec
 		Build\MQTTnet.Extensions.Rpc.nuspec = Build\MQTTnet.Extensions.Rpc.nuspec
+		Build\MQTTnet.NETStandard.nuspec = Build\MQTTnet.NETStandard.nuspec
 		Build\MQTTnet.nuspec = Build\MQTTnet.nuspec
 		Build\upload.ps1 = Build\upload.ps1
 	EndProjectSection

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -20,12 +20,17 @@ namespace MQTTnet.AspNetCore
         }
 
         public string Endpoint => Connection.ConnectionId;
-        public bool IsSecureConnection => false; // TODO: Fix detection.
+        public bool IsSecureConnection => false; // TODO: Fix detection (WS vs. WSS).
+
         public ConnectionContext Connection { get; }
         public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
-        public event EventHandler ReadingPacketStarted;
-        public event EventHandler ReadingPacketCompleted;
 
+        public long BytesSent { get; } // TODO: Fix calculation.
+        public long BytesReceived { get; } // TODO: Fix calculation.
+
+        public Action ReadingPacketStartedCallback { get; set; }
+        public Action ReadingPacketCompletedCallback { get; set; }
+        
         private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
 
         public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
@@ -34,6 +39,7 @@ namespace MQTTnet.AspNetCore
             {
                 return tcp.StartAsync();
             }
+
             return Task.CompletedTask;
         }
 
@@ -80,7 +86,7 @@ namespace MQTTnet.AspNetCore
                             else
                             {
                                 // we did receive something but the message is not yet complete
-                                ReadingPacketStarted?.Invoke(this, EventArgs.Empty);
+                                ReadingPacketStartedCallback?.Invoke();
                             }
                         }
                         else if (readResult.IsCompleted)
@@ -99,7 +105,7 @@ namespace MQTTnet.AspNetCore
             }
             finally
             {
-                ReadingPacketCompleted?.Invoke(this, EventArgs.Empty);
+                ReadingPacketCompletedCallback?.Invoke();
             }
 
             cancellationToken.ThrowIfCancellationRequested();

Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs

@@ -44,9 +44,9 @@ namespace MQTTnet.AspNetCore
             clientCertificate?.Dispose();
 
             var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection);
-            var clientAdapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(), _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter)));
+            var channelAdapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(), _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter)));
 
-            var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(clientAdapter);
+            var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(channelAdapter);
             ClientAcceptedHandler?.Invoke(eventArgs);
 
             if (eventArgs.SessionTask != null)

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientExtensions.cs

@@ -14,106 +14,128 @@ namespace MQTTnet.Extensions.ManagedClient
     {
         public static IManagedMqttClient UseConnectedHandler(this IManagedMqttClient client, Func<MqttClientConnectedEventArgs, Task> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.ConnectedHandler = null;
-                return client;
+                return client.UseConnectedHandler((IMqttClientConnectedHandler)null);
             }
 
-            client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(handler);
-            return client;
+            return client.UseConnectedHandler(new MqttClientConnectedHandlerDelegate(handler));
         }
 
         public static IManagedMqttClient UseConnectedHandler(this IManagedMqttClient client, Action<MqttClientConnectedEventArgs> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.ConnectedHandler = null;
-                return client;
+                return client.UseConnectedHandler((IMqttClientConnectedHandler)null);
             }
 
-            client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(handler);
+            return client.UseConnectedHandler(new MqttClientConnectedHandlerDelegate(handler));
+        }
+
+        public static IManagedMqttClient UseConnectedHandler(this IManagedMqttClient client, IMqttClientConnectedHandler handler)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
+            client.ConnectedHandler = handler;
             return client;
         }
 
         public static IManagedMqttClient UseDisconnectedHandler(this IManagedMqttClient client, Func<MqttClientDisconnectedEventArgs, Task> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.DisconnectedHandler = null;
-                return client;
+                return client.UseDisconnectedHandler((IMqttClientDisconnectedHandler)null);
             }
 
-            client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(handler);
-            return client;
+            return client.UseDisconnectedHandler(new MqttClientDisconnectedHandlerDelegate(handler));
         }
 
         public static IManagedMqttClient UseDisconnectedHandler(this IManagedMqttClient client, Action<MqttClientDisconnectedEventArgs> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.DisconnectedHandler = null;
-                return client;
+                return client.UseDisconnectedHandler((IMqttClientDisconnectedHandler)null);
             }
 
-            client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(handler);
+            return client.UseDisconnectedHandler(new MqttClientDisconnectedHandlerDelegate(handler));
+        }
+
+        public static IManagedMqttClient UseDisconnectedHandler(this IManagedMqttClient client, IMqttClientDisconnectedHandler handler)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
+            client.DisconnectedHandler = handler;
             return client;
         }
 
         public static IManagedMqttClient UseApplicationMessageReceivedHandler(this IManagedMqttClient client, Func<MqttApplicationMessageReceivedEventArgs, Task> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.ApplicationMessageReceivedHandler = null;
-                return client;
+                return client.UseApplicationMessageReceivedHandler((IMqttApplicationMessageReceivedHandler)null);
             }
 
-            client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler);
-
-            return client;
+            return client.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(handler));
         }
 
         public static IManagedMqttClient UseApplicationMessageReceivedHandler(this IManagedMqttClient client, Action<MqttApplicationMessageReceivedEventArgs> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.ApplicationMessageReceivedHandler = null;
-                return client;
+                return client.UseApplicationMessageReceivedHandler((IMqttApplicationMessageReceivedHandler)null);
             }
 
-            client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler);
+            return client.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(handler));
+        }
 
+        public static IManagedMqttClient UseApplicationMessageReceivedHandler(this IManagedMqttClient client, IMqttApplicationMessageReceivedHandler handler)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
+            client.ApplicationMessageReceivedHandler = handler;
             return client;
         }
 
-        public static Task SubscribeAsync(this IManagedMqttClient managedClient, params TopicFilter[] topicFilters)
+        public static Task SubscribeAsync(this IManagedMqttClient client, params TopicFilter[] topicFilters)
         {
-            if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
+            if (client == null) throw new ArgumentNullException(nameof(client));
 
-            return managedClient.SubscribeAsync(topicFilters);
+            return client.SubscribeAsync(topicFilters);
         }
 
-        public static Task SubscribeAsync(this IManagedMqttClient managedClient, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        public static Task SubscribeAsync(this IManagedMqttClient client, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
-            if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
+            if (client == null) throw new ArgumentNullException(nameof(client));
             if (topic == null) throw new ArgumentNullException(nameof(topic));
 
-            return managedClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
+            return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
         }
 
-        public static Task SubscribeAsync(this IManagedMqttClient managedClient, string topic)
+        public static Task SubscribeAsync(this IManagedMqttClient client, string topic)
         {
-            if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
+            if (client == null) throw new ArgumentNullException(nameof(client));
             if (topic == null) throw new ArgumentNullException(nameof(topic));
 
-            return managedClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());
+            return client.SubscribeAsync(new TopicFilterBuilder().WithTopic(topic).Build());
         }
 
-        public static Task UnsubscribeAsync(this IManagedMqttClient managedClient, params string[] topicFilters)
+        public static Task UnsubscribeAsync(this IManagedMqttClient client, params string[] topicFilters)
         {
-            if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
+            if (client == null) throw new ArgumentNullException(nameof(client));
 
-            return managedClient.UnsubscribeAsync(topicFilters);
+            return client.UnsubscribeAsync(topicFilters);
         }
 
         public static async Task PublishAsync(this IManagedMqttClient client, IEnumerable<MqttApplicationMessage> applicationMessages)
@@ -190,12 +212,16 @@ namespace MQTTnet.Extensions.ManagedClient
 
         public static Task<MqttClientPublishResult> PublishAsync(this IManagedMqttClient client, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder, CancellationToken cancellationToken)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             var message = builder(new MqttApplicationMessageBuilder()).Build();
             return client.PublishAsync(message, cancellationToken);
         }
 
         public static Task<MqttClientPublishResult> PublishAsync(this IManagedMqttClient client, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             var message = builder(new MqttApplicationMessageBuilder()).Build();
             return client.PublishAsync(message, CancellationToken.None);
         }

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -119,11 +119,6 @@ namespace MQTTnet.Extensions.Rpc
                 return Task.FromResult(0);
             }
 
-            if (tcs.Task.IsCompleted || tcs.Task.IsCanceled)
-            {
-                return Task.FromResult(0);
-            }
-
             tcs.TrySetResult(eventArgs.ApplicationMessage.Payload);
 
             return Task.FromResult(0);

Source/MQTTnet.Server/MQTTnet.Server.csproj

@@ -27,23 +27,10 @@
     <NoWarn>1701;1702</NoWarn>
   </PropertyGroup>
 
-  <ItemGroup>
-    <Content Remove="wwwroot\app\app.ts" />
-  </ItemGroup>
-
   <ItemGroup>
     <Content Update="appsettings.json">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </Content>
-    <Content Update="wwwroot\app\app.js">
-      <CopyToOutputDirectory>Never</CopyToOutputDirectory>
-    </Content>
-    <Content Update="wwwroot\app\app.js.map">
-      <CopyToOutputDirectory>Never</CopyToOutputDirectory>
-    </Content>
-    <Content Update="wwwroot\Index.html">
-      <CopyToOutputDirectory>Never</CopyToOutputDirectory>
-    </Content>
   </ItemGroup>
 
   <ItemGroup>
@@ -81,9 +68,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <TypeScriptCompile Include="wwwroot\app\app.ts">
-      <CopyToOutputDirectory>Never</CopyToOutputDirectory>
-    </TypeScriptCompile>
+    <Folder Include="wwwroot\" />
   </ItemGroup>
 
 </Project>

Source/MQTTnet.Server/Program.cs

@@ -50,7 +50,7 @@ namespace MQTTnet.Server
 
             Console.ForegroundColor = ConsoleColor.White;
             Console.WriteLine(@"
- Version:    1.0.0-alpha1
+ Version:    1.0.0-alpha2
  License:    MIT (read LICENSE file)
  Sponsoring: https://opencollective.com/mqttnet
  Support:    https://github.com/chkr1011/MQTTnet/issues

Source/MQTTnet.Server/tsconfig.json

@@ -1,12 +0,0 @@
-{
-  "compilerOptions": {
-    "noImplicitAny": false,
-    "noEmitOnError": true,
-    "removeComments": false,
-    "sourceMap": true,
-    "target": "es5"
-  },
-  "exclude": [
-    "node_modules"
-  ]
-}

Source/MQTTnet.Server/wwwroot/Index.html

@@ -1,35 +0,0 @@
-<!DOCTYPE html>
-<html>
-<head>
-    <meta charset="utf-8" />
-    <title>MQTTnet Server test</title>
-</head>
-<body>
-<script src="node_modules/systemjs/dist/system.js"></script>
-<script>
-    System.config({
-        packages: {
-            '': {
-                defaultExtension: 'js'
-            }
-        },
-        map: {
-            mqtt: 'node_modules/mqtt/dist/mqtt.min'
-        }
-    });
-    System.import('app/app');
-</script>
-    
-<div>
-    <p id="state"></p>
-    <input id="topic" placeholder="topic" />
-    <input id="msg" placeholder="msg" />
-    <button id="publish">send</button>
-</div>
-
-<div>
-    <ul id="msgs" ></ul>
-</div>
-
-</body>
-</html>

Source/MQTTnet.Server/wwwroot/app/app.ts

@@ -1,58 +0,0 @@
-import { connect } from "mqtt";
-
-var client = connect('ws://' + location.host + '/mqtt',
-    {
-        clientId: "client" + Math.floor(Math.random() * 6) + 1
-    });
-
-window.onbeforeunload = () => {
-    client.end();
-};
-
-var publishButton = document.getElementById("publish");
-var topicInput = <HTMLInputElement>document.getElementById("topic");
-var msgInput = <HTMLInputElement>document.getElementById("msg");
-var stateParagraph = document.getElementById("state");
-var msgsList = <HTMLUListElement>document.getElementById("msgs");
-
-publishButton.onclick = click => {
-    var topic = topicInput.value;
-    var msg = msgInput.value;
-    client.publish(topic, msg);
-};
-
-client.on('connect', () => {
-    client.subscribe('#', { qos: 0 }, (err, granted) => {
-        console.log(err);
-    });
-    client.publish('presence', 'Hello mqtt');
-
-    stateParagraph.innerText = "connected";
-    showMsg("[connect]");
-});
-
-client.on("error", e => {
-    showMsg("error: " + e.message);
-});
-
-client.on("reconnect", () => {
-    stateParagraph.innerText = "reconnecting";
-    showMsg("[reconnect]");
-});
-
-client.on('message', (topic, message) => {
-    showMsg(topic + ": " + message.toString());
-});
-
-function showMsg(msg: string) {
-    //console.log(msg);
-
-    var node = document.createElement("LI");
-    node.appendChild(document.createTextNode(msg));
-
-    msgsList.appendChild(node);
-
-    if (msgsList.childElementCount > 50) {
-        msgsList.removeChild(msgsList.childNodes[0]);
-    }
-}

Source/MQTTnet/Adapter/IMqttChannelAdapter.cs

@@ -13,10 +13,14 @@ namespace MQTTnet.Adapter
         bool IsSecureConnection { get; }
 
         MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
-        
-        event EventHandler ReadingPacketStarted;
 
-        event EventHandler ReadingPacketCompleted;
+        long BytesSent { get; }
+
+        long BytesReceived { get; }
+
+        Action ReadingPacketStartedCallback { get; set; }
+
+        Action ReadingPacketCompletedCallback { get; set; }
 
         Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken);
 

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -25,16 +25,19 @@ namespace MQTTnet.Adapter
         private readonly MqttPacketReader _packetReader;
 
         private readonly byte[] _fixedHeaderBuffer = new byte[2];
-       
+
         private bool _isDisposed;
 
+        private long _bytesReceived;
+        private long _bytesSent;
+
         public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetChildLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
             _channel = channel ?? throw new ArgumentNullException(nameof(channel));
             PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
-            
+
             _packetReader = new MqttPacketReader(_channel);
 
             _logger = logger.CreateChildLogger(nameof(MqttChannelAdapter));
@@ -46,9 +49,12 @@ namespace MQTTnet.Adapter
 
         public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
 
-        public event EventHandler ReadingPacketStarted;
-        public event EventHandler ReadingPacketCompleted;
+        public long BytesSent => Interlocked.Read(ref _bytesSent);
+        public long BytesReceived => Interlocked.Read(ref _bytesReceived);
 
+        public Action ReadingPacketStartedCallback { get; set; }
+        public Action ReadingPacketCompletedCallback { get; set; }
+            
         public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
@@ -62,7 +68,7 @@ namespace MQTTnet.Adapter
                 else
                 {
                     await MqttTaskTimeout.WaitAsync(t => _channel.ConnectAsync(t), timeout, cancellationToken).ConfigureAwait(false);
-                }               
+                }
             }
             catch (Exception exception)
             {
@@ -101,7 +107,7 @@ namespace MQTTnet.Adapter
                 WrapException(exception);
             }
         }
-        
+
         public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken)
         {
             await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
@@ -118,7 +124,9 @@ namespace MQTTnet.Adapter
                     await MqttTaskTimeout.WaitAsync(
                         t => _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, t), timeout, cancellationToken).ConfigureAwait(false);
                 }
-                
+
+                Interlocked.Add(ref _bytesReceived, packetData.Count);
+
                 PacketFormatterAdapter.FreeBuffer();
 
                 _logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet);
@@ -159,7 +167,9 @@ namespace MQTTnet.Adapter
                     return null;
                 }
 
-                if (!PacketFormatterAdapter.ProtocolVersion.HasValue)
+                Interlocked.Add(ref _bytesSent, receivedMqttPacket.TotalLength);
+
+                if (PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.Unknown)
                 {
                     PacketFormatterAdapter.DetectProtocolVersion(receivedMqttPacket);
                 }
@@ -190,10 +200,21 @@ namespace MQTTnet.Adapter
             return null;
         }
 
+        public void ResetStatistics()
+        {
+            Interlocked.Exchange(ref _bytesReceived, 0L);
+            Interlocked.Exchange(ref _bytesSent, 0L);
+        }
+
         private async Task<ReceivedMqttPacket> ReceiveAsync(CancellationToken cancellationToken)
         {
             var readFixedHeaderResult = await _packetReader.ReadFixedHeaderAsync(_fixedHeaderBuffer, cancellationToken).ConfigureAwait(false);
 
+            if (cancellationToken.IsCancellationRequested)
+            {
+                return null;
+            }
+
             try
             {
                 if (readFixedHeaderResult.ConnectionClosed)
@@ -201,7 +222,7 @@ namespace MQTTnet.Adapter
                     return null;
                 }
 
-                ReadingPacketStarted?.Invoke(this, EventArgs.Empty);
+                ReadingPacketStartedCallback?.Invoke();
 
                 var fixedHeader = readFixedHeaderResult.FixedHeader;
                 if (fixedHeader.RemainingLength == 0)
@@ -221,13 +242,7 @@ namespace MQTTnet.Adapter
                         chunkSize = bytesLeft;
                     }
 
-#if WINDOWS_UWP
-                    var readBytes = await _channel.ReadAsync(body, bodyOffset, (int)chunkSize, cancellationToken).ConfigureAwait(false);
-#else
-                    // async/await is not used to avoid the overhead of context switches. We assume that the remaining data
-                    // has been sent from the sender directly after the initial bytes.
-                    var readBytes = _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
-#endif
+                    var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
 
                     if (cancellationToken.IsCancellationRequested)
                     {
@@ -238,7 +253,7 @@ namespace MQTTnet.Adapter
                     {
                         return null;
                     }
-                    
+
                     bodyOffset += readBytes;
                 } while (bodyOffset < body.Length);
 
@@ -247,7 +262,7 @@ namespace MQTTnet.Adapter
             }
             finally
             {
-                ReadingPacketCompleted?.Invoke(this, EventArgs.Empty);
+                ReadingPacketCompletedCallback?.Invoke();
             }
         }
 
@@ -268,15 +283,19 @@ namespace MQTTnet.Adapter
 
         private static bool IsWrappedException(Exception exception)
         {
-            return exception is TaskCanceledException ||
-                   exception is OperationCanceledException ||
+            return exception is OperationCanceledException ||
                    exception is MqttCommunicationTimedOutException ||
                    exception is MqttCommunicationException;
         }
 
         private static void WrapException(Exception exception)
         {
-            if (exception is IOException && exception.InnerException is SocketException socketException)
+            if (exception is IOException && exception.InnerException is SocketException innerException)
+            {
+                exception = innerException;
+            }
+
+            if (exception is SocketException socketException)
             {
                 if (socketException.SocketErrorCode == SocketError.ConnectionAborted ||
                     socketException.SocketErrorCode == SocketError.OperationAborted)

Source/MQTTnet/Adapter/MqttServerAdapterClientAcceptedEventArgs.cs

@@ -5,12 +5,12 @@ namespace MQTTnet.Adapter
 {
     public class MqttServerAdapterClientAcceptedEventArgs : EventArgs
     {
-        public MqttServerAdapterClientAcceptedEventArgs(IMqttChannelAdapter client)
+        public MqttServerAdapterClientAcceptedEventArgs(IMqttChannelAdapter channelAdapter)
         {
-            Client = client ?? throw new ArgumentNullException(nameof(client));
+            ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
         }
 
-        public IMqttChannelAdapter Client { get; }
+        public IMqttChannelAdapter ChannelAdapter { get; }
 
         public Task SessionTask { get; set; }
     }

Source/MQTTnet/Client/MqttClientExtensions.cs

@@ -17,82 +17,97 @@ namespace MQTTnet.Client
     {
         public static IMqttClient UseConnectedHandler(this IMqttClient client, Func<MqttClientConnectedEventArgs, Task> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.ConnectedHandler = null;
-                return client;
+                return client.UseConnectedHandler((IMqttClientConnectedHandler)null);
             }
 
-            client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(handler);
-            return client;
+            return client.UseConnectedHandler(new MqttClientConnectedHandlerDelegate(handler));
         }
 
         public static IMqttClient UseConnectedHandler(this IMqttClient client, Action<MqttClientConnectedEventArgs> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.ConnectedHandler = null;
-                return client;
+                return client.UseConnectedHandler((IMqttClientConnectedHandler)null);
             }
 
-            client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(handler);
+            return client.UseConnectedHandler(new MqttClientConnectedHandlerDelegate(handler));
+        }
+
+        public static IMqttClient UseConnectedHandler(this IMqttClient client, IMqttClientConnectedHandler handler)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
+            client.ConnectedHandler = handler;
             return client;
         }
 
         public static IMqttClient UseDisconnectedHandler(this IMqttClient client, Func<MqttClientDisconnectedEventArgs, Task> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.DisconnectedHandler = null;
-                return client;
+                return client.UseDisconnectedHandler((IMqttClientDisconnectedHandler)null);
             }
 
-            client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(handler);
-            return client;
+            return client.UseDisconnectedHandler(new MqttClientDisconnectedHandlerDelegate(handler));
         }
 
         public static IMqttClient UseDisconnectedHandler(this IMqttClient client, Action<MqttClientDisconnectedEventArgs> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.DisconnectedHandler = null;
-                return client;
+                return client.UseDisconnectedHandler((IMqttClientDisconnectedHandler)null);
             }
 
-            client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(handler);
+            return client.UseDisconnectedHandler(new MqttClientDisconnectedHandlerDelegate(handler));
+        }
+
+        public static IMqttClient UseDisconnectedHandler(this IMqttClient client, IMqttClientDisconnectedHandler handler)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
+            client.DisconnectedHandler = handler;
             return client;
         }
 
         public static IMqttClient UseApplicationMessageReceivedHandler(this IMqttClient client, Func<MqttApplicationMessageReceivedEventArgs, Task> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.ApplicationMessageReceivedHandler = null;
-                return client;
+                return client.UseApplicationMessageReceivedHandler((IMqttApplicationMessageReceivedHandler)null);
             }
 
-            client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler);
-
-            return client;
+            return client.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(handler));
         }
 
         public static IMqttClient UseApplicationMessageReceivedHandler(this IMqttClient client, Action<MqttApplicationMessageReceivedEventArgs> handler)
         {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
             if (handler == null)
             {
-                client.ApplicationMessageReceivedHandler = null;
-                return client;
+                return client.UseApplicationMessageReceivedHandler((IMqttApplicationMessageReceivedHandler)null);
             }
 
-            client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler);
-
-            return client;
+            return client.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(handler));
         }
 
         public static IMqttClient UseApplicationMessageReceivedHandler(this IMqttClient client, IMqttApplicationMessageReceivedHandler handler)
         {
-            client.ApplicationMessageReceivedHandler = handler;
+            if (client == null) throw new ArgumentNullException(nameof(client));
 
+            client.ApplicationMessageReceivedHandler = handler;
             return client;
         }
 

Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs

@@ -1,4 +1,6 @@
-namespace MQTTnet.Client.Options
+using System.Net.Sockets;
+
+namespace MQTTnet.Client.Options
 {
     public class MqttClientTcpOptions : IMqttClientChannelOptions
     {
@@ -6,7 +8,13 @@
 
         public int? Port { get; set; }
 
-        public int BufferSize { get; set; } = 4096;
+        public int BufferSize { get; set; } = 1024;
+
+        public bool DualMode { get; set; } = true;
+
+        public bool NoDelay { get; set; } = true;
+
+        public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
 
         public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
 

Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs

@@ -20,19 +20,22 @@ namespace MQTTnet.Formatter
             UseProtocolVersion(protocolVersion);
         }
 
-        public MqttProtocolVersion? ProtocolVersion { get; private set; }
+        public MqttProtocolVersion ProtocolVersion { get; private set; } = MqttProtocolVersion.Unknown;
 
         public IMqttDataConverter DataConverter
         {
             get
             {
                 ThrowIfFormatterNotSet();
+
                 return _formatter.DataConverter;
             }
         }
 
         public ArraySegment<byte> Encode(MqttBasePacket packet)
         {
+            if (packet == null) throw new ArgumentNullException(nameof(packet));
+
             ThrowIfFormatterNotSet();
 
             return _formatter.Encode(packet);
@@ -40,6 +43,8 @@ namespace MQTTnet.Formatter
 
         public MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket)
         {
+            if (receivedMqttPacket == null) throw new ArgumentNullException(nameof(receivedMqttPacket));
+
             ThrowIfFormatterNotSet();
 
             return _formatter.Decode(receivedMqttPacket);
@@ -64,6 +69,11 @@ namespace MQTTnet.Formatter
 
         private void UseProtocolVersion(MqttProtocolVersion protocolVersion)
         {
+            if (protocolVersion == MqttProtocolVersion.Unknown)
+            {
+                throw new InvalidOperationException("MQTT protocol version is invalid.");
+            }
+
             ProtocolVersion = protocolVersion;
 
             switch (protocolVersion)

Source/MQTTnet/Formatter/MqttPacketReader.cs

@@ -3,7 +3,6 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Channel;
 using MQTTnet.Exceptions;
-using MQTTnet.Internal;
 
 namespace MQTTnet.Formatter
 {
@@ -31,8 +30,11 @@ namespace MQTTnet.Formatter
             {
                 var bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
 
-                cancellationToken.ThrowIfCancellationRequested();
-
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return null;
+                }
+                
                 if (bytesRead == 0)
                 {
                     return new ReadFixedHeaderResult
@@ -100,13 +102,19 @@ namespace MQTTnet.Formatter
                     return null;
                 }
 
-                var buffer = ReadByte(cancellationToken);
-                if (!buffer.HasValue)
+                var readCount = _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).GetAwaiter().GetResult();
+
+                if (cancellationToken.IsCancellationRequested)
                 {
                     return null;
                 }
 
-                encodedByte = buffer.Value;
+                if (readCount == 0)
+                {
+                    return null;
+                }
+
+                encodedByte = _singleByteBuffer[0];
 
                 value += (encodedByte & 127) * multiplier;
                 multiplier *= 128;
@@ -114,26 +122,8 @@ namespace MQTTnet.Formatter
 
             return value;
         }
-
-        private byte? ReadByte(CancellationToken cancellationToken)
-        {
-            var readCount = _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
-
-            if (cancellationToken.IsCancellationRequested)
-            {
-                return null;
-            }
-
-            if (readCount == 0)
-            {
-                return null;
-            }
-
-            return _singleByteBuffer[0];
-        }
-
 #else
-        
+
         private async Task<int?> ReadBodyLengthAsync(byte initialEncodedByte, CancellationToken cancellationToken)
         {
             var offset = 0;
@@ -154,13 +144,19 @@ namespace MQTTnet.Formatter
                     return null;
                 }
 
-                var buffer = await ReadByteAsync(cancellationToken).ConfigureAwait(false);
-                if (!buffer.HasValue)
+                var readCount = await _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false);
+
+                if (cancellationToken.IsCancellationRequested)
                 {
                     return null;
                 }
 
-                encodedByte = buffer.Value;
+                if (readCount == 0)
+                {
+                    return null;
+                }
+
+                encodedByte = _singleByteBuffer[0];
 
                 value += (encodedByte & 127) * multiplier;
                 multiplier *= 128;
@@ -168,24 +164,6 @@ namespace MQTTnet.Formatter
 
             return value;
         }
-
-        private async Task<byte?> ReadByteAsync(CancellationToken cancellationToken)
-        {
-            var readCount = await _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false);
-
-            if (cancellationToken.IsCancellationRequested)
-            {
-                return null;
-            }
-
-            if (readCount == 0)
-            {
-                return null;
-            }
-
-            return _singleByteBuffer[0];
-        }
-
 #endif
     }
 }

Source/MQTTnet/Formatter/MqttPacketWriter.cs

@@ -13,10 +13,14 @@ namespace MQTTnet.Formatter
     /// </summary>
     public class MqttPacketWriter
     {
+        private static readonly ArraySegment<byte> ZeroVariableLengthIntegerArray = new ArraySegment<byte>(new byte[1], 0, 1);
+        private static readonly ArraySegment<byte> ZeroTwoByteIntegerArray = new ArraySegment<byte>(new byte[2], 0, 2);
+
+        public static int InitialBufferSize = 128;
         public static int MaxBufferSize = 4096;
 
         // TODO: Consider using the ArrayPool here together with FreeBuffer.
-        private byte[] _buffer = new byte[128];
+        private byte[] _buffer = new byte[InitialBufferSize];
 
         private int _offset;
 
@@ -33,7 +37,7 @@ namespace MQTTnet.Formatter
         {
             if (value == 0)
             {
-                return new ArraySegment<byte>(new byte[1], 0, 1);
+                return ZeroVariableLengthIntegerArray;
             }
 
             if (value <= 127)
@@ -68,7 +72,14 @@ namespace MQTTnet.Formatter
 
         public void WriteWithLengthPrefix(string value)
         {
-            WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty));
+            if (string.IsNullOrEmpty(value))
+            {
+                Write(ZeroTwoByteIntegerArray);
+            }
+            else
+            {
+                WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value));
+            }         
         }
 
         public void WriteWithLengthPrefix(byte[] value)
@@ -77,8 +88,15 @@ namespace MQTTnet.Formatter
 
             EnsureAdditionalCapacity(value.Length + 2);
 
-            Write((ushort)value.Length);
-            Write(value, 0, value.Length);
+            if (value.Length == 0)
+            {
+                Write(ZeroTwoByteIntegerArray);   
+            }
+            else
+            {
+                Write((ushort)value.Length);
+                Write(value, 0, value.Length);
+            }           
         }
 
         public void Write(byte @byte)
@@ -103,19 +121,17 @@ namespace MQTTnet.Formatter
         {
             if (buffer == null) throw new ArgumentNullException(nameof(buffer));
 
+            if (count == 0)
+            {
+                return;
+            }
+
             EnsureAdditionalCapacity(count);
 
             Array.Copy(buffer, offset, _buffer, _offset, count);
             IncreasePosition(count);
         }
-
-        public void Write(ArraySegment<byte> buffer)
-        {
-            if (buffer == null) throw new ArgumentNullException(nameof(buffer));
-
-            Write(buffer.Array, buffer.Offset, buffer.Count);
-        }
-
+        
         public void Write(MqttPacketWriter propertyWriter)
         {
             if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter));
@@ -160,6 +176,11 @@ namespace MQTTnet.Formatter
             Array.Resize(ref _buffer, MaxBufferSize);
         }
 
+        private void Write(ArraySegment<byte> buffer)
+        {
+            Write(buffer.Array, buffer.Offset, buffer.Count);
+        }
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void EnsureAdditionalCapacity(int additionalCapacity)
         {

Source/MQTTnet/Formatter/MqttProtocolVersion.cs

@@ -2,6 +2,8 @@
 {
     public enum MqttProtocolVersion
     {
+        Unknown = 0,
+
         V310 = 3,
         V311 = 4,
         V500 = 5

Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Generic;
 using MQTTnet.Adapter;
 using MQTTnet.Packets;
 

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -62,7 +62,7 @@ namespace MQTTnet.Implementations
             if (_socket == null)
             {
                 _socket = new StreamSocket();
-                _socket.Control.NoDelay = true;
+                _socket.Control.NoDelay = _options.NoDelay;
                 _socket.Control.KeepAlive = true;
             }
 

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -6,6 +6,7 @@ using System.Security.Cryptography.X509Certificates;
 using System.Threading.Tasks;
 using System.IO;
 using System.Linq;
+using System.Runtime.ExceptionServices;
 using System.Threading;
 using MQTTnet.Channel;
 using MQTTnet.Client.Options;
@@ -17,12 +18,8 @@ namespace MQTTnet.Implementations
         private readonly IMqttClientOptions _clientOptions;
         private readonly MqttClientTcpOptions _options;
 
-        private Socket _socket;
         private Stream _stream;
 
-        /// <summary>
-        /// called on client sockets are created in connect
-        /// </summary>
         public MqttTcpChannel(IMqttClientOptions clientOptions)
         {
             _clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
@@ -31,48 +28,60 @@ namespace MQTTnet.Implementations
             IsSecureConnection = clientOptions.ChannelOptions?.TlsOptions?.UseTls == true;
         }
 
-        /// <summary>
-        /// called on server, sockets are passed in
-        /// connect will not be called
-        /// </summary>
-        public MqttTcpChannel(Socket socket, SslStream sslStream)
+        public MqttTcpChannel(Stream stream)
         {
-            _socket = socket ?? throw new ArgumentNullException(nameof(socket));
+            _stream = stream ?? throw new ArgumentNullException(nameof(stream));
 
-            IsSecureConnection = sslStream != null;
-
-            CreateStream(sslStream);
+            IsSecureConnection = stream is SslStream;
         }
 
-        public string Endpoint => _socket?.RemoteEndPoint?.ToString();
+        public string Endpoint { get; private set; }
 
         public bool IsSecureConnection { get; }
 
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
-            if (_socket == null)
+            Socket socket;
+
+            if (_options.AddressFamily == AddressFamily.Unspecified)
             {
-                _socket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true };
+                socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+            }
+            else
+            {
+                socket = new Socket(_options.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
             }
 
+            socket.ReceiveBufferSize = _options.BufferSize;
+            socket.SendBufferSize = _options.BufferSize;
+            socket.DualMode = _options.DualMode;
+            socket.NoDelay = _options.NoDelay;
+
             // Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430
-            using (cancellationToken.Register(() => _socket.Dispose()))
+            using (cancellationToken.Register(() => socket.Dispose()))
             {
 #if NET452 || NET461
-                await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
+                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);
+                await socket.ConnectAsync(_options.Server, _options.GetPort()).ConfigureAwait(false);
 #endif
             }
 
-            SslStream sslStream = null;
+            var networkStream = new NetworkStream(socket, true);
+
             if (_options.TlsOptions.UseTls)
             {
-                sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
+                var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                 await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+
+                _stream = sslStream;
+            }
+            else
+            {
+                _stream = networkStream;
             }
 
-            CreateStream(sslStream);
+            Endpoint = socket.RemoteEndPoint?.ToString();
         }
 
         public Task DisconnectAsync(CancellationToken cancellationToken)
@@ -83,37 +92,58 @@ namespace MQTTnet.Implementations
 
         public async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-            // Workaround for: https://github.com/dotnet/corefx/issues/24430
-            using (cancellationToken.Register(Dispose))
+            try
             {
-                if (cancellationToken.IsCancellationRequested)
+                // Workaround for: https://github.com/dotnet/corefx/issues/24430
+                using (cancellationToken.Register(Dispose))
                 {
-                    return 0;
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return 0;
+                    }
+
+                    return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            catch (IOException exception)
+            {
+                if (exception.InnerException is SocketException socketException)
+                {
+                    ExceptionDispatchInfo.Capture(socketException).Throw();
                 }
 
-                return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                throw;
             }
         }
 
         public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-            // Workaround for: https://github.com/dotnet/corefx/issues/24430
-            using (cancellationToken.Register(Dispose))
+            try
             {
-                if (cancellationToken.IsCancellationRequested)
+                // Workaround for: https://github.com/dotnet/corefx/issues/24430
+                using (cancellationToken.Register(Dispose))
                 {
-                    return;
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return;
+                    }
+
+                    await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            catch (IOException exception)
+            {
+                if (exception.InnerException is SocketException socketException)
+                {
+                    ExceptionDispatchInfo.Capture(socketException).Throw();
                 }
 
-                await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
-                //await _stream.FlushAsync(cancellationToken);
+                throw;
             }
         }
 
         public void Dispose()
         {
-            _socket = null;
-
             // When the stream is disposed it will also close the socket and this will also dispose it.
             // So there is no need to dispose the socket again.
             // https://stackoverflow.com/questions/3601521/should-i-manually-dispose-the-socket-after-closing-it
@@ -127,6 +157,8 @@ namespace MQTTnet.Implementations
             catch (NullReferenceException)
             {
             }
+
+            _stream = null;
         }
 
         private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
@@ -175,18 +207,6 @@ namespace MQTTnet.Implementations
 
             return certificates;
         }
-
-        private void CreateStream(Stream stream)
-        {
-            if (stream != null)
-            {
-                _stream = stream;
-            }
-            else
-            {
-                _stream = new NetworkStream(_socket, true);
-            }
-        }
     }
 }
 #endif

Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs

@@ -36,7 +36,7 @@ namespace MQTTnet.Implementations
                 _listener = new StreamSocketListener();
 
                 // This also affects the client sockets.
-                _listener.Control.NoDelay = true;
+                _listener.Control.NoDelay = options.DefaultEndpointOptions.NoDelay;
                 _listener.Control.KeepAlive = true;
                 _listener.Control.QualityOfService = SocketQualityOfService.LowLatency;
                 _listener.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync;

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -1,5 +1,6 @@
 #if !WINDOWS_UWP
 using System;
+using System.IO;
 using System.Net;
 using System.Net.Security;
 using System.Net.Sockets;
@@ -35,14 +36,14 @@ namespace MQTTnet.Implementations
             _tlsCertificate = tlsCertificate;
             _cancellationToken = cancellationToken;
             _logger = logger.CreateChildLogger(nameof(MqttTcpServerListener));
-            
+
             if (_options is MqttServerTlsTcpEndpointOptions tlsOptions)
             {
                 _tlsOptions = tlsOptions;
             }
         }
 
-        public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler;
+        public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
 
         public void Start()
         {
@@ -58,53 +59,7 @@ namespace MQTTnet.Implementations
             _logger.Info($"Starting TCP listener for {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
 
             _socket.Listen(_options.ConnectionBacklog);
-            Task.Run(AcceptClientConnectionsAsync, _cancellationToken);
-        }
-
-        private async Task AcceptClientConnectionsAsync()
-        {
-            while (!_cancellationToken.IsCancellationRequested)
-            {
-                try
-                {
-#if NET452 || NET461
-                    var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false);
-#else
-                    var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
-#endif
-                    clientSocket.NoDelay = true;
-
-                    SslStream sslStream = null;
-
-                    if (_tlsCertificate != null)
-                    {
-                        sslStream = new SslStream(new NetworkStream(clientSocket), false);
-                        await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, _tlsOptions.SslProtocol, false).ConfigureAwait(false);
-                    }
-
-                    _logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
-                        clientSocket.RemoteEndPoint,
-                        _socket.LocalEndPoint,
-                        _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
-
-                    var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketFormatterAdapter(), _logger);
-                    ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
-                }
-                catch (ObjectDisposedException)
-                {
-                    // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
-                }
-                catch (Exception exception)
-                {
-                    if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
-                    {
-                        return;
-                    }
-
-                    _logger.Error(exception, $"Error while accepting connection at TCP listener {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
-                    await Task.Delay(TimeSpan.FromSeconds(1), _cancellationToken).ConfigureAwait(false);
-                }
-            }
+            Task.Run(() => AcceptClientConnectionsAsync(_cancellationToken), _cancellationToken);
         }
 
         public void Dispose()
@@ -115,6 +70,78 @@ namespace MQTTnet.Implementations
             _tlsCertificate?.Dispose();
 #endif
         }
+
+        private async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken)
+        {
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                try
+                {
+#if NET452 || NET461
+                    var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false);
+#else
+                    var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
+#endif
+                    await HandleClientConnection(clientSocket).ConfigureAwait(false);
+                }
+                catch (Exception exception)
+                {
+                    _logger.Error(exception, $"Error while accepting connection at TCP listener {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
+                    await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
+                }
+            }
+        }
+
+        private async Task HandleClientConnection(Socket clientSocket)
+        {
+            Stream stream = null;
+
+            try
+            {
+                clientSocket.NoDelay = _options.NoDelay;
+
+                stream = new NetworkStream(clientSocket, true);
+
+                if (_tlsCertificate != null)
+                {
+                    var sslStream = new SslStream(stream, false);
+                    await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, _tlsOptions.SslProtocol, false).ConfigureAwait(false);
+                    stream = sslStream;
+                }
+
+                _logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
+                    clientSocket.RemoteEndPoint,
+                    _socket.LocalEndPoint,
+                    _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
+
+                var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(stream), new MqttPacketFormatterAdapter(), _logger);
+                ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
+            }
+            catch (Exception exception)
+            {
+                if (exception is ObjectDisposedException)
+                {
+                    // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
+                    return;
+                }
+
+                if (exception is SocketException socketException && socketException.SocketErrorCode == SocketError.OperationAborted)
+                {
+                    return;
+                }
+
+                try
+                {
+                    // Dispose already allocated resources.
+                    stream?.Dispose();
+                    clientSocket?.Dispose();
+                }
+                catch (Exception disposeException)
+                {
+                    throw new AggregateException(exception, disposeException);
+                }
+            }
+        }
     }
 }
 #endif

Source/MQTTnet/Internal/AsyncBlockingQueue.cs

@@ -8,7 +8,10 @@ namespace MQTTnet.Internal
     public sealed class AsyncQueue<TItem> : IDisposable
     {
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
-        private readonly ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
+
+        private ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
+
+        public int Count => _queue.Count;
 
         public void Enqueue(TItem item)
         {
@@ -16,9 +19,9 @@ namespace MQTTnet.Internal
             _semaphore.Release();
         }
 
-        public async Task<TItem> DequeueAsync(CancellationToken cancellationToken)
+        public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
         {
-            while (true)
+            while (!cancellationToken.IsCancellationRequested)
             {
                 await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
 
@@ -26,9 +29,26 @@ namespace MQTTnet.Internal
 
                 if (_queue.TryDequeue(out var item))
                 {
-                    return item;
+                    return new AsyncQueueDequeueResult<TItem>(true, item);
                 }
             }
+
+            return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+        }
+
+        public AsyncQueueDequeueResult<TItem> TryDequeue()
+        {
+            if (_queue.TryDequeue(out var item))
+            {
+                return new AsyncQueueDequeueResult<TItem>(true, item);
+            }
+
+            return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+        }
+
+        public void Clear()
+        {
+            Interlocked.Exchange(ref _queue, new ConcurrentQueue<TItem>());
         }
 
         public void Dispose()

Source/MQTTnet/Internal/AsyncQueueDequeueResult.cs

@@ -0,0 +1,15 @@
+namespace MQTTnet.Internal
+{
+    public class AsyncQueueDequeueResult<TItem>
+    {
+        public AsyncQueueDequeueResult(bool isSuccess, TItem item)
+        {
+            IsSuccess = isSuccess;
+            Item = item;
+        }
+
+        public bool IsSuccess { get; }
+
+        public TItem Item { get; }
+    }
+}

Source/MQTTnet/MQTTnet.csproj

@@ -54,7 +54,7 @@
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
-    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.1.9" />
+    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.8" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)'=='net452'">
@@ -62,9 +62,5 @@
 
   <ItemGroup Condition="'$(TargetFramework)'=='net461'">
   </ItemGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.2" />
-  </ItemGroup>
-
+  
 </Project>

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -34,8 +34,10 @@ namespace MQTTnet.Server
         private readonly MqttConnectPacket _connectPacket;
 
         private DateTime _lastPacketReceivedTimestamp;
+        private DateTime _lastNonKeepAlivePacketReceivedTimestamp;
+
         private long _receivedPacketsCount;
-        private long _sentPacketsCount;
+        private long _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere.
         private long _receivedApplicationMessagesCount;
         private long _sentApplicationMessagesCount;
 
@@ -64,6 +66,7 @@ namespace MQTTnet.Server
             _keepAliveMonitor = new MqttClientKeepAliveMonitor(this, _logger);
 
             _lastPacketReceivedTimestamp = DateTime.UtcNow;
+            _lastNonKeepAlivePacketReceivedTimestamp = _lastPacketReceivedTimestamp;
         }
 
         public string ClientId => _connectPacket.ClientId;
@@ -85,7 +88,7 @@ namespace MQTTnet.Server
         {
             status.ClientId = ClientId;
             status.Endpoint = _endpoint;
-            status.ProtocolVersion = _channelAdapter.PacketFormatterAdapter.ProtocolVersion.Value;
+            status.ProtocolVersion = _channelAdapter.PacketFormatterAdapter.ProtocolVersion;
 
             status.ReceivedApplicationMessagesCount = Interlocked.Read(ref _receivedApplicationMessagesCount);
             status.SentApplicationMessagesCount = Interlocked.Read(ref _sentApplicationMessagesCount);
@@ -94,8 +97,10 @@ namespace MQTTnet.Server
             status.SentPacketsCount = Interlocked.Read(ref _sentPacketsCount);
 
             status.LastPacketReceivedTimestamp = _lastPacketReceivedTimestamp;
+            status.LastNonKeepAlivePacketReceivedTimestamp = _lastNonKeepAlivePacketReceivedTimestamp;
 
-            //status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived;
+            status.BytesSent = _channelAdapter.BytesSent;
+            status.BytesReceived = _channelAdapter.BytesReceived;
         }
         
         //public void ClearPendingApplicationMessages()
@@ -116,25 +121,15 @@ namespace MQTTnet.Server
             return _packageReceiverTask;
         }
 
-        public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage)
-        {
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
-
-            Session.EnqueueApplicationMessage(applicationMessage, senderClientId, isRetainedApplicationMessage);
-
-            _logger.Verbose("Enqueued application message (ClientId: {0}).", ClientId);
-        }
-
         private async Task<MqttClientDisconnectType> RunInternalAsync()
         {
             var disconnectType = MqttClientDisconnectType.NotClean;
             try
             {
                 _logger.Info("Client '{0}': Session started.", ClientId);
-                //_eventDispatcher.OnClientConnected(ClientId);
-
-                _channelAdapter.ReadingPacketStarted += OnAdapterReadingPacketStarted;
-                _channelAdapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted;
+                
+                _channelAdapter.ReadingPacketStartedCallback = OnAdapterReadingPacketStarted;
+                _channelAdapter.ReadingPacketCompletedCallback = OnAdapterReadingPacketCompleted;
 
                 Session.WillMessage = _connectPacket.WillMessage;
 
@@ -166,7 +161,12 @@ namespace MQTTnet.Server
 
                     Interlocked.Increment(ref _sentPacketsCount);
                     _lastPacketReceivedTimestamp = DateTime.UtcNow;
-                    
+
+                    if (!(packet is MqttPingReqPacket || packet is MqttPingRespPacket))
+                    {
+                        _lastNonKeepAlivePacketReceivedTimestamp = _lastPacketReceivedTimestamp;
+                    }
+
                     _keepAliveMonitor.PacketReceived();
 
                     if (packet is MqttPublishPacket publishPacket)
@@ -243,12 +243,11 @@ namespace MQTTnet.Server
 
                 _packetDispatcher.Reset();
 
-                _channelAdapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
-                _channelAdapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
+                _channelAdapter.ReadingPacketStartedCallback = null;
+                _channelAdapter.ReadingPacketCompletedCallback = null;
 
                 _logger.Info("Client '{0}': Session stopped.", ClientId);
-                //_eventDispatcher.OnClientDisconnected(ClientId);
-
+                
                 _packageReceiverTask = null;
             }
 
@@ -265,7 +264,7 @@ namespace MQTTnet.Server
             var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
             foreach (var applicationMessage in retainedMessages)
             {
-                EnqueueApplicationMessage(applicationMessage, ClientId, true);
+                Session.EnqueueApplicationMessage(applicationMessage, ClientId, true);
             }
         }
 
@@ -354,29 +353,11 @@ namespace MQTTnet.Server
             };
 
             return SendAsync(pubRecPacket);
-
-            ////Task.Run(async () =>
-            ////{
-            ////    using (var pubRelPacketAwaiter = _packetDispatcher.AddPacketAwaiter<MqttPubRelPacket>(publishPacket.PacketIdentifier))
-            ////    {
-            ////        await pubRelPacketAwaiter.WaitOneAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
-
-            ////        var pubCompPacket = new MqttPubCompPacket
-            ////        {
-            ////            PacketIdentifier = publishPacket.PacketIdentifier,
-            ////            ReasonCode = MqttPubCompReasonCode.Success
-            ////        };
-
-            ////        await SendAsync(pubCompPacket).ConfigureAwait(false);
-            ////    }
-            ////});
-
-            //await Task.FromResult(0);
         }
 
         private async Task SendPendingPacketsAsync(CancellationToken cancellationToken)
         {
-            MqttPendingApplicationMessage queuedApplicationMessage = null;
+            MqttQueuedApplicationMessage queuedApplicationMessage = null;
             MqttPublishPacket publishPacket = null;
 
             try
@@ -501,12 +482,12 @@ namespace MQTTnet.Server
             }
         }
 
-        private void OnAdapterReadingPacketCompleted(object sender, EventArgs e)
+        private void OnAdapterReadingPacketCompleted()
         {
             _keepAliveMonitor?.Resume();
         }
 
-        private void OnAdapterReadingPacketStarted(object sender, EventArgs e)
+        private void OnAdapterReadingPacketStarted()
         {
             _keepAliveMonitor?.Pause();
         }

Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs

@@ -3,7 +3,6 @@ using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
-using MQTTnet.Packets;
 
 namespace MQTTnet.Server
 {

Source/MQTTnet/Server/MqttClientSession.cs

@@ -1,20 +1,26 @@
 using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
+using MQTTnet.Diagnostics;
 using MQTTnet.Server.Status;
 
 namespace MQTTnet.Server
 {
     public class MqttClientSession
     {
+        private readonly IMqttNetChildLogger _logger;
+
         private readonly DateTime _createdTimestamp = DateTime.UtcNow;
 
-        public MqttClientSession(string clientId, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions)
+        public MqttClientSession(string clientId, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetChildLogger logger)
         {
             ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
 
             SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, eventDispatcher, serverOptions);
             ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);
+
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            _logger = logger.CreateChildLogger(nameof(MqttClientSession));
         }
 
         public string ClientId { get; }
@@ -35,12 +41,15 @@ namespace MQTTnet.Server
                 return;
             }
 
+            _logger.Verbose("Queued application message with topic '{0}' (ClientId: {1}).", applicationMessage.Topic, ClientId);
+
             ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage);
         }
 
         public async Task SubscribeAsync(ICollection<TopicFilter> topicFilters, MqttRetainedMessagesManager retainedMessagesManager)
         {
             await SubscriptionsManager.SubscribeAsync(topicFilters).ConfigureAwait(false);
+
             var matchingRetainedMessages = await retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
             foreach (var matchingRetainedMessage in matchingRetainedMessages)
             {

Source/MQTTnet/Server/MqttClientSessionApplicationMessagesQueue.cs

@@ -1,54 +1,29 @@
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
 using System;
-using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 
 namespace MQTTnet.Server
 {
-    public class MqttPendingApplicationMessage
-    {
-        public MqttApplicationMessage ApplicationMessage { get; set; }
-
-        public string SenderClientId { get; set; }
-
-        public bool IsRetainedMessage { get; set; }
-
-        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
-
-        public bool IsDuplicate { get; set; }
-    }
-
     public class MqttClientSessionApplicationMessagesQueue : IDisposable
     {
-        private readonly Queue<MqttPendingApplicationMessage> _messageQueue = new Queue<MqttPendingApplicationMessage>();
-        private readonly AsyncAutoResetEvent _messageQueueLock = new AsyncAutoResetEvent();
-
+        private readonly AsyncQueue<MqttQueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttQueuedApplicationMessage>();
+        
         private readonly IMqttServerOptions _options;
 
         public MqttClientSessionApplicationMessagesQueue(IMqttServerOptions options)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
-
         }
 
-        public int Count
-        {
-            get
-            {
-                lock (_messageQueue)
-                {
-                    return _messageQueue.Count;
-                }
-            }
-        }
+        public int Count => _messageQueue.Count;
 
         public void Enqueue(MqttApplicationMessage applicationMessage, string senderClientId, MqttQualityOfServiceLevel qualityOfServiceLevel, bool isRetainedMessage)
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            Enqueue(new MqttPendingApplicationMessage
+            Enqueue(new MqttQueuedApplicationMessage
             {
                 ApplicationMessage = applicationMessage,
                 SenderClientId = senderClientId,
@@ -59,39 +34,23 @@ namespace MQTTnet.Server
 
         public void Clear()
         {
-            lock (_messageQueue)
-            {
-                _messageQueue.Clear();
-            }
+            _messageQueue.Clear();
         }
 
-        public void Dispose()
+        public async Task<MqttQueuedApplicationMessage> TakeAsync(CancellationToken cancellationToken)
         {
-        }
-
-        public async Task<MqttPendingApplicationMessage> TakeAsync(CancellationToken cancellationToken)
-        {
-            // TODO: Create a blocking queue from this.
-
-            while (!cancellationToken.IsCancellationRequested)
+            var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
+            if (!dequeueResult.IsSuccess)
             {
-                lock (_messageQueue)
-                {
-                    if (_messageQueue.Count > 0)
-                    {
-                        return _messageQueue.Dequeue();
-                    }
-                }
-
-                await _messageQueueLock.WaitOneAsync(cancellationToken).ConfigureAwait(false);
+                return null;
             }
 
-            return null;
+            return dequeueResult.Item;
         }
 
-        public void Enqueue(MqttPendingApplicationMessage enqueuedApplicationMessage)
+        public void Enqueue(MqttQueuedApplicationMessage queuedApplicationMessage)
         {
-            if (enqueuedApplicationMessage == null) throw new ArgumentNullException(nameof(enqueuedApplicationMessage));
+            if (queuedApplicationMessage == null) throw new ArgumentNullException(nameof(queuedApplicationMessage));
 
             lock (_messageQueue)
             {
@@ -104,14 +63,17 @@ namespace MQTTnet.Server
 
                     if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
                     {
-                        _messageQueue.Dequeue();
+                        _messageQueue.TryDequeue();
                     }
                 }
 
-                _messageQueue.Enqueue(enqueuedApplicationMessage);
+                _messageQueue.Enqueue(queuedApplicationMessage);
             }
+        }
 
-            _messageQueueLock.Set();
+        public void Dispose()
+        {
+            _messageQueue.Dispose();
         }
     }
 }

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -68,7 +68,7 @@ namespace MQTTnet.Server
 
             foreach (var connection in _connections.Values)
             {
-                var clientStatus = new MqttClientStatus(connection, this);
+                var clientStatus = new MqttClientStatus(connection);
                 connection.FillStatus(clientStatus);
                 
                 var sessionStatus = new MqttSessionStatus(connection.Session, this);
@@ -175,7 +175,8 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                var queuedApplicationMessage = await _messageQueue.DequeueAsync(cancellationToken).ConfigureAwait(false);
+                var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
+                var queuedApplicationMessage = dequeueResult.Item;
 
                 var sender = queuedApplicationMessage.Sender;
                 var applicationMessage = queuedApplicationMessage.ApplicationMessage;
@@ -331,8 +332,6 @@ namespace MQTTnet.Server
                 {
                     if (connectPacket.CleanSession)
                     {
-                        // TODO: Check if required.
-                        //session.Dispose();
                         session = null;
                         
                         _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
@@ -345,7 +344,7 @@ namespace MQTTnet.Server
 
                 if (session == null)
                 {
-                    session = new MqttClientSession(connectPacket.ClientId, _eventDispatcher, _options);
+                    session = new MqttClientSession(connectPacket.ClientId, _eventDispatcher, _options, _logger);
                     _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
                 }
 

Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs

@@ -0,0 +1,17 @@
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server
+{
+    public class MqttQueuedApplicationMessage
+    {
+        public MqttApplicationMessage ApplicationMessage { get; set; }
+
+        public string SenderClientId { get; set; }
+
+        public bool IsRetainedMessage { get; set; }
+
+        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
+
+        public bool IsDuplicate { get; set; }
+    }
+}

Source/MQTTnet/Server/MqttServer.cs

@@ -186,7 +186,7 @@ namespace MQTTnet.Server
 
         private void OnClientAccepted(MqttServerAdapterClientAcceptedEventArgs eventArgs)
         {
-            eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.Client);
+            eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.ChannelAdapter);
         }
     }
 }

Source/MQTTnet/Server/MqttServerExtensions.cs

@@ -2,14 +2,46 @@
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
-using MQTTnet.Client;
 using MQTTnet.Client.Publishing;
+using MQTTnet.Client.Receiving;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
     public static class MqttServerExtensions
     {
+        public static IMqttServer UseApplicationMessageReceivedHandler(this IMqttServer server, Func<MqttApplicationMessageReceivedEventArgs, Task> handler)
+        {
+            if (server == null) throw new ArgumentNullException(nameof(server));
+
+            if (handler == null)
+            {
+                return server.UseApplicationMessageReceivedHandler((IMqttApplicationMessageReceivedHandler)null);
+            }
+
+            return server.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(handler));
+        }
+
+        public static IMqttServer UseApplicationMessageReceivedHandler(this IMqttServer server, Action<MqttApplicationMessageReceivedEventArgs> handler)
+        {
+            if (server == null) throw new ArgumentNullException(nameof(server));
+
+            if (handler == null)
+            {
+                return server.UseApplicationMessageReceivedHandler((IMqttApplicationMessageReceivedHandler)null);
+            }
+
+            return server.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(handler));
+        }
+
+        public static IMqttServer UseApplicationMessageReceivedHandler(this IMqttServer server, IMqttApplicationMessageReceivedHandler handler)
+        {
+            if (server == null) throw new ArgumentNullException(nameof(server));
+
+            server.ApplicationMessageReceivedHandler = handler;
+            return server;
+        }
+
         public static Task SubscribeAsync(this IMqttServer server, string clientId, params TopicFilter[] topicFilters)
         {
             if (server == null) throw new ArgumentNullException(nameof(server));
@@ -120,12 +152,16 @@ namespace MQTTnet.Server
 
         public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder, CancellationToken cancellationToken)
         {
+            if (server == null) throw new ArgumentNullException(nameof(server));
+
             var message = builder(new MqttApplicationMessageBuilder()).Build();
             return server.PublishAsync(message, cancellationToken);
         }
 
         public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder)
         {
+            if (server == null) throw new ArgumentNullException(nameof(server));
+
             var message = builder(new MqttApplicationMessageBuilder()).Build();
             return server.PublishAsync(message, CancellationToken.None);
         }

Source/MQTTnet/Server/MqttServerTcpEndpointBaseOptions.cs

@@ -10,6 +10,8 @@ namespace MQTTnet.Server
 
         public int ConnectionBacklog { get; set; } = 10;
 
+        public bool NoDelay { get; set; } = true;
+
 #if WINDOWS_UWP
         public int BufferSize { get; set; } = 4096;
 #endif

Source/MQTTnet/Server/Status/IMqttClientStatus.cs

@@ -26,6 +26,10 @@ namespace MQTTnet.Server.Status
 
         IMqttSessionStatus Session { get; }
 
+        long BytesSent { get; }
+
+        long BytesReceived { get; }
+
         Task DisconnectAsync();
     }
 }

Source/MQTTnet/Server/Status/MqttClientStatus.cs

@@ -6,13 +6,11 @@ namespace MQTTnet.Server.Status
 {
     public class MqttClientStatus : IMqttClientStatus
     {
-        private readonly MqttClientSessionsManager _sessionsManager;
         private readonly MqttClientConnection _connection;
 
-        public MqttClientStatus(MqttClientConnection connection, MqttClientSessionsManager sessionsManager)
+        public MqttClientStatus(MqttClientConnection connection)
         {
             _connection = connection;
-            _sessionsManager = sessionsManager;
         }
 
         public string ClientId { get; set; }
@@ -35,6 +33,10 @@ namespace MQTTnet.Server.Status
 
         public IMqttSessionStatus Session { get; set; }
 
+        public long BytesSent { get; set; }
+
+        public long BytesReceived { get; set; }
+
         public Task DisconnectAsync()
         {
             return _connection.StopAsync();

Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj

@@ -7,7 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.9.0" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.0.1" />
     <PackageReference Include="MSTest.TestAdapter" Version="1.4.0" />
     <PackageReference Include="MSTest.TestFramework" Version="1.4.0" />
   </ItemGroup>

Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj

@@ -9,7 +9,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="BenchmarkDotNet" Version="0.11.3" />
+    <PackageReference Include="BenchmarkDotNet" Version="0.11.5" />
     <PackageReference Include="System.IO.Pipelines" Version="4.5.2" />
     <PackageReference Include="Microsoft.AspNetCore" Version="2.1.6" />
   </ItemGroup>

Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs

@@ -21,7 +21,7 @@ namespace MQTTnet.Benchmarks
         {
             var factory = new MqttFactory();
             var tcpServer = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger());
-            tcpServer.ClientAcceptedHandler += args => _serverChannel = (IMqttChannel)args.Client.GetType().GetField("_channel", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance).GetValue(args.Client);
+            tcpServer.ClientAcceptedHandler += args => _serverChannel = (IMqttChannel)args.ChannelAdapter.GetType().GetField("_channel", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance).GetValue(args.ChannelAdapter);
 
             _mqttServer = factory.CreateMqttServer(new[] { tcpServer }, new MqttNetLogger());
 

Tests/MQTTnet.Core.Tests/AsyncAutoResentEvent_Tests.cs

@@ -8,11 +8,11 @@ namespace MQTTnet.Tests
 {
     [TestClass]
     // Inspired from the vs-threading tests (https://github.com/Microsoft/vs-threading/blob/master/src/Microsoft.VisualStudio.Threading.Tests/AsyncAutoResetEventTests.cs)
-    public class AsyncAutoResetEventTests
+    public class AsyncAutoResetEvent_Tests
     {
         private readonly AsyncAutoResetEvent _aare;
 
-        public AsyncAutoResetEventTests()
+        public AsyncAutoResetEvent_Tests()
         {
             _aare = new AsyncAutoResetEvent();
         }

Tests/MQTTnet.Core.Tests/AsyncLock_Tests.cs

@@ -6,7 +6,7 @@ using MQTTnet.Internal;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class AsyncLockTests
+    public class AsyncLock_Tests
     {
         [TestMethod]
         public void AsyncLock()

Tests/MQTTnet.Core.Tests/AsyncQueue_Tests.cs

@@ -16,11 +16,27 @@ namespace MQTTnet.Tests
             queue.Enqueue("2");
             queue.Enqueue("3");
 
-            Assert.AreEqual("1", await queue.DequeueAsync(CancellationToken.None));
-            Assert.AreEqual("2", await queue.DequeueAsync(CancellationToken.None));
-            Assert.AreEqual("3", await queue.DequeueAsync(CancellationToken.None));
+            Assert.AreEqual("1", (await queue.TryDequeueAsync(CancellationToken.None)).Item);
+            Assert.AreEqual("2", (await queue.TryDequeueAsync(CancellationToken.None)).Item);
+            Assert.AreEqual("3", (await queue.TryDequeueAsync(CancellationToken.None)).Item);
         }
 
+        [TestMethod]
+        public void Count()
+        {
+            var queue = new AsyncQueue<string>();
+
+            queue.Enqueue("1");
+            Assert.AreEqual(1, queue.Count);
+
+            queue.Enqueue("2");
+            Assert.AreEqual(2, queue.Count);
+
+            queue.Enqueue("3");
+            Assert.AreEqual(3, queue.Count);
+        }
+
+
         [TestMethod]
         public async Task Preserve_ProcessAsync()
         {
@@ -31,7 +47,7 @@ namespace MQTTnet.Tests
             {
                 while (sum < 6)
                 {
-                    sum += await queue.DequeueAsync(CancellationToken.None);
+                    sum += (await queue.TryDequeueAsync(CancellationToken.None)).Item;
                 }
             });
             
@@ -47,5 +63,35 @@ namespace MQTTnet.Tests
             Assert.AreEqual(6, sum);
             Assert.AreEqual(TaskStatus.RanToCompletion, worker.Status);
         }
+
+        [TestMethod]
+        public void Dequeue_Sync()
+        {
+            var queue = new AsyncQueue<string>();
+            queue.Enqueue("1");
+            queue.Enqueue("2");
+            queue.Enqueue("3");
+
+            Assert.AreEqual("1", queue.TryDequeue().Item);
+            Assert.AreEqual("2", queue.TryDequeue().Item);
+            Assert.AreEqual("3", queue.TryDequeue().Item);
+        }
+
+        [TestMethod]
+        public void Clear()
+        {
+            var queue = new AsyncQueue<string>();
+            queue.Enqueue("1");
+            queue.Enqueue("2");
+            queue.Enqueue("3");
+
+            queue.Clear();
+            Assert.AreEqual(0, queue.Count);
+
+            queue.Enqueue("4");
+
+            Assert.AreEqual(1, queue.Count);
+            Assert.AreEqual("4", queue.TryDequeue().Item);
+        }
     }
 }

Tests/MQTTnet.Core.Tests/BlockingQueue_Tests.cs

@@ -6,7 +6,7 @@ using MQTTnet.Internal;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class BlockingQueueTests
+    public class BlockingQueue_Tests
     {
         [TestMethod]
         public void Preserve_Order()

Tests/MQTTnet.Core.Tests/Extension_Tests.cs

@@ -9,26 +9,31 @@ using MQTTnet.Internal;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class ExtensionTests
+    public class Extension_Tests
     {
         [ExpectedException(typeof(MqttCommunicationTimedOutException))]
         [TestMethod]
         public async Task TimeoutAfter()
         {
-            await MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct), TimeSpan.FromMilliseconds(100), CancellationToken.None);
+            await MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct),
+                TimeSpan.FromMilliseconds(100), CancellationToken.None);
         }
 
         [ExpectedException(typeof(MqttCommunicationTimedOutException))]
         [TestMethod]
         public async Task TimeoutAfterWithResult()
         {
-            await MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct).ContinueWith(t => 5, ct), TimeSpan.FromMilliseconds(100), CancellationToken.None);
+            await MqttTaskTimeout.WaitAsync(
+                ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct).ContinueWith(t => 5, ct),
+                TimeSpan.FromMilliseconds(100), CancellationToken.None);
         }
 
         [TestMethod]
         public async Task TimeoutAfterCompleteInTime()
         {
-            var result = await MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(100), ct).ContinueWith(t => 5, ct), TimeSpan.FromMilliseconds(500), CancellationToken.None);
+            var result = await MqttTaskTimeout.WaitAsync(
+                ct => Task.Delay(TimeSpan.FromMilliseconds(100), ct).ContinueWith(t => 5, ct),
+                TimeSpan.FromMilliseconds(500), CancellationToken.None);
             Assert.AreEqual(5, result);
         }
 
@@ -77,7 +82,8 @@ namespace MQTTnet.Tests
             var tasks = Enumerable.Range(0, 100000)
                 .Select(i =>
                 {
-                    return MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(1), ct), TimeSpan.FromMinutes(1), CancellationToken.None);
+                    return MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(1), ct),
+                        TimeSpan.FromMinutes(1), CancellationToken.None);
                 });
 
             await Task.WhenAll(tasks);
@@ -92,4 +98,4 @@ namespace MQTTnet.Tests
             }
         }
     }
-}
+}

Tests/MQTTnet.Core.Tests/MQTTnet.Tests.csproj

@@ -9,7 +9,7 @@
   <ItemGroup>
     <PackageReference Include="MSTest.TestAdapter" Version="1.4.0" />
     <PackageReference Include="MSTest.TestFramework" Version="1.4.0" />
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.9.0" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.0.1" />
   </ItemGroup>
 
   <ItemGroup>

Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs

@@ -12,7 +12,7 @@ using MQTTnet.Tests.Mockups;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class ManagedMqttClientTests
+    public class ManagedMqttClient_Tests
     {
         [TestMethod]
         public async Task Drop_New_Messages_On_Full_Queue()

Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs

@@ -21,7 +21,7 @@ namespace MQTTnet.Tests.Mockups
 
         private readonly List<Exception> _exceptions = new List<Exception>();
 
-        private IMqttServer _server;
+        public IMqttServer Server { get; private set; }
 
         public bool IgnoreClientLogErrors { get; set; }
 
@@ -72,15 +72,15 @@ namespace MQTTnet.Tests.Mockups
 
         public async Task<IMqttServer> StartServerAsync(MqttServerOptionsBuilder options)
         {
-            if (_server != null)
+            if (Server != null)
             {
                 throw new InvalidOperationException("Server already started.");
             }
 
-            _server = _mqttFactory.CreateMqttServer(_serverLogger);
-            await _server.StartAsync(options.WithDefaultEndpointPort(ServerPort).Build());
+            Server = _mqttFactory.CreateMqttServer(_serverLogger);
+            await Server.StartAsync(options.WithDefaultEndpointPort(ServerPort).Build());
 
-            return _server;
+            return Server;
         }
 
         public Task<IMqttClient> ConnectClientAsync()
@@ -122,7 +122,7 @@ namespace MQTTnet.Tests.Mockups
                 mqttClient?.Dispose();
             }
 
-            _server?.StopAsync().GetAwaiter().GetResult();
+            Server?.StopAsync().GetAwaiter().GetResult();
 
             ThrowIfLogErrors();
 

Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs

@@ -20,8 +20,11 @@ namespace MQTTnet.Tests.Mockups
 
         public MqttPacketFormatterAdapter PacketFormatterAdapter { get; } = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311);
 
-        public event EventHandler ReadingPacketStarted;
-        public event EventHandler ReadingPacketCompleted;
+        public long BytesSent { get; }
+        public long BytesReceived { get; }
+
+        public Action ReadingPacketStartedCallback { get; set; }
+        public Action ReadingPacketCompletedCallback { get; set; }
 
         public void Dispose()
         {

Tests/MQTTnet.Core.Tests/MqttApplicationMessageBuilder_Tests.cs

@@ -7,7 +7,7 @@ using MQTTnet.Protocol;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttApplicationMessageBuilderTests
+    public class MqttApplicationMessageBuilder_Tests
     {
         [TestMethod]
         public void CreateApplicationMessage_TopicOnly()

Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs

@@ -295,5 +295,49 @@ namespace MQTTnet.Tests
                 Assert.IsFalse(client.IsConnected);
             }
         }
+
+        [TestMethod]
+        public async Task Frequent_Connects()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+
+                var clients = new List<IMqttClient>();
+                for (var i = 0; i < 100; i++)
+                {
+                    clients.Add(await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithClientId("a")));
+                }
+
+                var clientStatus = await testEnvironment.Server.GetClientStatusAsync();
+                var sessionStatus = await testEnvironment.Server.GetSessionStatusAsync();
+
+                for (var i = 0; i < 98; i++)
+                {
+                    Assert.IsFalse(clients[i].IsConnected);
+                }
+                
+                Assert.IsTrue(clients[99].IsConnected);
+
+                Assert.AreEqual(1, clientStatus.Count);
+                Assert.AreEqual(1, sessionStatus.Count);
+
+                var receiveClient = clients[99];
+                object receivedPayload = null;
+                receiveClient.UseApplicationMessageReceivedHandler(e =>
+                {
+                    receivedPayload = e.ApplicationMessage.ConvertPayloadToString();
+                });
+
+                await receiveClient.SubscribeAsync("x");
+
+                var sendClient = await testEnvironment.ConnectClientAsync();
+                await sendClient.PublishAsync("x", "1");
+
+                await Task.Delay(100);
+
+                Assert.AreEqual("1", receivedPayload);
+            }
+        }
     }
 }

Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs

@@ -8,7 +8,7 @@ using System.Threading.Tasks;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttFactoryTests
+    public class MqttFactory_Tests
     {
         [TestMethod]
         public async Task Create_Managed_Client_With_Logger()

Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitor_Tests.cs

@@ -5,14 +5,13 @@ using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
-using MQTTnet.Packets;
 using MQTTnet.Server;
 using MQTTnet.Server.Status;
 
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttKeepAliveMonitorTests
+    public class MqttKeepAliveMonitor_Tests
     {
         [TestMethod]
         public void KeepAlive_Timeout()

Tests/MQTTnet.Core.Tests/MqttPacketIdentifierProvider_Tests.cs

@@ -4,7 +4,7 @@ using MQTTnet.Client;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttPacketIdentifierProviderTests
+    public class MqttPacketIdentifierProvider_Tests
     {
         [TestMethod]
         public void Reset()

Tests/MQTTnet.Core.Tests/MqttPacketReader_Tests.cs

@@ -8,7 +8,7 @@ using MQTTnet.Internal;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttPacketReaderTests
+    public class MqttPacketReader_Tests
     {
         [TestMethod]
         public async Task MqttPacketReader_EmptyStream()

Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs

@@ -14,7 +14,7 @@ using MQTTnet.Protocol;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttPacketSerializerTests
+    public class MqttPacketSerializer_Tests
     {
         [TestMethod]
         public void SerializeV310_MqttConnectPacket()

Tests/MQTTnet.Core.Tests/MqttPacketWriter_Tests.cs

@@ -4,7 +4,7 @@ using MQTTnet.Formatter;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttPacketWriterTests
+    public class MqttPacketWriter_Tests
     {
         [TestMethod]
         public void WritePacket()

Tests/MQTTnet.Core.Tests/MqttSubscriptionsManager_Tests.cs

@@ -6,7 +6,7 @@ using MQTTnet.Server;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class MqttSubscriptionsManagerTests
+    public class MqttSubscriptionsManager_Tests
     {
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeSingleSuccess()

Tests/MQTTnet.Core.Tests/MqttTcpChannel_Tests.cs

@@ -0,0 +1,74 @@
+using System;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Implementations;
+
+namespace MQTTnet.Tests
+{
+    [TestClass]
+    public class MqttTcpChannel_Tests
+    {
+        [TestMethod]
+        public async Task Dispose_Channel_While_Used()
+        {
+            var ct = new CancellationTokenSource();
+
+            try
+            {
+                var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+                serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50000));
+                serverSocket.Listen(0);
+
+#pragma warning disable 4014
+                Task.Run(async () =>
+#pragma warning restore 4014
+                {
+                    while (!ct.IsCancellationRequested)
+                    {
+                        var client = await serverSocket.AcceptAsync();
+                        var data = new byte[] { 128 };
+                        await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
+                    }
+                }, ct.Token);
+
+                var clientSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+                await clientSocket.ConnectAsync(IPAddress.Loopback, 50000);
+
+                await Task.Delay(100, ct.Token);
+
+                var tcpChannel = new MqttTcpChannel(new NetworkStream(clientSocket, true));
+
+                var buffer = new byte[1];
+                await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token);
+
+                Assert.AreEqual(128, buffer[0]);
+
+                // This block should fail after dispose.
+#pragma warning disable 4014
+                Task.Run(() =>
+#pragma warning restore 4014
+                {
+                    Task.Delay(200, ct.Token);
+                    tcpChannel.Dispose();
+                }, ct.Token);
+
+                try
+                {
+                    await tcpChannel.ReadAsync(buffer, 0, 1, CancellationToken.None);
+                }
+                catch (Exception exception)
+                {
+                    Assert.IsInstanceOfType(exception, typeof(SocketException));
+                    Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode);
+                }
+            }
+            finally
+            {
+                ct.Cancel(false);
+            }
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/RoundtripTime_Tests.cs

@@ -11,7 +11,7 @@ using MQTTnet.Server;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class RoundtripTimeTests
+    public class RoundtripTime_Tests
     {
         [TestMethod]
         public async Task Round_Trip_Time()

Tests/MQTTnet.Core.Tests/Server_Status_Tests.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Linq;
+using System.Linq;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using System.Threading.Tasks;
 using MQTTnet.Client.Options;
@@ -148,7 +147,9 @@ namespace MQTTnet.Tests
                     var clientStatus = await server.GetClientStatusAsync();
 
                     Assert.AreEqual(i, clientStatus.First().SentApplicationMessagesCount, "SAMC invalid!");
-                    Assert.AreEqual(i, clientStatus.First().SentPacketsCount, "SPC invalid!");
+
+                    // + 1 because CONNECT is also counted.
+                    Assert.AreEqual(i + 1, clientStatus.First().SentPacketsCount, "SPC invalid!");
 
                     // +1 because ConnACK package is already counted.
                     Assert.AreEqual(i + 1, clientStatus.First().ReceivedPacketsCount, "RPC invalid!");

Tests/MQTTnet.Core.Tests/TopicFilterComparer_Tests.cs

@@ -4,7 +4,7 @@ using MQTTnet.Server;
 namespace MQTTnet.Tests
 {
     [TestClass]
-    public class TopicFilterComparerTests
+    public class TopicFilterComparer_Tests
     {
         [TestMethod]
         public void TopicFilterComparer_DirectMatch()

Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs

@@ -160,7 +160,7 @@ namespace MQTTnet.TestApp.NetCore
                         msgCount += msgs.Count;
                         //send multiple
                     }
-                    
+
                     var now = DateTime.Now;
                     if (last < now - TimeSpan.FromSeconds(1))
                     {
@@ -193,5 +193,158 @@ namespace MQTTnet.TestApp.NetCore
             Interlocked.Increment(ref count);
             return Task.Run(() => client.PublishAsync(applicationMessage));
         }
+
+        public static async Task RunQoS2Test()
+        {
+            try
+            {
+                var mqttServer = new MqttFactory().CreateMqttServer();
+                await mqttServer.StartAsync(new MqttServerOptions());
+
+                var options = new MqttClientOptions
+                {
+                    ChannelOptions = new MqttClientTcpOptions
+                    {
+                        Server = "127.0.0.1"
+                    },
+                    CleanSession = true
+                };
+
+                var client = new MqttFactory().CreateMqttClient();
+                await client.ConnectAsync(options);
+
+                var message = new MqttApplicationMessage
+                {
+                    Topic = "A/B/C",
+                    Payload = Encoding.UTF8.GetBytes("Hello World"),
+                    QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce
+                };
+
+                var stopwatch = new Stopwatch();
+
+                var iteration = 1;
+                while (true)
+                {
+                    stopwatch.Restart();
+
+                    var sentMessagesCount = 0;
+                    while (stopwatch.ElapsedMilliseconds < 1000)
+                    {
+                        await client.PublishAsync(message).ConfigureAwait(false);
+                        sentMessagesCount++;
+                    }
+
+                    Console.WriteLine($"Sent {sentMessagesCount} messages in iteration #" + iteration);
+
+                    iteration++;
+                }
+            }
+            catch (Exception exception)
+            {
+                Console.WriteLine(exception);
+            }
+        }
+
+        public static async Task RunQoS1Test()
+        {
+            try
+            {
+                var mqttServer = new MqttFactory().CreateMqttServer();
+                await mqttServer.StartAsync(new MqttServerOptions());
+
+                var options = new MqttClientOptions
+                {
+                    ChannelOptions = new MqttClientTcpOptions
+                    {
+                        Server = "127.0.0.1"
+                    },
+                    CleanSession = true
+                };
+
+                var client = new MqttFactory().CreateMqttClient();
+                await client.ConnectAsync(options);
+
+                var message = new MqttApplicationMessage
+                {
+                    Topic = "A/B/C",
+                    Payload = Encoding.UTF8.GetBytes("Hello World"),
+                    QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce
+                };
+
+                var stopwatch = new Stopwatch();
+
+                var iteration = 1;
+                while (true)
+                {
+                    stopwatch.Restart();
+
+                    var sentMessagesCount = 0;
+                    while (stopwatch.ElapsedMilliseconds < 1000)
+                    {
+                        await client.PublishAsync(message).ConfigureAwait(false);
+                        sentMessagesCount++;
+                    }
+
+                    Console.WriteLine($"Sent {sentMessagesCount} messages in iteration #" + iteration);
+
+                    iteration++;
+                }
+            }
+            catch (Exception exception)
+            {
+                Console.WriteLine(exception);
+            }
+        }
+
+        public static async Task RunQoS0Test()
+        {
+            try
+            {
+                var mqttServer = new MqttFactory().CreateMqttServer();
+                await mqttServer.StartAsync(new MqttServerOptions());
+
+                var options = new MqttClientOptions
+                {
+                    ChannelOptions = new MqttClientTcpOptions
+                    {
+                        Server = "127.0.0.1"
+                    },
+                    CleanSession = true
+                };
+
+                var client = new MqttFactory().CreateMqttClient();
+                await client.ConnectAsync(options);
+
+                var message = new MqttApplicationMessage
+                {
+                    Topic = "A/B/C",
+                    Payload = Encoding.UTF8.GetBytes("Hello World"),
+                    QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
+                };
+
+                var stopwatch = new Stopwatch();
+
+                var iteration = 1;
+                while (true)
+                {
+                    stopwatch.Restart();
+
+                    var sentMessagesCount = 0;
+                    while (stopwatch.ElapsedMilliseconds < 1000)
+                    {
+                        await client.PublishAsync(message).ConfigureAwait(false);
+                        sentMessagesCount++;
+                    }
+
+                    Console.WriteLine($"Sent {sentMessagesCount} messages in iteration #" + iteration);
+
+                    iteration++;
+                }
+            }
+            catch (Exception exception)
+            {
+                Console.WriteLine(exception);
+            }
+        }
     }
 }

Tests/MQTTnet.TestApp.NetCore/Program.cs

@@ -25,6 +25,9 @@ namespace MQTTnet.TestApp.NetCore
             Console.WriteLine("7 = Client flow test");
             Console.WriteLine("8 = Start performance test (client only)");
             Console.WriteLine("9 = Start server (no trace)");
+            Console.WriteLine("a = Start QoS 2 benchmark");
+            Console.WriteLine("b = Start QoS 1 benchmark");
+            Console.WriteLine("c = Start QoS 0 benchmark");
 
             var pressedKey = Console.ReadKey(true);
             if (pressedKey.KeyChar == '1')
@@ -66,61 +69,21 @@ namespace MQTTnet.TestApp.NetCore
                 ServerTest.RunEmptyServer();
                 return;
             }
-
+            else if (pressedKey.KeyChar == 'a')
+            {
+                Task.Run(PerformanceTest.RunQoS2Test);
+            }
+            else if (pressedKey.KeyChar == 'b')
+            {
+                Task.Run(PerformanceTest.RunQoS1Test);
+            }
+            else if (pressedKey.KeyChar == 'c')
+            {
+                Task.Run(PerformanceTest.RunQoS0Test);
+            }
+            
             Thread.Sleep(Timeout.Infinite);
         }
-
-        // This code is used at the Wiki on GitHub!
-        // ReSharper disable once UnusedMember.Local
-        private static async void WikiCode()
-        {
-            {
-                var client = new MqttFactory().CreateMqttClient();
-
-                var options = new MqttClientOptionsBuilder()
-                    .WithClientId("Client1")
-                    .WithTcpServer("broker.hivemq.com")
-                    .WithCredentials("bud", "%spencer%")
-                    .WithTls()
-                    .Build();
-
-                await client.ConnectAsync(options);
-
-                var message = new MqttApplicationMessageBuilder()
-                    .WithTopic("MyTopic")
-                    .WithPayload("Hello World")
-                    .WithExactlyOnceQoS()
-                    .WithRetainFlag()
-                    .Build();
-
-                await client.PublishAsync(message);
-            }
-
-            {
-                var factory = new MqttFactory();
-                var client = factory.CreateMqttClient();
-            }
-
-            {
-                // Write all trace messages to the console window.
-                MqttNetGlobalLogger.LogMessagePublished += (s, e) =>
-                {
-                    var trace = $">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}";
-                    if (e.TraceMessage.Exception != null)
-                    {
-                        trace += Environment.NewLine + e.TraceMessage.Exception.ToString();
-                    }
-
-                    Console.WriteLine(trace);
-                };
-            }
-
-            {
-                // Use a custom log ID for the logger.
-                var factory = new MqttFactory();
-                var mqttClient = factory.CreateMqttClient(new MqttNetLogger("MyCustomId"));
-            }
-        }
     }
 
     public class RetainedMessageHandler : IMqttServerStorage

Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj

@@ -12,7 +12,7 @@
     <DefaultLanguage>en-US</DefaultLanguage>
     <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
     <TargetPlatformVersion Condition=" '$(TargetPlatformVersion)' == '' ">10.0.17134.0</TargetPlatformVersion>
-    <TargetPlatformMinVersion>10.0.16299.0</TargetPlatformMinVersion>
+    <TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
     <MinimumVisualStudioVersion>14</MinimumVisualStudioVersion>
     <FileAlignment>512</FileAlignment>
     <ProjectTypeGuids>{A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
@@ -142,10 +142,10 @@
   </ItemGroup>
   <ItemGroup>
     <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform">
-      <Version>6.2.2</Version>
+      <Version>6.2.8</Version>
     </PackageReference>
     <PackageReference Include="Microsoft.Toolkit.Uwp.UI.Controls">
-      <Version>5.0.0</Version>
+      <Version>4.0.0</Version>
     </PackageReference>
   </ItemGroup>
   <PropertyGroup Condition=" '$(VisualStudioVersion)' == '' or '$(VisualStudioVersion)' &lt; '14.0' ">

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

@@ -10,7 +10,6 @@ using MQTTnet.Client;
 using MQTTnet.Client.Connecting;
 using MQTTnet.Client.Disconnecting;
 using MQTTnet.Client.Options;
-using MQTTnet.Client.Receiving;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Extensions.ManagedClient;
@@ -469,18 +468,6 @@ namespace MQTTnet.TestApp.UniversalWindows
 
         private async Task WikiCode()
         {
-            {
-                // Write all trace messages to the console window.
-                MqttNetGlobalLogger.LogMessagePublished += (s, e) =>
-                {
-                    Console.WriteLine($">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}");
-                    if (e.TraceMessage.Exception != null)
-                    {
-                        Console.WriteLine(e.TraceMessage.Exception);
-                    }
-                };
-            }
-
             {
                 // Use a custom identifier for the trace messages.
                 var clientOptions = new MqttClientOptionsBuilder()
@@ -490,20 +477,77 @@ namespace MQTTnet.TestApp.UniversalWindows
             {
                 // Create a new MQTT client.
                 var factory = new MqttFactory();
-                var mqttClient = factory.CreateMqttClient();
+                var client = factory.CreateMqttClient();
 
+                // Create TCP based options using the builder.
+                var options = new MqttClientOptionsBuilder()
+                    .WithClientId("Client1")
+                    .WithTcpServer("broker.hivemq.com")
+                    .WithCredentials("bud", "%spencer%")
+                    .WithTls()
+                    .WithCleanSession()
+                    .Build();
+
+                await client.ConnectAsync(options);
+
+                // Reconnecting
+
+                client.UseDisconnectedHandler(async e =>
                 {
-                    // Create TCP based options using the builder.
-                    var options = new MqttClientOptionsBuilder()
-                        .WithClientId("Client1")
-                        .WithTcpServer("broker.hivemq.com")
-                        .WithCredentials("bud", "%spencer%")
-                        .WithTls()
-                        .WithCleanSession()
-                        .Build();
+                    Console.WriteLine("### DISCONNECTED FROM SERVER ###");
+                    await Task.Delay(TimeSpan.FromSeconds(5));
 
-                    await mqttClient.ConnectAsync(options);
-                }
+                    try
+                    {
+                        await client.ConnectAsync(options);
+                    }
+                    catch
+                    {
+                        Console.WriteLine("### RECONNECTING FAILED ###");
+                    }
+                });
+
+                // Consuming messages
+
+                client.UseApplicationMessageReceivedHandler(e =>
+                {
+                    Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
+                    Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");
+                    Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
+                    Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
+                    Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}");
+                    Console.WriteLine();
+                });
+
+                // Subscribe after connect
+
+                client.UseConnectedHandler(async e =>
+                {
+                    Console.WriteLine("### CONNECTED WITH SERVER ###");
+
+                    // Subscribe to a topic
+                    await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build());
+
+                    Console.WriteLine("### SUBSCRIBED ###");
+                });
+
+                // Subscribe to a topic
+                await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build());
+
+                // Unsubscribe from a topic
+                await client.UnsubscribeAsync("my/topic");
+
+                // Publish an application message
+                var applicationMessage = new MqttApplicationMessageBuilder()
+                    .WithTopic("A/B/C")
+                    .WithPayload("Hello World")
+                    .WithAtLeastOnceQoS()
+                    .Build();
+
+                await client.PublishAsync(applicationMessage);
+            }
+
+            {
 
                 {
                     // Use TCP connection.
@@ -525,8 +569,6 @@ namespace MQTTnet.TestApp.UniversalWindows
                     var options = new MqttClientOptionsBuilder()
                         .WithWebSocketServer("broker.hivemq.com:8000/mqtt")
                         .Build();
-
-                    await mqttClient.ConnectAsync(options);
                 }
 
                 {
@@ -549,23 +591,6 @@ namespace MQTTnet.TestApp.UniversalWindows
                         },
                     };
                 }
-
-                {
-                    // Subscribe to a topic
-                    await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build());
-
-                    // Unsubscribe from a topic
-                    await mqttClient.UnsubscribeAsync("my/topic");
-
-                    // Publish an application message
-                    var applicationMessage = new MqttApplicationMessageBuilder()
-                        .WithTopic("A/B/C")
-                        .WithPayload("Hello World")
-                        .WithAtLeastOnceQoS()
-                        .Build();
-
-                    await mqttClient.PublishAsync(applicationMessage);
-                }
             }
 
             // ----------------------------------
@@ -700,6 +725,39 @@ namespace MQTTnet.TestApp.UniversalWindows
                 await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build());
                 await mqttClient.StartAsync(options);
             }
+
+            {
+                // Use a custom log ID for the logger.
+                var factory = new MqttFactory();
+                var client = factory.CreateMqttClient(new MqttNetLogger("MyCustomId"));
+            }
+
+            {
+                var client = new MqttFactory().CreateMqttClient();
+
+                var message = new MqttApplicationMessageBuilder()
+                    .WithTopic("MyTopic")
+                    .WithPayload("Hello World")
+                    .WithExactlyOnceQoS()
+                    .WithRetainFlag()
+                    .Build();
+
+                await client.PublishAsync(message);
+            }
+
+            {
+                // Write all trace messages to the console window.
+                MqttNetGlobalLogger.LogMessagePublished += (s, e) =>
+                {
+                    var trace = $">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}";
+                    if (e.TraceMessage.Exception != null)
+                    {
+                        trace += Environment.NewLine + e.TraceMessage.Exception.ToString();
+                    }
+
+                    Console.WriteLine(trace);
+                };
+            }
         }
 
         #endregion