Update project files.

Based on https://github.com/chkr1011/MQTTnet/issues/1138#issuecomment-846649937

.github/workflows/codeql-analysis.yml

@@ -0,0 +1,67 @@
+# For most projects, this workflow file will not need changing; you simply need
+# to commit it to your repository.
+#
+# You may wish to alter this file to override the set of languages analyzed,
+# or to provide custom queries or build logic.
+#
+# ******** NOTE ********
+# We have attempted to detect the languages in your repository. Please check
+# the `language` matrix defined below to confirm you have the correct set of
+# supported CodeQL languages.
+#
+name: "CodeQL"
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    # The branches below must be a subset of the branches above
+    branches: [ master ]
+  schedule:
+    - cron: '23 9 * * 6'
+
+jobs:
+  analyze:
+    name: Analyze
+    runs-on: ubuntu-latest
+
+    strategy:
+      fail-fast: false
+      matrix:
+        language: [ 'csharp' ]
+        # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python' ]
+        # Learn more:
+        # https://docs.github.com/en/free-pro-team@latest/github/finding-security-vulnerabilities-and-errors-in-your-code/configuring-code-scanning#changing-the-languages-that-are-analyzed
+
+    steps:
+    - name: Checkout repository
+      uses: actions/checkout@v2
+
+    # Initializes the CodeQL tools for scanning.
+    - name: Initialize CodeQL
+      uses: github/codeql-action/init@v1
+      with:
+        languages: ${{ matrix.language }}
+        # If you wish to specify custom queries, you can do so here or in a config file.
+        # By default, queries listed here will override any specified in a config file.
+        # Prefix the list here with "+" to use these queries and those in the config file.
+        # queries: ./path/to/local/query, your-org/your-repo/queries@main
+
+    # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
+    # If this step fails, then you should remove it and run the build manually (see below)
+    - name: Autobuild
+      uses: github/codeql-action/autobuild@v1
+
+    # ℹ️ Command-line programs to run using the OS shell.
+    # 📚 https://git.io/JvXDl
+
+    # ✏️ If the Autobuild fails above, remove it and uncomment the following three lines
+    #    and modify them (or add more) to build your code if your project
+    #    uses a compiled language
+
+    #- run: |
+    #   make bootstrap
+    #   make release
+
+    - name: Perform CodeQL Analysis
+      uses: github/codeql-action/analyze@v1

.github/workflows/deb-packaging.yml

@@ -0,0 +1,62 @@
+# This is a basic workflow to help you get started with Actions
+
+name: Deb Packaging
+
+# Controls when the action will run. 
+on:
+  # Allows you to run this workflow manually from the Actions tab
+  workflow_dispatch
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Parse version
+        run: echo VERSION=$(grep '<Version>' Source/MQTTnet.Server/MQTTnet.Server.csproj | grep [0-9\.]* -o) >> $GITHUB_ENV
+
+      - name: Build MQTTnetServer
+        run: dotnet publish Source/MQTTnet.Server/MQTTnet.Server.csproj --configuration Release --self-contained --runtime linux-x64 --framework net5.0
+
+      - name: Set up installation directory
+        run: mkdir -p packaging/opt/MQTTnetServer
+
+      - name: Copy meta data
+        run: |
+          cp -R Build/deb-meta/* packaging/
+          chmod 755 packaging/DEBIAN/postinst
+          chmod 755 packaging/DEBIAN/postrm
+          chmod 755 packaging/DEBIAN/prerm
+
+      - name: Move artifacts to packaging directory
+        run: cp -R Source/MQTTnet.Server/bin/Release/net5.0/linux-x64/publish/* packaging/opt/MQTTnetServer
+
+      - name: Adjust files
+        run: |
+          rm packaging/opt/MQTTnetServer/appsettings.Development.json
+          mv packaging/opt/MQTTnetServer/appsettings.json packaging/opt/MQTTnetServer/appsettings.template.json
+
+      - name: Adjust permissions
+        run: |
+          cd packaging/opt/MQTTnetServer
+          find . -type f | xargs chmod -R 644
+          chmod 755 MQTTnet.Server
+
+      - name: Generate MD5s
+        run: |
+          cd packaging/          
+          md5sum $(find * -type f -not -path 'DEBIAN/*') > DEBIAN/md5sums
+      
+      - name: Patch meta
+        run: sed -i 's/\VERSIONPLACEHOLDER/${{ env.VERSION }}/' packaging/DEBIAN/control
+
+      - name: Package everything
+        run: dpkg-deb -v --build packaging/ mqttnet-server_${{ env.VERSION }}-1_amd64.deb
+      
+      - name: Save artifact
+        uses: actions/upload-artifact@v2
+        with:
+          name: mqttnet-server
+          path: mqttnet-server_${{ env.VERSION }}-1_amd64.deb

.gitignore

@@ -294,3 +294,5 @@ __pycache__/
 /Tests/MQTTnet.TestApp.NetCore/RetainedMessages.json
 
 Build/NuGet/
+
+Build/nuget.exe

Build/MQTTnet.AspNetCore.nuspec

@@ -38,7 +38,7 @@
     <file src="..\Images\Logo_128x128.png" target="images\" />
 
     <!-- .NET 5.0 -->
-    <file src="..\Source\MQTTnet.AspNetCore\bin\Release\net5.0\MQTTnet.*" target="lib\net5.0\" />
+    <file src="..\Source\MQTTnet.AspNetCore\bin\Release\net5.0\MQTTnet.AspNetCore.*" target="lib\net5.0\" />
 
     <!-- .NET Standard 2.0 -->
     <file src="..\Source\MQTTnet.AspNetCore\bin\Release\netstandard2.0\MQTTnet.AspNetCore.*" target="lib\netstandard2.0\" />

Build/MQTTnet.Extensions.ManagedClient.nuspec

@@ -27,7 +27,7 @@
     <file src="..\Images\Logo_128x128.png" target="images\" />
 
     <!-- .NET 5.0 -->
-    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\net5.0\MQTTnet.*" target="lib\net5.0\" />
+    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\net5.0\MQTTnet.Extensions.ManagedClient.*" target="lib\net5.0\" />
 
     <!-- .NET Standard 1.3 -->
     <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard1.3\MQTTnet.Extensions.ManagedClient.*" target="lib\netstandard1.3\" />
@@ -35,6 +35,9 @@
     <!-- .NET Standard 2.0 -->
     <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard2.0\MQTTnet.Extensions.ManagedClient.*" target="lib\netstandard2.0\" />
 
+    <!-- .NET Standard 2.1 -->
+    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard2.1\MQTTnet.Extensions.ManagedClient.*" target="lib\netstandard2.1\" />
+
     <!-- Universal Windows -->
     <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\uap10.0\MQTTnet.Extensions.ManagedClient.*" target="lib\uap10.0\" />
 
@@ -42,7 +45,7 @@
     <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\net452\MQTTnet.Extensions.ManagedClient.*" target="lib\net452\" />
 
     <!-- .NET Framework 4.6.0 will use binaries from 4.5.2. -->
-    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\net461\MQTTnet.Extensions.Rpc.*" target="lib\net461\" />
+    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\net461\MQTTnet.Extensions.ManagedClient.*" target="lib\net461\" />
 
     <!-- Everything above .NET Framework 4.6.1 will use the binaries from 4.6.1. -->
   </files>

Build/MQTTnet.Extensions.Rpc.nuspec

@@ -27,7 +27,7 @@
     <file src="..\Images\Logo_128x128.png" target="images\" />
 
     <!-- .NET 5.0 -->
-    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\net5.0\MQTTnet.*" target="lib\net5.0\" />
+    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\net5.0\MQTTnet.Extensions.Rpc.*" target="lib\net5.0\" />
 
     <!-- .NET Standard 1.3 -->
     <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard1.3\MQTTnet.Extensions.Rpc.*" target="lib\netstandard1.3\" />
@@ -35,6 +35,9 @@
     <!-- .NET Standard 2.0 -->
     <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard2.0\MQTTnet.Extensions.Rpc.*" target="lib\netstandard2.0\" />
 
+    <!-- .NET Standard 2.1 -->
+    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard2.1\MQTTnet.Extensions.Rpc.*" target="lib\netstandard2.1\" />
+
     <!-- Universal Windows -->
     <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\uap10.0\MQTTnet.Extensions.Rpc.*" target="lib\uap10.0\" />
 

Build/MQTTnet.Extensions.WebSocket4Net.nuspec

@@ -15,22 +15,8 @@
     <copyright>Copyright Christian Kratky 2016-2020</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</tags>
     <dependencies>
-    <group targetFramework="netstandard1.3">
+      <dependency id="MQTTnet" version="$nugetVersion" />
-        <dependency id="MQTTnet" version="$nugetVersion" />
+      <dependency id="WebSocket4Net" version="0.15.2" />
-        <dependency id="WebSocket4Net" version="0.15.2" />
-      </group>
-      <group targetFramework="netstandard2.0">
-        <dependency id="MQTTnet" version="$nugetVersion" />
-        <dependency id="WebSocket4Net" version="0.15.2" />
-      </group>
-      <group targetFramework="netstandard2.1">
-        <dependency id="MQTTnet" version="$nugetVersion" />
-        <dependency id="WebSocket4Net" version="0.15.2" />
-      </group>
-      <group targetFramework="uap10.0">
-        <dependency id="MQTTnet" version="$nugetVersion" />
-        <dependency id="WebSocket4Net" version="0.15.2" />
-      </group>
     </dependencies>
   </metadata>
 
@@ -42,7 +28,7 @@
     <file src="..\Images\Logo_128x128.png" target="images\" />
 
     <!-- .NET 5.0 -->
-    <file src="..\Source\MQTTnet.Extensions.WebSocket4Net\bin\Release\net5.0\MQTTnet.*" target="lib\net5.0\" />
+    <file src="..\Source\MQTTnet.Extensions.WebSocket4Net\bin\Release\net5.0\MQTTnet.Extensions.WebSocket4Net.*" target="lib\net5.0\" />
 
     <!-- .NET Standard 1.3 -->
     <file src="..\Source\MQTTnet.Extensions.WebSocket4Net\bin\Release\netstandard1.3\MQTTnet.Extensions.WebSocket4Net.*" target="lib\netstandard1.3\" />
@@ -50,6 +36,9 @@
     <!-- .NET Standard 2.0 -->
     <file src="..\Source\MQTTnet.Extensions.WebSocket4Net\bin\Release\netstandard2.0\MQTTnet.Extensions.WebSocket4Net.*" target="lib\netstandard2.0\" />
 
+    <!-- .NET Standard 2.1 -->
+    <file src="..\Source\MQTTnet.Extensions.WebSocket4Net\bin\Release\netstandard2.1\MQTTnet.Extensions.WebSocket4Net.*" target="lib\netstandard2.1\" />
+
     <!-- Universal Windows -->
     <file src="..\Source\MQTTnet.Extensions.WebSocket4Net\bin\Release\uap10.0\MQTTnet.Extensions.WebSocket4Net.*" target="lib\uap10.0\" />
 

Build/MQTTnet.nuspec

@@ -12,54 +12,32 @@
     <requireLicenseAcceptance>true</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</description>
     <releaseNotes>
-* [Core] Added .NET 5.0 TFMs (thanks to @StefanOssendorf, @JanEggers).
+* [Core] Fixed _DISCONNECT_ packet reading for protocol version 3.1.0.DisconnectPacket
-* [Core] Added support for TLS 1.3 (requires .NET Core 3.1+) (thanks to @Dvergatal).
+* [Client] Added support for deferred message approval (#1075, thanks to @tkurucsai).
-* [Core] Added support for _MaximumQoS_ transmit when using MQTTv5 (thanks to @nayato).
+* [Client] Exposed missing APIs for .NET 5.0 build (thanks to @yyjdelete).
-* [Core] Aligned the format of some log messages.
+* [Client] Improved disconnect handling (#1134, thanks to @yyjdelete).
-* [Client] Fixed an issue with connection state detection (thanks to @tobiasfrick).
+* [Client] Fix TLS parameter options builder (#1104).
-* [Client] Fixed an issue when receiving partial QoS 2 packets after a reconnect.
+* [Server] Add new overload for options builder.
-* [Client] Added new API for raw packet inspection.
+* [Server] Exposed endpoint in _MqttServerClientDisconnectedEventArgs_.
-* [ManagedClient] Removed _AutoReconnect_ since this does no longer work (BREAKING CHANGE!).
+* [Server] Remove client disconnected handler from wrong implementation location (BREAKING CHANGE).
-* [RpcClient] Moved the RPC topic validation to the topic generation strategy.
-* [RcpClient] Added interface for _MqttRpcClient_.
-* [Server] Reduced async tasks count by moving dedicated keep alive tasks per connection to shared one.
-* [Server] Session takeover and keep alive timeout are now properly set in DISCONNECT packet.
-* [Server] Fixed bug in PubRel packet generation (MQTTv5 only).
-* [Server] Improved message processing performance (+ ~5%).
-* [Server] Fix wrong usage of client session items for undelivered messages.
-* [Server] Allow to respond with a reason code in PUBACK/PUBREC (thanks to @muneebmajeed).
-* [Server] Fixed topic filter comparer on edge cases, e.g. "foo/" matching "foo/#" (thanks to @dagophil).
-* [Core] Added code documentation.
 
 Git commit: $gitCommit
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2020</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</tags>
     <dependencies>
+      <group targetFramework="net452" />
+      <group targetFramework="net461" />
       <group targetFramework="netstandard1.3">
-        <dependency id="NETStandard.Library" version="1.3.0" />
+        <dependency id="NETStandard.Library" version="1.6.1" />
-        <dependency id="System.Net.Security" version="4.3.2" />
-        <dependency id="System.Net.WebSockets" version="4.3.0" />
-        <dependency id="System.Net.WebSockets.Client" version="4.3.2" />
-      </group>
-      <group targetFramework="netstandard2.0">
-        <dependency id="NETStandard.Library" version="2.0.0" />
-        <dependency id="System.Net.Security" version="4.3.2" />
-        <dependency id="System.Net.WebSockets" version="4.3.0" />
-        <dependency id="System.Net.WebSockets.Client" version="4.3.2" />
-      </group>
-      <group targetFramework="netstandard2.1">
-        <dependency id="NETStandard.Library" version="2.0.0" />
-        <dependency id="System.Net.Security" version="4.3.2" />
-        <dependency id="System.Net.WebSockets" version="4.3.0" />
-        <dependency id="System.Net.WebSockets.Client" version="4.3.2" />
-      </group>
-      <group targetFramework="netcoreapp3.1">
-        <dependency id="NETStandard.Library" version="2.0.0" />
         <dependency id="System.Net.Security" version="4.3.2" />
         <dependency id="System.Net.WebSockets" version="4.3.0" />
         <dependency id="System.Net.WebSockets.Client" version="4.3.2" />
       </group>
+      <group targetFramework="netstandard2.0" />
+      <group targetFramework="netstandard2.1" />
+      <group targetFramework="netcoreapp3.1" />
+      <group targetFramework="net5.0" />
       <group targetFramework="uap10.0">
         <dependency id="Microsoft.NETCore.UniversalWindowsPlatform" version="6.2.10" />
       </group>

Build/build.ps1

@@ -31,10 +31,12 @@ vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\M
 &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.1" /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="netcoreapp3.1" /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"
+&$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net5.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"
 &$msbuild ..\Source\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netcoreapp3.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
+&$msbuild ..\Source\MQTTnet.AspNetCore\MQTTnet.AspNetCore.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net5.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"
@@ -43,6 +45,7 @@ vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\M
 &$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="netstandard2.1" /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"
+&$msbuild ..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net5.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"
@@ -51,6 +54,7 @@ vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\M
 &$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="netstandard2.1" /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"
+&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net5.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 
 # Build the WebSocket4Net extension
 &$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.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"
@@ -59,6 +63,7 @@ vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\M
 &$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.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.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.1" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 &$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.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"
+&$msbuild ..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net5.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
 
 # Create NuGet packages.
 
@@ -97,7 +102,7 @@ Remove-Item "nuget.exe" -Force -Recurse -ErrorAction SilentlyContinue
 # Build MQTTnet.Server Portable
 &dotnet publish ..\Source\MQTTnet.Server\MQTTnet.Server.csproj --configuration Release /p:FileVersion=$assemblyVersion /p:Version=$nugetVersion --framework net5.0
 
-$source = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\Release\netcoreapp3.1\publish"
+$source = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\Release\net5.0\publish"
 $destination = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\MQTTnet.Server-Portable-v$nugetVersion.zip"
 If(Test-path $destination) {Remove-item $destination}
  Add-Type -assembly "system.io.compression.filesystem"
@@ -108,7 +113,7 @@ If(Test-path $destination) {Remove-item $destination}
 # Build MQTTnet.Server Linux-x64
 &dotnet publish ..\Source\MQTTnet.Server\MQTTnet.Server.csproj --configuration Release /p:FileVersion=$assemblyVersion /p:Version=$nugetVersion --self-contained --runtime linux-x64 --framework net5.0
 
-$source = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\Release\netcoreapp3.1\linux-x64\publish"
+$source = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\Release\net5.0\linux-x64\publish"
 $destination = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\MQTTnet.Server-Linux-x64-v$nugetVersion.zip"
 If(Test-path $destination) {Remove-item $destination}
  Add-Type -assembly "system.io.compression.filesystem"
@@ -119,7 +124,7 @@ If(Test-path $destination) {Remove-item $destination}
 # Build MQTTnet.Server Linux-ARM
 &dotnet publish ..\Source\MQTTnet.Server\MQTTnet.Server.csproj --configuration Release /p:FileVersion=$assemblyVersion /p:Version=$nugetVersion --self-contained --runtime linux-arm --framework net5.0
 
-$source = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\Release\netcoreapp3.1\linux-ARM\publish"
+$source = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\Release\net5.0\linux-ARM\publish"
 $destination = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\MQTTnet.Server-Linux-ARM-v$nugetVersion.zip"
 If(Test-path $destination) {Remove-item $destination}
  Add-Type -assembly "system.io.compression.filesystem"
@@ -130,7 +135,7 @@ If(Test-path $destination) {Remove-item $destination}
 # Build MQTTnet.Server Windows-x64
 &dotnet publish ..\Source\MQTTnet.Server\MQTTnet.Server.csproj --configuration Release /p:FileVersion=$assemblyVersion /p:Version=$nugetVersion --self-contained --runtime win-x64 --framework net5.0
 
-$source = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\Release\netcoreapp3.1\win-x64\publish"
+$source = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\Release\net5.0\win-x64\publish"
 $destination = (Convert-Path .) + "\..\Source\MQTTnet.Server\bin\MQTTnet.Server-Windows-x64-v$nugetVersion.zip"
 If(Test-path $destination) {Remove-item $destination}
  Add-Type -assembly "system.io.compression.filesystem"

Build/deb-meta/DEBIAN/control

@@ -0,0 +1,10 @@
+Package: mqttnet-server
+Version: VERSIONPLACEHOLDER
+Section: base
+Priority: optional
+Architecture: amd64
+Depends: libssl1.1
+Suggests:
+Maintainer: Damian Wolgast <packages@asymetrixs.net>
+Description:
+  MQTTnet Server is a standalone cross platform MQTT server (like mosquitto) basing on this library.

Build/deb-meta/DEBIAN/copyright

@@ -0,0 +1,26 @@
+The source of MQTTnet is at:
+https://github.com/chkr1011/MQTTnet
+Project is maintained by Christian Kratky
+
+
+MIT License
+
+Copyright (c) 2021 Damian Wolgast
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Build/deb-meta/DEBIAN/postinst

@@ -0,0 +1,44 @@
+#!/bin/bash
+
+# Check if user exits
+USER=mqttnetsrv
+USEREXISTS=$(id -u $USER >/dev/null 2>&1; echo $?)
+EC=0
+BASEPATH=/opt
+INSTALLPATH=$BASEPATH/MQTTnetServer
+
+# If user does not exist, create it
+if (( $USEREXISTS == 1 )); then
+    useradd -d $INSTALLPATH  -r $USER
+    EC=$(echo $?)
+
+    if (( $EC == 1 )); then
+        exit 1;
+    fi
+fi
+
+# Set permissions on files
+chown -R root:root $INSTALLPATH
+
+# Check if config does not exist and create it
+if [ ! -f "$INSTALLPATH/appsettings.json" ]; then
+    cp $INSTALLPATH/appsettings.template.json $INSTALLPATH/appsettings.json
+fi
+
+
+# Check if setcap exists, does not exist in docker containers
+SCAP=$(which setcap >/dev/null; echo $?)
+if [ "$SCAP" -eq "0" ]; then
+    # Enable MQTTServer.NET to bind to IP interface
+    setcap CAP_NET_BIND_SERVICE=+eip $INSTALLPATH/MQTTnet.Server
+fi
+
+chmod 644 /etc/systemd/system/mqttnet-server.service
+
+# Check if systemctl exists, does not exist in docker containers
+SCTL=$(which systemctl >/dev/null; echo $?)
+
+if [ "$SCTL" -eq "0" ]; then
+    # Reload systemd because of new service file
+    systemctl daemon-reload
+fi

Build/deb-meta/DEBIAN/postrm

@@ -0,0 +1,8 @@
+#!/bin/bash
+
+# Check if systemctl exists, does not exist in docker containers
+SCTL=$(which systemctl >/dev/null; echo $?)
+if [ "$SCTL" -eq "0" ]; then
+    # Reload systemd because of new service file
+    systemctl daemon-reload
+fi

Build/deb-meta/DEBIAN/prerm

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+BASEPATH=/opt
+INSTALLPATH=$BASEPATH/MQTTnetServer
+
+# Check if systemctl exists, does not exist in docker containers
+SCTL=$(which systemctl >/dev/null; echo $?)
+if [ "$SCTL" -eq "0" ]; then
+    # Stop service before uninstalling
+    systemctl stop mqttnet-server
+fi
+
+# Check if setcap exists, does not exist in docker containers
+SCAP=$(which setcap >/dev/null; echo $?)
+if [ "$SCAP" -eq "0" ]; then
+    # Remove permission to open ports
+    setcap CAP_NET_BIND_SERVICE=-eip $INSTALLPATH/MQTTnet.Server
+fi

Build/deb-meta/etc/systemd/system/mqttnet-server.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=MQTTnetServer
+After=network.target
+
+[Service]
+ExecStart=/opt/MQTTnetServer/MQTTnet.Server
+WorkingDirectory=/opt/MQTTnetServer
+StandardOutput=inherit
+StandardError=inherit
+Restart=always
+User=mqttnetsrv
+
+[Install]
+WantedBy=multi-user.target

MQTTnet.sln.DotSettings

@@ -12,4 +12,5 @@
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=unsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;net5.0</TargetFrameworks>
@@ -8,6 +8,9 @@
     <Company />
     <Authors />
     <PackageId />
+    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
+    <SignAssembly>false</SignAssembly>
+    <DelaySign>false</DelaySign>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>

Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs

@@ -7,12 +7,8 @@ namespace MQTTnet.Extensions.Rpc
 {
     public interface IMqttRpcClient : IDisposable
     {
-        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel);
-
-        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken);
-
         Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel);
 
-        Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken);
+        Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken);
     }
 }

Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;net5.0</TargetFrameworks>
@@ -8,6 +8,9 @@
     <Company />
     <Authors />
     <PackageId />
+    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
+    <SignAssembly>false</SignAssembly>
+    <DelaySign>false</DelaySign>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -5,9 +5,10 @@ using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
 using MQTTnet.Protocol;
 using System;
 using System.Collections.Concurrent;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Client.Subscribing;
+using MQTTnet.Implementations;
 
 namespace MQTTnet.Extensions.Rpc
 {
@@ -35,22 +36,27 @@ namespace MQTTnet.Extensions.Rpc
             _mqttClient.ApplicationMessageReceivedHandler = _applicationMessageReceivedHandler;
         }
 
-        public Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
-            return ExecuteAsync(timeout, methodName, Encoding.UTF8.GetBytes(payload), qualityOfServiceLevel, CancellationToken.None);
+            using (var timeoutToken = new CancellationTokenSource(timeout))
+            {
+                try
+                {
+                    return await ExecuteAsync(methodName, payload, qualityOfServiceLevel, timeoutToken.Token).ConfigureAwait(false);
+                }
+                catch (OperationCanceledException exception)
+                {
+                    if (timeoutToken.IsCancellationRequested)
+                    {
+                        throw new MqttCommunicationTimedOutException(exception);
+                    }
+
+                    throw;
+                }
+            }
         }
 
-        public Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken)
+        public async Task<byte[]> ExecuteAsync(string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken)
-        {
-            return ExecuteAsync(timeout, methodName, Encoding.UTF8.GetBytes(payload), qualityOfServiceLevel, cancellationToken);
-        }
-
-        public Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
-        {
-            return ExecuteAsync(timeout, methodName, payload, qualityOfServiceLevel, CancellationToken.None);
-        }
-
-        public async Task<byte[]> ExecuteAsync(TimeSpan timeout, string methodName, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, CancellationToken cancellationToken)
         {
             if (methodName == null) throw new ArgumentNullException(nameof(methodName));
 
@@ -88,62 +94,54 @@ namespace MQTTnet.Extensions.Rpc
 
             try
             {
-                var tcs = new TaskCompletionSource<byte[]>();
+#if NET452
-                if (!_waitingCalls.TryAdd(responseTopic, tcs))
+                var promise = new TaskCompletionSource<byte[]>();
+#else
+                var promise = new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously);
+#endif
+                
+                if (!_waitingCalls.TryAdd(responseTopic, promise))
                 {
                     throw new InvalidOperationException();
                 }
 
-                await _mqttClient.SubscribeAsync(responseTopic, qualityOfServiceLevel).ConfigureAwait(false);
+                var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
-                await _mqttClient.PublishAsync(requestMessage).ConfigureAwait(false);
+                    .WithTopicFilter(responseTopic, qualityOfServiceLevel)
+                    .Build();
 
-                using (var timeoutCts = new CancellationTokenSource(timeout))
+                await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false);
-                using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token))
+                await _mqttClient.PublishAsync(requestMessage, cancellationToken).ConfigureAwait(false);
+
+                using (cancellationToken.Register(() => { promise.TrySetCanceled(); }))
                 {
-                    linkedCts.Token.Register(() =>
+                    return await promise.Task.ConfigureAwait(false);
-                    {
-                        if (!tcs.Task.IsCompleted && !tcs.Task.IsFaulted && !tcs.Task.IsCanceled)
-                        {
-                            tcs.TrySetCanceled();
-                        }
-                    });
-
-                    try
-                    {
-                        var result = await tcs.Task.ConfigureAwait(false);
-                        timeoutCts.Cancel(false);
-                        return result;
-                    }
-                    catch (OperationCanceledException exception)
-                    {
-                        if (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
-                        {
-                            throw new MqttCommunicationTimedOutException(exception);
-                        }
-                        else
-                        {
-                            throw;
-                        }
-                    }
                 }
             }
             finally
             {
                 _waitingCalls.TryRemove(responseTopic, out _);
+
                 await _mqttClient.UnsubscribeAsync(responseTopic).ConfigureAwait(false);
             }
         }
 
         Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
         {
-            if (!_waitingCalls.TryRemove(eventArgs.ApplicationMessage.Topic, out var tcs))
+            if (!_waitingCalls.TryRemove(eventArgs.ApplicationMessage.Topic, out var promise))
             {
-                return Task.FromResult(0);
+                return PlatformAbstractionLayer.CompletedTask;
             }
 
-            tcs.TrySetResult(eventArgs.ApplicationMessage.Payload);
+#if NET452
+            Task.Run(() => promise.TrySetResult(eventArgs.ApplicationMessage.Payload));
+#else
+            promise.TrySetResult(eventArgs.ApplicationMessage.Payload);
+#endif
 
-            return Task.FromResult(0);
+            // Set this message to handled to that other code can avoid execution etc.
+            eventArgs.IsHandled = true;
+
+            return PlatformAbstractionLayer.CompletedTask;
         }
 
         public void Dispose()
@@ -158,4 +156,4 @@ namespace MQTTnet.Extensions.Rpc
             _waitingCalls.Clear();
         }
     }
-}
+}

Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Text;
+using System.Threading.Tasks;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Extensions.Rpc
+{
+    public static class MqttRpcClientExtensions
+    {
+        public static Task<byte[]> ExecuteAsync(this IMqttRpcClient client, TimeSpan timeout, string methodName, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
+
+            var buffer = Encoding.UTF8.GetBytes(payload ?? string.Empty);
+
+            return client.ExecuteAsync(timeout, methodName, buffer, qualityOfServiceLevel);
+        }
+    }
+}

Source/MQTTnet.Extensions.Rpc/RpcAwareApplicationMessageReceivedHandler.cs

@@ -18,14 +18,15 @@ namespace MQTTnet.Extensions.Rpc
 
         public IMqttApplicationMessageReceivedHandler OriginalHandler { get; }
 
-        public Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
+        public async Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
         {
+            // First try to check if there is a pending RPC call! 
+            await _handleReceivedApplicationMessageAsync(eventArgs).ConfigureAwait(false);
+            
             if (OriginalHandler != null)
             {
-                return OriginalHandler.HandleApplicationMessageReceivedAsync(eventArgs);
+                await OriginalHandler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false);
             }
-
-            return _handleReceivedApplicationMessageAsync(eventArgs);
         }
     }
 }

Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;net5.0</TargetFrameworks>
@@ -8,6 +8,9 @@
     <Company />
     <Authors />
     <PackageId />
+    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
+    <SignAssembly>false</SignAssembly>
+    <DelaySign>false</DelaySign>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>

Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs

@@ -55,7 +55,7 @@ namespace MQTTnet.Server.Logging
                 case MqttNetLogLevel.Error: return LogLevel.Error;
                 case MqttNetLogLevel.Warning: return LogLevel.Warning;
                 case MqttNetLogLevel.Info: return LogLevel.Information;
-                case MqttNetLogLevel.Verbose: return LogLevel.Debug;
+                case MqttNetLogLevel.Verbose: return LogLevel.Trace;
             }
 
             return LogLevel.Debug;

Source/MQTTnet/Adapter/MqttConnectingFailedException.cs

@@ -1,17 +1,19 @@
-using MQTTnet.Client.Connecting;
+using System;
+using MQTTnet.Client.Connecting;
 using MQTTnet.Exceptions;
 
 namespace MQTTnet.Adapter
 {
-    public class MqttConnectingFailedException : MqttCommunicationException
+    public sealed class MqttConnectingFailedException : MqttCommunicationException
     {
-        public MqttConnectingFailedException(MqttClientAuthenticateResult result)
+        public MqttConnectingFailedException(string message, Exception innerException, MqttClientAuthenticateResult authenticateResult)
-            : base($"Connecting with MQTT server failed ({result.ResultCode.ToString()}).")
+            : base(message, innerException)
         {
-            Result = result;
+            Result = authenticateResult;
         }
 
         public MqttClientAuthenticateResult Result { get; }
-        public MqttClientConnectResultCode ResultCode => Result.ResultCode;
+
+        public MqttClientConnectResultCode ResultCode => Result?.ResultCode ?? MqttClientConnectResultCode.UnspecifiedError;
     }
 }

Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectReason.cs

@@ -12,7 +12,7 @@
         ServerBusy = 137,
         ServerShuttingDown = 139,
         BadAuthenticationMethod = 140,
-        KeepaliveTimeout = 141,
+        KeepAliveTimeout = 141,
         SessionTakenOver = 142,
         TopicFilterInvalid = 143,
         TopicNameInvalid = 144,

Source/MQTTnet/Client/MqttClient.cs

@@ -14,7 +14,6 @@ using MQTTnet.PacketDispatcher;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using System;
-using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Implementations;
@@ -25,8 +24,6 @@ namespace MQTTnet.Client
     {
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
-        readonly Stopwatch _sendTracker = new Stopwatch();
-        readonly Stopwatch _receiveTracker = new Stopwatch();
         readonly object _disconnectLock = new object();
 
         readonly IMqttClientAdapterFactory _adapterFactory;
@@ -41,10 +38,12 @@ namespace MQTTnet.Client
 
         IMqttChannelAdapter _adapter;
         bool _cleanDisconnectInitiated;
-        long _isDisconnectPending;
+        volatile int _connectionStatus;
-        bool _isConnected;
+
         MqttClientDisconnectReason _disconnectReason;
 
+        DateTime _lastPacketSentTimestamp;
+
         public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
@@ -59,7 +58,7 @@ namespace MQTTnet.Client
 
         public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
 
-        public bool IsConnected => _isConnected && Interlocked.Read(ref _isDisconnectPending) == 0;
+        public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
 
         public IMqttClientOptions Options { get; private set; }
 
@@ -72,6 +71,11 @@ namespace MQTTnet.Client
 
             ThrowIfDisposed();
 
+            if (CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connecting, MqttClientConnectionStatus.Disconnected) != MqttClientConnectionStatus.Disconnected)
+            {
+                throw new InvalidOperationException("Not allowed to connect while connect/disconnect is pending.");
+            }
+
             MqttClientAuthenticateResult authenticateResult = null;
 
             try
@@ -79,12 +83,11 @@ namespace MQTTnet.Client
                 Options = options;
 
                 _packetIdentifierProvider.Reset();
-                _packetDispatcher.Cancel();
+                _packetDispatcher.CancelAll();
 
                 _backgroundCancellationTokenSource = new CancellationTokenSource();
                 var backgroundCancellationToken = _backgroundCancellationTokenSource.Token;
 
-                _isDisconnectPending = 0;
                 var adapter = _adapterFactory.CreateClientAdapter(options);
                 _adapter = adapter;
 
@@ -102,15 +105,14 @@ namespace MQTTnet.Client
                     authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false);
                 }
 
-                _sendTracker.Restart();
+                _lastPacketSentTimestamp = DateTime.UtcNow;
-                _receiveTracker.Restart();
 
                 if (Options.KeepAlivePeriod != TimeSpan.Zero)
                 {
                     _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(backgroundCancellationToken), backgroundCancellationToken);
                 }
 
-                _isConnected = true;
+                CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting);
 
                 _logger.Info("Connected.");
 
@@ -128,10 +130,7 @@ namespace MQTTnet.Client
 
                 _logger.Error(exception, "Error while connecting with server.");
 
-                if (!DisconnectIsPending())
+                await DisconnectInternalAsync(null, exception, authenticateResult).ConfigureAwait(false);
-                {
-                    await DisconnectInternalAsync(null, exception, authenticateResult).ConfigureAwait(false);
-                }
 
                 throw;
             }
@@ -143,7 +142,8 @@ namespace MQTTnet.Client
 
             ThrowIfDisposed();
 
-            if (DisconnectIsPending())
+            var clientWasConnected = IsConnected;
+            if (DisconnectIsPendingOrFinished())
             {
                 return;
             }
@@ -153,7 +153,7 @@ namespace MQTTnet.Client
                 _disconnectReason = MqttClientDisconnectReason.NormalDisconnection;
                 _cleanDisconnectInitiated = true;
 
-                if (_isConnected)
+                if (clientWasConnected)
                 {
                     var disconnectPacket = _adapter.PacketFormatterAdapter.DataConverter.CreateDisconnectPacket(options);
                     await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false);
@@ -161,13 +161,13 @@ namespace MQTTnet.Client
             }
             finally
             {
-                await DisconnectInternalAsync(null, null, null).ConfigureAwait(false);
+                await DisconnectCoreAsync(null, null, null, clientWasConnected).ConfigureAwait(false);
             }
         }
 
         public Task PingAsync(CancellationToken cancellationToken)
         {
-            return SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken);
+            return SendAndReceiveAsync<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken);
         }
 
         public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken)
@@ -232,21 +232,21 @@ namespace MQTTnet.Client
             switch (applicationMessage.QualityOfServiceLevel)
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
-                    {
+                {
-                        return PublishAtMostOnce(publishPacket, cancellationToken);
+                    return PublishAtMostOnce(publishPacket, cancellationToken);
-                    }
+                }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
-                    {
+                {
-                        return PublishAtLeastOnceAsync(publishPacket, cancellationToken);
+                    return PublishAtLeastOnceAsync(publishPacket, cancellationToken);
-                    }
+                }
                 case MqttQualityOfServiceLevel.ExactlyOnce:
-                    {
+                {
-                        return PublishExactlyOnceAsync(publishPacket, cancellationToken);
+                    return PublishExactlyOnceAsync(publishPacket, cancellationToken);
-                    }
+                }
                 default:
-                    {
+                {
-                        throw new NotSupportedException();
+                    throw new NotSupportedException();
-                    }
+                }
             }
         }
 
@@ -280,16 +280,25 @@ namespace MQTTnet.Client
 
         async Task<MqttClientAuthenticateResult> AuthenticateAsync(IMqttChannelAdapter channelAdapter, MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken)
         {
-            var connectPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnectPacket(
+            MqttClientAuthenticateResult result;
-                willApplicationMessage,
-                Options);
 
-            var connAckPacket = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket, cancellationToken).ConfigureAwait(false);
+            try
-            var result = channelAdapter.PacketFormatterAdapter.DataConverter.CreateClientConnectResult(connAckPacket);
+            {
+                var connectPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnectPacket(
+                    willApplicationMessage,
+                    Options);
+
+                var connAckPacket = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket, cancellationToken).ConfigureAwait(false);
+                result = channelAdapter.PacketFormatterAdapter.DataConverter.CreateClientConnectResult(connAckPacket);
+            }
+            catch (Exception exception)
+            {
+                throw new MqttConnectingFailedException($"Error while authenticating. {exception.Message}", exception, null);
+            }
 
             if (result.ResultCode != MqttClientConnectResultCode.Success)
             {
-                throw new MqttConnectingFailedException(result);
+                throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
             }
 
             _logger.Verbose("Authenticated MQTT connection with server established.");
@@ -299,7 +308,7 @@ namespace MQTTnet.Client
 
         void ThrowIfNotConnected()
         {
-            if (!IsConnected || Interlocked.Read(ref _isDisconnectPending) == 1)
+            if (!IsConnected)
             {
                 throw new MqttCommunicationException("The client is not connected.");
             }
@@ -310,12 +319,21 @@ namespace MQTTnet.Client
             if (IsConnected) throw new MqttProtocolViolationException(message);
         }
 
-        async Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientAuthenticateResult authenticateResult)
+        Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientAuthenticateResult authenticateResult)
         {
-            var clientWasConnected = _isConnected;
+            var clientWasConnected = IsConnected;
+            
+            if (!DisconnectIsPendingOrFinished())
+            {
+                return DisconnectCoreAsync(sender, exception, authenticateResult, clientWasConnected);
+            }
+            
+            return PlatformAbstractionLayer.CompletedTask;
+        }
 
+        async Task DisconnectCoreAsync(Task sender, Exception exception, MqttClientAuthenticateResult authenticateResult, bool clientWasConnected)
+        {
             TryInitiateDisconnect();
-            _isConnected = false;
 
             try
             {
@@ -339,8 +357,6 @@ namespace MQTTnet.Client
                 var keepAliveTask = WaitForTaskAsync(_keepAlivePacketsSenderTask, sender);
 
                 await Task.WhenAll(receiverTask, publishPacketReceiverTask, keepAliveTask).ConfigureAwait(false);
-
-                _publishPacketReceiverQueue?.Dispose();
             }
             catch (Exception e)
             {
@@ -350,6 +366,7 @@ namespace MQTTnet.Client
             {
                 Cleanup();
                 _cleanDisconnectInitiated = false;
+                CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnected, MqttClientConnectionStatus.Disconnecting);
 
                 _logger.Info("Disconnected.");
 
@@ -382,7 +399,7 @@ namespace MQTTnet.Client
         {
             cancellationToken.ThrowIfCancellationRequested();
 
-            _sendTracker.Restart();
+            _lastPacketSentTimestamp = DateTime.UtcNow;
 
             return _adapter.SendPacketAsync(packet, cancellationToken);
         }
@@ -391,23 +408,22 @@ namespace MQTTnet.Client
         {
             cancellationToken.ThrowIfCancellationRequested();
 
-            ushort identifier = 0;
+            ushort packetIdentifier = 0;
-            if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier > 0)
+            if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier)
             {
-                identifier = packetWithIdentifier.PacketIdentifier;
+                packetIdentifier = packetWithIdentifier.PacketIdentifier;
             }
 
-            using (var packetAwaiter = _packetDispatcher.AddAwaiter<TResponsePacket>(identifier))
+            using (var packetAwaiter = _packetDispatcher.AddAwaiter<TResponsePacket>(packetIdentifier))
             {
                 try
                 {
-                    _sendTracker.Restart();
+                    await SendAsync(requestPacket, cancellationToken).ConfigureAwait(false);
-                    await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
                 }
-                catch (Exception e)
+                catch (Exception exception)
                 {
-                    _logger.Warning(e, "Error when sending request packet ({0}).", requestPacket.GetType().Name);
+                    _logger.Warning(exception, "Error when sending request packet ({0}).", requestPacket.GetType().Name);
-                    packetAwaiter.Cancel();
+                    packetAwaiter.Fail(exception);
                 }
 
                 try
@@ -437,15 +453,15 @@ namespace MQTTnet.Client
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     // Values described here: [MQTT-3.1.2-24].
-                    var waitTime = keepAlivePeriod - _sendTracker.Elapsed;
+                    var timeWithoutPacketSent = DateTime.UtcNow - _lastPacketSentTimestamp;
 
-                    if (waitTime <= TimeSpan.Zero)
+                    if (timeWithoutPacketSent > keepAlivePeriod)
                     {
-                        await SendAndReceiveAsync<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false);
+                        await PingAsync(cancellationToken).ConfigureAwait(false);
                     }
 
                     // Wait a fixed time in all cases. Calculation of the remaining time is complicated
-                    // due to some edge cases and was buggy in the past. Now we wait half a second because the
+                    // due to some edge cases and was buggy in the past. Now we wait several ms because the
                     // min keep alive value is one second so that the server will wait 1.5 seconds for a PING
                     // packet.
                     await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken).ConfigureAwait(false);
@@ -471,10 +487,7 @@ namespace MQTTnet.Client
                     _logger.Error(exception, "Error exception while sending/receiving keep alive packets.");
                 }
 
-                if (!DisconnectIsPending())
+                await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
-                {
-                    await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
-                }
             }
             finally
             {
@@ -499,10 +512,7 @@ namespace MQTTnet.Client
 
                     if (packet == null)
                     {
-                        if (!DisconnectIsPending())
+                        await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
-                        {
-                            await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
-                        }
 
                         return;
                     }
@@ -529,12 +539,9 @@ namespace MQTTnet.Client
                     _logger.Error(exception, "Error while receiving packets.");
                 }
 
-                _packetDispatcher.Dispatch(exception);
+                _packetDispatcher.FailAll(exception);
 
-                if (!DisconnectIsPending())
+                await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-                {
-                    await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-                }
             }
             finally
             {
@@ -546,8 +553,6 @@ namespace MQTTnet.Client
         {
             try
             {
-                _receiveTracker.Restart();
-
                 if (packet is MqttPublishPacket publishPacket)
                 {
                     EnqueueReceivedPublishPacket(publishPacket);
@@ -560,10 +565,6 @@ namespace MQTTnet.Client
                 {
                     await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false);
                 }
-                else if (packet is MqttPingReqPacket)
-                {
-                    await SendAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false);
-                }
                 else if (packet is MqttDisconnectPacket disconnectPacket)
                 {
                     await ProcessReceivedDisconnectPacket(disconnectPacket).ConfigureAwait(false);
@@ -572,9 +573,20 @@ namespace MQTTnet.Client
                 {
                     await ProcessReceivedAuthPacket(authPacket).ConfigureAwait(false);
                 }
+                else if (packet is MqttPingRespPacket)
+                {
+                    _packetDispatcher.TryDispatch(packet);
+                }
+                else if (packet is MqttPingReqPacket)
+                {
+                    throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a Client to the Server only.");
+                }
                 else
                 {
-                    _packetDispatcher.Dispatch(packet);
+                    if (!_packetDispatcher.TryDispatch(packet))
+                    {
+                        throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
+                    }
                 }
             }
             catch (Exception exception)
@@ -596,12 +608,9 @@ namespace MQTTnet.Client
                     _logger.Error(exception, "Error while receiving packets.");
                 }
 
-                _packetDispatcher.Dispatch(exception);
+                _packetDispatcher.FailAll(exception);
 
-                if (!DisconnectIsPending())
+                await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-                {
-                    await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
-                }
             }
         }
 
@@ -630,40 +639,11 @@ namespace MQTTnet.Client
                     }
 
                     var publishPacket = publishPacketDequeueResult.Item;
+                    var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
 
-                    if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
+                    if (eventArgs.AutoAcknowledge)
                     {
-                        await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
+                        await eventArgs.AcknowledgeAsync(cancellationToken).ConfigureAwait(false);
-                    }
-                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
-                    {
-                        var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
-
-                        if (!eventArgs.ProcessingFailed)
-                        {
-                            await SendAsync(new MqttPubAckPacket
-                            {
-                                PacketIdentifier = publishPacket.PacketIdentifier,
-                                ReasonCode = (MqttPubAckReasonCode)eventArgs.ReasonCode
-                            }, cancellationToken).ConfigureAwait(false);
-                        }
-                    }
-                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
-                    {
-                        var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
-
-                        if (!eventArgs.ProcessingFailed)
-                        {
-                            await SendAsync(new MqttPubRecPacket
-                            {
-                                PacketIdentifier = publishPacket.PacketIdentifier,
-                                ReasonCode = (MqttPubRecReasonCode)eventArgs.ReasonCode
-                            }, cancellationToken).ConfigureAwait(false);
-                        }
-                    }
-                    else
-                    {
-                        throw new MqttProtocolViolationException("Received a not supported QoS level.");
                     }
                 }
                 catch (OperationCanceledException)
@@ -676,17 +656,44 @@ namespace MQTTnet.Client
             }
         }
 
+        internal Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs, CancellationToken cancellationToken)
+        {
+            if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
+            {
+                // no response required
+            }
+            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+            {
+                if (!eventArgs.ProcessingFailed)
+                {
+                    var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(eventArgs.PublishPacket, eventArgs.ReasonCode);
+                    return SendAsync(pubAckPacket, cancellationToken);
+                }
+            }
+            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
+            {
+                if (!eventArgs.ProcessingFailed)
+                {
+                    var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(eventArgs.PublishPacket, eventArgs.ReasonCode);
+                    return SendAsync(pubRecPacket, cancellationToken);
+                }
+            }
+            else
+            {
+                throw new MqttProtocolViolationException("Received a not supported QoS level.");
+            }
+
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
         Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken)
         {
             if (!_packetDispatcher.TryDispatch(pubRecPacket))
             {
                 // The packet is unknown. Probably due to a restart of the client.
                 // So wen send this to the server to trigger a full resend of the message.
-                return SendAsync(new MqttPubRelPacket
+                var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
-                {
+                return SendAsync(pubRelPacket, cancellationToken);
-                    PacketIdentifier = pubRecPacket.PacketIdentifier,
-                    ReasonCode = MqttPubRelReasonCode.PacketIdentifierNotFound
-                }, cancellationToken);
             }
 
             return PlatformAbstractionLayer.CompletedTask;
@@ -694,26 +701,18 @@ namespace MQTTnet.Client
 
         Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
         {
-            return SendAsync(new MqttPubCompPacket
+            var pubCompPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            {
+            return SendAsync(pubCompPacket, cancellationToken);
-                PacketIdentifier = pubRelPacket.PacketIdentifier,
-                ReasonCode = MqttPubCompReasonCode.Success
-            }, cancellationToken);
         }
 
         Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket)
         {
-            _disconnectReason = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection);
+            _disconnectReason = (MqttClientDisconnectReason) (disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.NormalDisconnection);
 
             // Also dispatch disconnect to waiting threads to generate a proper exception.
-            _packetDispatcher.Dispatch(disconnectPacket);
+            _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
 
-            if (!DisconnectIsPending())
+            return DisconnectInternalAsync(_packetReceiverTask, null, null);
-            {
-                return DisconnectInternalAsync(_packetReceiverTask, null, null);
-            }
-
-            return PlatformAbstractionLayer.CompletedTask;
         }
 
         Task ProcessReceivedAuthPacket(MqttAuthPacket authPacket)
@@ -731,14 +730,16 @@ namespace MQTTnet.Client
         {
             // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
             await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false);
-            return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(null);
+            return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(null);
         }
 
         async Task<MqttClientPublishResult> PublishAtLeastOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
             publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-            var response = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
+            
-            return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(response);
+            var pubAckPacket = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
+            
+            return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubAckPacket);
         }
 
         async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
@@ -747,21 +748,17 @@ namespace MQTTnet.Client
 
             var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
 
-            var pubRelPacket = new MqttPubRelPacket
+            var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            {
-                PacketIdentifier = publishPacket.PacketIdentifier,
-                ReasonCode = MqttPubRelReasonCode.Success
-            };
 
             var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
 
-            return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(pubRecPacket, pubCompPacket);
+            return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubRecPacket, pubCompPacket);
         }
 
         async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
         {
             var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
-            var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage);
+            var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage, publishPacket, AcknowledgeReceivedPublishPacket);
 
             var handler = ApplicationMessageReceivedHandler;
             if (handler != null)
@@ -803,11 +800,36 @@ namespace MQTTnet.Client
             }
         }
 
-        bool DisconnectIsPending()
+        bool DisconnectIsPendingOrFinished()
         {
-            // This will read the _isDisconnectPending and set it to "1" afterwards regardless of the value.
+            var connectionStatus = (MqttClientConnectionStatus)_connectionStatus;
-            // So the first caller will get a "false" and all subsequent ones will get "true".
+            
-            return Interlocked.CompareExchange(ref _isDisconnectPending, 1, 0) != 0;
+            do
+            {
+                switch (connectionStatus)
+                {
+                    case MqttClientConnectionStatus.Disconnected:
+                    case MqttClientConnectionStatus.Disconnecting:
+                        return true;
+                    case MqttClientConnectionStatus.Connected:
+                    case MqttClientConnectionStatus.Connecting:
+                        // This will compare the _connectionStatus to old value and set it to "MqttClientConnectionStatus.Disconnecting" afterwards.
+                        // So the first caller will get a "false" and all subsequent ones will get "true".
+                        var curStatus = CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnecting, connectionStatus);
+                        if (curStatus == connectionStatus)
+                        {
+                            return false;
+                        }
+                        
+                        connectionStatus = curStatus;
+                        break;
+                }
+            } while (true);
+        }
+
+        MqttClientConnectionStatus CompareExchangeConnectionStatus(MqttClientConnectionStatus value, MqttClientConnectionStatus comparand)
+        {
+            return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand);
         }
     }
-}
+}

Source/MQTTnet/Client/MqttClientConnectionStatus.cs

@@ -0,0 +1,10 @@
+namespace MQTTnet.Client
+{
+    public enum MqttClientConnectionStatus
+    {
+        Disconnected,
+        Disconnecting,
+        Connected,
+        Connecting
+    }
+}

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -252,7 +252,11 @@ namespace MQTTnet.Client.Options
         {
             if (optionsBuilder == null) throw new ArgumentNullException(nameof(optionsBuilder));
 
-            _tlsParameters = new MqttClientOptionsBuilderTlsParameters();
+            _tlsParameters = new MqttClientOptionsBuilderTlsParameters
+            {
+                UseTls = true
+            };
+            
             optionsBuilder(_tlsParameters);
             return this;
         }
@@ -287,7 +291,7 @@ namespace MQTTnet.Client.Options
 #pragma warning disable CS0618 // Type or member is obsolete
                         CertificateValidationCallback = _tlsParameters.CertificateValidationCallback,
 #pragma warning restore CS0618 // Type or member is obsolete
-#if NETCOREAPP3_1
+#if NETCOREAPP3_1 || NET5_0
                         ApplicationProtocols = _tlsParameters.ApplicationProtocols,
 #endif
                         CertificateValidationHandler = _tlsParameters.CertificateValidationHandler,

Source/MQTTnet/Client/Options/MqttClientOptionsBuilderTlsParameters.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Net.Security;
 using System.Security.Authentication;
@@ -31,9 +31,9 @@ namespace MQTTnet.Client.Options
         public IEnumerable<X509Certificate> Certificates { get; set; }
 #endif
 
-#if NETCOREAPP3_1
+#if NETCOREAPP3_1 || NET5_0
 	    public List<SslApplicationProtocol> ApplicationProtocols { get;set; }
-#endif 
+#endif
 
 	    public bool AllowUntrustedCertificates { get; set; }
 

Source/MQTTnet/Client/Options/MqttClientTlsOptions.cs

@@ -22,7 +22,7 @@ namespace MQTTnet.Client.Options
         public List<X509Certificate> Certificates { get; set; }
 #endif
 
-#if NETCOREAPP3_1
+#if NETCOREAPP3_1 || NET5_0
         public List<SslApplicationProtocol> ApplicationProtocols { get; set; }
 #endif
 

Source/MQTTnet/Formatter/IMqttDataConverter.cs

@@ -14,24 +14,32 @@ namespace MQTTnet.Formatter
 {
     public interface IMqttDataConverter
     {
-        MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage);
-
-        MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket);
-
-        MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket);
-
         MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket);
 
         MqttClientAuthenticateResult CreateClientConnectResult(MqttConnAckPacket connAckPacket);
 
+        MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket);
+
+        MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket);
+        
+        MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket);
+        
+        MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket);
+
         MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, IMqttClientOptions options);
 
         MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext);
+        
+        MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage);
 
-        MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket);
+        MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
-
+        
-        MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket);
+        MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
 
+        MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
+        
+        MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
+        
         MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options);
 
         MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, Server.MqttClientSubscribeResult subscribeResult);
@@ -41,9 +49,5 @@ namespace MQTTnet.Formatter
         MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes);
 
         MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options);
-
-        MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket);
-
-        MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket);
     }
 }

Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs

@@ -31,21 +31,43 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket)
+        public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
         {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+            
             return new MqttPubAckPacket
             {
-                PacketIdentifier = publishPacket.PacketIdentifier,
+                PacketIdentifier = publishPacket.PacketIdentifier
-                ReasonCode = MqttPubAckReasonCode.Success
             };
         }
 
-        public MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket)
+        public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
         {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+            
             return new MqttPubRecPacket
             {
-                PacketIdentifier = publishPacket.PacketIdentifier,
+                PacketIdentifier = publishPacket.PacketIdentifier
-                ReasonCode = MqttPubRecReasonCode.Success
+            };
+        }
+
+        public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+        {
+            if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket));
+            
+            return new MqttPubCompPacket
+            {
+                PacketIdentifier = pubRelPacket.PacketIdentifier
+            };
+        }
+
+        public MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+        {
+            if (pubRecPacket == null) throw new ArgumentNullException(nameof(pubRecPacket));
+            
+            return new MqttPubRelPacket
+            {
+                PacketIdentifier = pubRecPacket.PacketIdentifier
             };
         }
 
@@ -58,7 +80,8 @@ namespace MQTTnet.Formatter.V3
                 Topic = publishPacket.Topic,
                 Payload = publishPacket.Payload,
                 QualityOfServiceLevel = publishPacket.QualityOfServiceLevel,
-                Retain = publishPacket.Retain
+                Retain = publishPacket.Retain,
+                Dup = publishPacket.Dup
             };
         }
 
@@ -184,6 +207,9 @@ namespace MQTTnet.Formatter.V3
 
         public MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, Server.MqttClientSubscribeResult subscribeResult)
         {
+            if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
+            if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult));
+
             var subackPacket = new MqttSubAckPacket
             {
                 PacketIdentifier = subscribePacket.PacketIdentifier
@@ -206,6 +232,9 @@ namespace MQTTnet.Formatter.V3
 
         public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes)
         {
+            if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
+            if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes));
+
             return new MqttUnsubAckPacket
             {
                 PacketIdentifier = unsubscribePacket.PacketIdentifier,
@@ -215,15 +244,10 @@ namespace MQTTnet.Formatter.V3
 
         public MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options)
         {
-            if (options.ReasonCode != MqttClientDisconnectReason.NormalDisconnection || options.ReasonString != null)
-            {
-                throw new MqttProtocolViolationException("Reason codes and reason string for disconnect are only supported for MQTTv5.");
-            }
-
             return new MqttDisconnectPacket();
         }
 
-        public MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket)
+        public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket)
         {
             return new MqttClientPublishResult
             {
@@ -232,7 +256,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        public MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
+        public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
         {
             if (pubRecPacket == null || pubCompPacket == null)
             {

Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs

@@ -15,7 +15,7 @@ using MqttClientSubscribeResult = MQTTnet.Client.Subscribing.MqttClientSubscribe
 
 namespace MQTTnet.Formatter.V5
 {
-    public class MqttV500DataConverter : IMqttDataConverter
+    public sealed class MqttV500DataConverter : IMqttDataConverter
     {
         public MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage)
         {
@@ -27,7 +27,7 @@ namespace MQTTnet.Formatter.V5
                 Payload = applicationMessage.Payload,
                 QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel,
                 Retain = applicationMessage.Retain,
-                Dup = false,
+                Dup = applicationMessage.Dup,
                 Properties = new MqttPublishPacketProperties
                 {
                     ContentType = applicationMessage.ContentType,
@@ -49,32 +49,61 @@ namespace MQTTnet.Formatter.V5
             return packet;
         }
 
-        public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket)
+        public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
         {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
             return new MqttPubAckPacket
             {
                 PacketIdentifier = publishPacket.PacketIdentifier,
-                ReasonCode = MqttPubAckReasonCode.Success
+                ReasonCode = (MqttPubAckReasonCode)(int)reasonCode
             };
         }
 
-        public MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket)
+        public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
         {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
             return new MqttPubRecPacket
             {
                 PacketIdentifier = publishPacket.PacketIdentifier,
-                ReasonCode = MqttPubRecReasonCode.Success
+                ReasonCode = (MqttPubRecReasonCode)(int)reasonCode
+            };
+        }
+
+        public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+        {
+            if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket));
+
+            return new MqttPubCompPacket
+            {
+                PacketIdentifier = pubRelPacket.PacketIdentifier,
+                ReasonCode = (MqttPubCompReasonCode)(int)reasonCode
+            };
+        }
+
+        public MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+        {
+            if (pubRecPacket == null) throw new ArgumentNullException(nameof(pubRecPacket));
+
+            return new MqttPubRelPacket
+            {
+                PacketIdentifier = pubRecPacket.PacketIdentifier,
+                ReasonCode = (MqttPubRelReasonCode)(int)reasonCode
             };
         }
 
         public MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket)
         {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
             return new MqttApplicationMessage
             {
                 Topic = publishPacket.Topic,
                 Payload = publishPacket.Payload,
                 QualityOfServiceLevel = publishPacket.QualityOfServiceLevel,
                 Retain = publishPacket.Retain,
+                Dup = publishPacket.Dup,
                 ResponseTopic = publishPacket.Properties?.ResponseTopic,
                 ContentType = publishPacket.Properties?.ContentType,
                 CorrelationData = publishPacket.Properties?.CorrelationData,
@@ -93,7 +122,7 @@ namespace MQTTnet.Formatter.V5
             return new MqttClientAuthenticateResult
             {
                 IsSessionPresent = connAckPacket.IsSessionPresent,
-                ResultCode = (MqttClientConnectResultCode)connAckPacket.ReasonCode.Value,
+                ResultCode = (MqttClientConnectResultCode)(int)(connAckPacket.ReasonCode ?? 0),
                 WildcardSubscriptionAvailable = connAckPacket.Properties?.WildcardSubscriptionAvailable,
                 RetainAvailable = connAckPacket.Properties?.RetainAvailable,
                 AssignedClientIdentifier = connAckPacket.Properties?.AssignedClientIdentifier,
@@ -144,6 +173,8 @@ namespace MQTTnet.Formatter.V5
 
         public MqttConnAckPacket CreateConnAckPacket(MqttConnectionValidatorContext connectionValidatorContext)
         {
+            if (connectionValidatorContext == null) throw new ArgumentNullException(nameof(connectionValidatorContext));
+
             return new MqttConnAckPacket
             {
                 ReasonCode = connectionValidatorContext.ReasonCode,
@@ -213,6 +244,9 @@ namespace MQTTnet.Formatter.V5
 
         public MqttSubAckPacket CreateSubAckPacket(MqttSubscribePacket subscribePacket, Server.MqttClientSubscribeResult subscribeResult)
         {
+            if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
+            if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult));
+
             var subackPacket = new MqttSubAckPacket
             {
                 PacketIdentifier = subscribePacket.PacketIdentifier
@@ -240,6 +274,9 @@ namespace MQTTnet.Formatter.V5
 
         public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes)
         {
+            if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
+            if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes));
+
             return new MqttUnsubAckPacket
             {
                 PacketIdentifier = unsubscribePacket.PacketIdentifier,
@@ -263,7 +300,7 @@ namespace MQTTnet.Formatter.V5
             return packet;
         }
 
-        public MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket)
+        public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket)
         {
             var result = new MqttClientPublishResult
             {
@@ -276,7 +313,7 @@ namespace MQTTnet.Formatter.V5
             {
                 // QoS 0 has no response. So we treat it as a success always.
                 // Both enums have the same values. So it can be easily converted.
-                result.ReasonCode = (MqttClientPublishReasonCode)pubAckPacket.ReasonCode;
+                result.ReasonCode = (MqttClientPublishReasonCode)(int)(pubAckPacket.ReasonCode ?? 0);
 
                 result.PacketIdentifier = pubAckPacket.PacketIdentifier;
             }
@@ -284,7 +321,7 @@ namespace MQTTnet.Formatter.V5
             return result;
         }
 
-        public MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
+        public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
         {
             if (pubRecPacket == null || pubCompPacket == null)
             {
@@ -317,10 +354,10 @@ namespace MQTTnet.Formatter.V5
             if (pubRecPacket.ReasonCode.HasValue)
             {
                 // Both enums share the same values.
-                result.ReasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode.Value;
+                result.ReasonCode = (MqttClientPublishReasonCode)(pubRecPacket.ReasonCode ?? 0);
             }
 
             return result;
         }
     }
-}
+}

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -75,23 +75,11 @@ namespace MQTTnet.Implementations
                     _socket.Control.IgnorableServerCertificateErrors.Add(ignorableChainValidationResult);
                 }
 
-#if NETCOREAPP3_1 || NET5_0
-                var socketProtectionLevel = SocketProtectionLevel.Tls13;
-                if (_options.TlsOptions.SslProtocol == SslProtocols.Tls12)
-                {
-                    socketProtectionLevel = SocketProtectionLevel.Tls12;
-                }
-                else if (_options.TlsOptions.SslProtocol == SslProtocols.Tls11)
-                {
-                    socketProtectionLevel = SocketProtectionLevel.Tls11;
-                }
-#else
                 var socketProtectionLevel = SocketProtectionLevel.Tls12;
                 if (_options.TlsOptions.SslProtocol == SslProtocols.Tls11)
                 {
                     socketProtectionLevel = SocketProtectionLevel.Tls11;
                 }
-#endif
                 else if (_options.TlsOptions.SslProtocol == SslProtocols.Tls)
                 {
                     socketProtectionLevel = SocketProtectionLevel.Tls10;

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -83,7 +83,7 @@ namespace MQTTnet.Implementations
                     var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                     try
                     {
-#if NETCOREAPP3_1
+#if NETCOREAPP3_1 || NET5_0
                         var sslOptions = new SslClientAuthenticationOptions
                         {
                             ApplicationProtocols = _tcpOptions.TlsOptions.ApplicationProtocols,
@@ -100,7 +100,7 @@ namespace MQTTnet.Implementations
                     }
                     catch
                     {
-#if NETSTANDARD2_1 || NETCOREAPP3_1
+#if NETSTANDARD2_1 || NETCOREAPP3_1 || NET5_0
                         await sslStream.DisposeAsync().ConfigureAwait(false);
 #else
                         sslStream.Dispose();

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -98,7 +98,7 @@ namespace MQTTnet.Implementations
         {
             _socket?.Dispose();
 
-#if NETSTANDARD1_3 || NETSTANDARD2_0 || NET461 || NET472
+#if !NET452
             _tlsCertificate?.Dispose();
 #endif
         }

Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs

@@ -17,7 +17,7 @@ namespace MQTTnet.Implementations
 
         public static void Sleep(TimeSpan timeout)
         {
-#if NET452 || NETSTANDARD2_0 || NETSTANDARD2_1 || NETCOREAPP3_1
+#if !NETSTANDARD1_3 && !WINDOWS_UWP
             System.Threading.Thread.Sleep(timeout);
 #else
             Task.Delay(timeout).Wait();

Source/MQTTnet/MQTTnet.csproj

@@ -35,14 +35,6 @@
   <PropertyGroup Condition="'$(Configuration)'=='Debug'">
     <DebugType>Full</DebugType>
   </PropertyGroup>
-  
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x86'" />
-
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'" />
-
-  <PropertyGroup Condition="'$(Configuration)|$(TargetFramework)|$(Platform)'=='Release|netstandard1.3|AnyCPU'">
-    <DefineConstants>RELEASE;NETSTANDARD1_3</DefineConstants>
-  </PropertyGroup>
 
   <ItemGroup>
     <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
@@ -54,20 +46,8 @@
     <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.2" />
   </ItemGroup>
 
-  <ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
-    <PackageReference Include="System.Net.Security" Version="4.3.2" />
-    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
-    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.2" />
-  </ItemGroup>
-
   <ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
     <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.10" />
   </ItemGroup>
 
-  <ItemGroup Condition="'$(TargetFramework)'=='net452'">
-  </ItemGroup>
-
-  <ItemGroup Condition="'$(TargetFramework)'=='net461'">
-  </ItemGroup>
-  
 </Project>

Source/MQTTnet/MqttApplicationMessage.cs

@@ -35,6 +35,14 @@ namespace MQTTnet
         /// The broker stores the last retained message and the corresponding QoS for that topic.
         /// </summary>
         public bool Retain { get; set; }
+        
+        /// <summary>
+        /// If the DUP flag is set to 0, it indicates that this is the first occasion that the Client or Server has attempted to send this MQTT PUBLISH Packet.
+        /// If the DUP flag is set to 1, it indicates that this might be re-delivery of an earlier attempt to send the Packet.
+        /// The DUP flag MUST be set to 1 by the Client or Server when it attempts to re-deliver a PUBLISH Packet [MQTT-3.3.1.-1].
+        /// The DUP flag MUST be set to 0 for all QoS 0 messages [MQTT-3.3.1-2].
+        /// </summary>
+        public bool Dup { get; set; }
 
         /// <summary>
         /// Gets or sets the user properties.

Source/MQTTnet/MqttApplicationMessageBuilder.cs

@@ -19,54 +19,56 @@ namespace MQTTnet
         /// - At least once (1): Message gets delivered at least once (one time or more often).
         /// - Exactly once  (2): Message gets delivered exactly once (It's ensured that the message only comes once).
         /// </summary>
-        private MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
+        MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
 
         /// <summary>
         /// The MQTT topic.
         /// In MQTT, the word topic refers to an UTF-8 string that the broker uses to filter messages for each connected client.
         /// The topic consists of one or more topic levels. Each topic level is separated by a forward slash (topic level separator). 
         /// </summary>
-        private string _topic;
+        string _topic;
 
         /// <summary>
         /// The payload.
         /// The payload is the data bytes sent via the MQTT protocol.
         /// </summary>
-        private byte[] _payload;
+        byte[] _payload;
 
         /// <summary>
         /// A value indicating whether the message should be retained or not.
         /// A retained message is a normal MQTT message with the retained flag set to true.
         /// The broker stores the last retained message and the corresponding QoS for that topic.
         /// </summary>
-        private bool _retain;
+        bool _retain;
+
+        bool _dup;
 
         /// <summary>
         /// The content type.
         /// The content type must be a UTF-8 encoded string. The content type value identifies the kind of UTF-8 encoded payload.
         /// </summary>
-        private string _contentType;
+        string _contentType;
 
         /// <summary>
         /// The response topic.
         /// In MQTT 5 the ability to publish a response topic was added in the publish message which allows you to implement the request/response pattern between clients that is common in web applications.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        private string _responseTopic;
+        string _responseTopic;
 
         /// <summary>
         /// The correlation data.
         /// In order for the sender to know what sent message the response refers to it can also send correlation data with the published message.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        private byte[] _correlationData;
+        byte[] _correlationData;
 
         /// <summary>
         /// The topic alias.
         /// Topic aliases were introduced are a mechanism for reducing the size of published packets by reducing the size of the topic field.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        private ushort? _topicAlias;
+        ushort? _topicAlias;
 
         /// <summary>
         /// The subscription identifiers.
@@ -75,7 +77,7 @@ namespace MQTTnet
         /// The broker will return the subscription identifier associated with this PUBLISH packet and the PUBLISH packet to the client when need to forward PUBLISH packets matching this subscription to this client.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        private List<uint> _subscriptionIdentifiers;
+        List<uint> _subscriptionIdentifiers;
 
         /// <summary>
         /// The message expiry interval.
@@ -85,7 +87,7 @@ namespace MQTTnet
         /// When the retained=true option is set on the PUBLISH message, this interval also defines how long a message is retained on a topic.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        private uint? _messageExpiryInterval;
+        uint? _messageExpiryInterval;
 
         /// <summary>
         /// The payload format indicator.
@@ -95,7 +97,7 @@ namespace MQTTnet
         /// If no payload format indicator is provided, the default value is 0.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        private MqttPayloadFormatIndicator? _payloadFormatIndicator;
+        MqttPayloadFormatIndicator? _payloadFormatIndicator;
 
         /// <summary>
         /// The user properties.
@@ -104,7 +106,7 @@ namespace MQTTnet
         /// The feature is very similar to the HTTP header concept.
         /// </summary>
         /// Hint: MQTT 5 feature only.
-        private List<MqttUserProperty> _userProperties;
+        List<MqttUserProperty> _userProperties;
 
         public MqttApplicationMessageBuilder WithTopic(string topic)
         {
@@ -126,12 +128,7 @@ namespace MQTTnet
                 return this;
             }
 
-            _payload = payload as byte[];
+            _payload = payload as byte[] ?? payload.ToArray();
-
-            if (_payload == null)
-            {
-                _payload = payload.ToArray();
-            }
 
             return this;
         }
@@ -201,6 +198,12 @@ namespace MQTTnet
             _retain = value;
             return this;
         }
+        
+        public MqttApplicationMessageBuilder WithDupFlag(bool value = true)
+        {
+            _dup = value;
+            return this;
+        }
 
         public MqttApplicationMessageBuilder WithAtLeastOnceQoS()
         {
@@ -338,12 +341,18 @@ namespace MQTTnet
                 throw new MqttProtocolViolationException("A Topic Alias of 0 is not permitted. A sender MUST NOT send a PUBLISH packet containing a Topic Alias which has the value 0 [MQTT-3.3.2-8].");
             }
 
+            if (_qualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce && _dup)
+            {
+                throw new MqttProtocolViolationException("The DUP flag MUST be set to 0 for all QoS 0 messages [MQTT-3.3.1-2].");
+            }
+
             var applicationMessage = new MqttApplicationMessage
             {
                 Topic = _topic,
                 Payload = _payload,
                 QualityOfServiceLevel = _qualityOfServiceLevel,
                 Retain = _retain,
+                Dup = _dup,
                 ContentType = _contentType,
                 ResponseTopic = _responseTopic,
                 CorrelationData = _correlationData,

Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs

@@ -1,15 +1,30 @@
-using System;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Packets;
 
 namespace MQTTnet
 {
     public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
     {
-        public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage)
+        readonly Func<MqttApplicationMessageReceivedEventArgs, CancellationToken, Task> _acknowledgeHandler;
+        
+        int _isAcknowledged;
+        
+        public MqttApplicationMessageReceivedEventArgs(
+            string clientId, 
+            MqttApplicationMessage applicationMessage,
+            MqttPublishPacket publishPacket,
+            Func<MqttApplicationMessageReceivedEventArgs, CancellationToken, Task> acknowledgeHandler)
         {
             ClientId = clientId;
             ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            PublishPacket = publishPacket;
+            _acknowledgeHandler = acknowledgeHandler;
         }
 
+        internal MqttPublishPacket PublishPacket { get; }
+        
         /// <summary>
         /// Gets the client identifier.
         /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
@@ -21,5 +36,33 @@ namespace MQTTnet
         public bool ProcessingFailed { get; set; }
 
         public MqttApplicationMessageReceivedReasonCode ReasonCode { get; set; } = MqttApplicationMessageReceivedReasonCode.Success;
+        
+        /// <summary>
+        /// Gets or sets whether this message was handled.
+        /// This value can be used in user code for custom control flow.
+        /// </summary>
+        public bool IsHandled { get; set; }
+
+        /// <summary>
+        /// Gets ir sets whether the library should send MQTT ACK packets automatically if required.
+        /// </summary>
+        public bool AutoAcknowledge { get; set; } = true;
+        
+        public object Tag { get; set; }
+        
+        public Task AcknowledgeAsync(CancellationToken cancellationToken) 
+        {
+            if (_acknowledgeHandler == null)
+            {
+                throw new NotSupportedException("Deferred acknowledgement of application message is not yet supported in MQTTnet server.");
+            }
+            
+            if (Interlocked.CompareExchange(ref _isAcknowledged, 1, 0) == 0)
+            {
+                return _acknowledgeHandler(this, cancellationToken);
+            }
+            
+            throw new InvalidOperationException("The application message is already acknowledged.");
+        }
     }
 }

Source/MQTTnet/MqttApplicationMessageReceivedReasonCode.cs

@@ -9,6 +9,7 @@
         NotAuthorized = 135,
         TopicNameInvalid = 144,
         PacketIdentifierInUse = 145,
+        PacketIdentifierNotFound = 146,
         QuotaExceeded = 151,
         PayloadFormatInvalid = 153
     }

Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs

@@ -5,6 +5,8 @@ namespace MQTTnet.PacketDispatcher
 {
     public interface IMqttPacketAwaiter : IDisposable
     {
+        MqttPacketAwaiterPacketFilter PacketFilter { get; }
+        
         void Complete(MqttBasePacket packet);
 
         void Fail(Exception exception);

Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs

@@ -9,12 +9,16 @@ namespace MQTTnet.PacketDispatcher
     public sealed class MqttPacketAwaiter<TPacket> : IMqttPacketAwaiter where TPacket : MqttBasePacket
     {
         readonly TaskCompletionSource<MqttBasePacket> _taskCompletionSource;
-        readonly ushort? _packetIdentifier;
         readonly MqttPacketDispatcher _owningPacketDispatcher;
 
-        public MqttPacketAwaiter(ushort? packetIdentifier, MqttPacketDispatcher owningPacketDispatcher)
+        public MqttPacketAwaiter(ushort packetIdentifier, MqttPacketDispatcher owningPacketDispatcher)
         {
-            _packetIdentifier = packetIdentifier;
+            PacketFilter = new MqttPacketAwaiterPacketFilter
+            {
+                Type = typeof(TPacket),
+                Identifier = packetIdentifier
+            };
+            
             _owningPacketDispatcher = owningPacketDispatcher ?? throw new ArgumentNullException(nameof(owningPacketDispatcher));
 #if NET452
             _taskCompletionSource = new TaskCompletionSource<MqttBasePacket>();
@@ -22,7 +26,9 @@ namespace MQTTnet.PacketDispatcher
             _taskCompletionSource = new TaskCompletionSource<MqttBasePacket>(TaskCreationOptions.RunContinuationsAsynchronously);
 #endif
         }
-
+        
+        public MqttPacketAwaiterPacketFilter PacketFilter { get; }
+        
         public async Task<TPacket> WaitOneAsync(TimeSpan timeout)
         {
             using (var timeoutToken = new CancellationTokenSource(timeout))
@@ -82,7 +88,7 @@ namespace MQTTnet.PacketDispatcher
 
         public void Dispose()
         {
-            _owningPacketDispatcher.RemoveAwaiter<TPacket>(_packetIdentifier);
+            _owningPacketDispatcher.RemoveAwaiter(this);
         }
     }
 }

Source/MQTTnet/PacketDispatcher/MqttPacketAwaiterPacketFilter.cs

@@ -0,0 +1,11 @@
+using System;
+
+namespace MQTTnet.PacketDispatcher
+{
+    public sealed class MqttPacketAwaiterPacketFilter
+    {
+        public Type Type { get; set; }
+        
+        public ushort Identifier { get; set; }
+    }
+}

Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs

@@ -1,103 +1,101 @@
-using MQTTnet.Exceptions;
 using MQTTnet.Packets;
 using System;
-using System.Collections.Concurrent;
+using System.Collections.Generic;
 
 namespace MQTTnet.PacketDispatcher
 {
     public sealed class MqttPacketDispatcher
     {
-        readonly ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter> _awaiters = new ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter>();
+        readonly List<IMqttPacketAwaiter> _awaiters = new List<IMqttPacketAwaiter>();
 
-        public void Dispatch(Exception exception)
+        public void FailAll(Exception exception)
         {
-            foreach (var awaiter in _awaiters)
+            if (exception == null) throw new ArgumentNullException(nameof(exception));
-            {
-                awaiter.Value.Fail(exception);
-            }
 
-            _awaiters.Clear();
+            lock (_awaiters)
+            {
+                foreach (var awaiter in _awaiters)
+                {
+                    awaiter.Fail(exception);
+                }
+
+                _awaiters.Clear();
+            }
         }
 
+        public void CancelAll()
+        {
+            lock (_awaiters)
+            {
+                foreach (var entry in _awaiters)
+                {
+                    entry.Cancel();
+                }
+
+                _awaiters.Clear();
+            }
+        }
+        
         public bool TryDispatch(MqttBasePacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
-
+            
-            if (packet is MqttDisconnectPacket disconnectPacket)
-            {
-                foreach (var packetAwaiter in _awaiters)
-                {
-                    packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
-                }
-
-                return true;
-            }
-
             ushort identifier = 0;
-            if (packet is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier > 0)
+            if (packet is IMqttPacketWithIdentifier packetWithIdentifier)
             {
                 identifier = packetWithIdentifier.PacketIdentifier;
             }
 
-            var type = packet.GetType();
+            var packetType = packet.GetType();
-            var key = new Tuple<ushort, Type>(identifier, type);
+            var matchingAwaiters = new List<IMqttPacketAwaiter>();
-
+            
-            if (_awaiters.TryRemove(key, out var awaiter))
+            lock (_awaiters)
             {
-                awaiter.Complete(packet);
+                for (var i = _awaiters.Count - 1; i >= 0; i--)
-                return true;
+                {
+                    var entry = _awaiters[i];
+
+                    // Note: The PingRespPacket will also arrive here and has NO identifier but there
+                    // is code which waits for it. So the code must be able to deal with filters which
+                    // are referring to the type only (identifier is 0)!
+                    if (entry.PacketFilter.Type != packetType || entry.PacketFilter.Identifier != identifier)
+                    {
+                        continue;
+                    }
+                    
+                    matchingAwaiters.Add(entry);
+                    _awaiters.RemoveAt(i);
+                }
+            }
+            
+            foreach (var matchingEntry in matchingAwaiters)
+            {
+                matchingEntry.Complete(packet);
             }
 
-            return false;
+            return matchingAwaiters.Count > 0;
         }
-
+        
-        public void Dispatch(MqttBasePacket packet)
+        public MqttPacketAwaiter<TResponsePacket> AddAwaiter<TResponsePacket>(ushort packetIdentifier) where TResponsePacket : MqttBasePacket
         {
-            if (packet == null) throw new ArgumentNullException(nameof(packet));
+            var awaiter = new MqttPacketAwaiter<TResponsePacket>(packetIdentifier, this);
 
-            if (!TryDispatch(packet))
+            lock (_awaiters)
             {
-                throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
+                _awaiters.Add(awaiter);
             }
-        }
+            
-
-        public void Cancel()
-        {
-            foreach (var awaiter in _awaiters)
-            {
-                awaiter.Value.Cancel();
-            }
-
-            _awaiters.Clear();
-        }
-
-        public MqttPacketAwaiter<TResponsePacket> AddAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
-        {
-            if (!identifier.HasValue)
-            {
-                identifier = 0;
-            }
-
-            var awaiter = new MqttPacketAwaiter<TResponsePacket>(identifier, this);
-
-            var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
-            if (!_awaiters.TryAdd(key, awaiter))
-            {
-                throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{key.Item2.Name}' with identifier {key.Item1}.");
-            }
-
             return awaiter;
         }
 
-        public void RemoveAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
+        public void RemoveAwaiter(IMqttPacketAwaiter awaiter)
         {
-            if (!identifier.HasValue)
+            if (awaiter == null) throw new ArgumentNullException(nameof(awaiter));
+            
+            lock (_awaiters)
             {
-                identifier = 0;
+                _awaiters.Remove(awaiter);
             }
-
-            var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
-            _awaiters.TryRemove(key, out _);
         }
     }
 }

Source/MQTTnet/Packets/MqttPingReqPacket.cs

@@ -3,7 +3,7 @@
     public sealed class MqttPingReqPacket : MqttBasePacket
     {
         // This is a minor performance improvement.
-        public static MqttPingReqPacket Instance = new MqttPingReqPacket();
+        public static readonly MqttPingReqPacket Instance = new MqttPingReqPacket();
 
         public override string ToString()
         {

Source/MQTTnet/Packets/MqttPingRespPacket.cs

@@ -3,7 +3,7 @@
     public sealed class MqttPingRespPacket : MqttBasePacket
     {
         // This is a minor performance improvement.
-        public static MqttPingRespPacket Instance = new MqttPingRespPacket();
+        public static readonly MqttPingRespPacket Instance = new MqttPingRespPacket();
 
         public override string ToString()
         {

Source/MQTTnet/Server/IMqttServerOptions.cs

@@ -8,7 +8,7 @@ namespace MQTTnet.Server
         /// Gets the client identifier.
         /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        string ClientId { get; set; }
+        string ClientId { get; }
 
         bool EnablePersistentSessions { get; }
 
@@ -23,16 +23,13 @@ namespace MQTTnet.Server
         IMqttServerUnsubscriptionInterceptor UnsubscriptionInterceptor { get; }
         IMqttServerApplicationMessageInterceptor ApplicationMessageInterceptor { get; }
         IMqttServerClientMessageQueueInterceptor ClientMessageQueueInterceptor { get; }
-
+        IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; }
+        
         MqttServerTcpEndpointOptions DefaultEndpointOptions { get; }
         MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; }
 
         IMqttServerStorage Storage { get; }
 
         IMqttRetainedMessagesManager RetainedMessagesManager { get; }
-
-        IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
-
-        IMqttServerClientDisconnectedHandler ClientDisconnectedInterceptor { get; set; }
     }
 }

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -13,6 +13,7 @@ using System;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Client.Disconnecting;
 
 namespace MQTTnet.Server
 {
@@ -42,7 +43,7 @@ namespace MQTTnet.Server
         long _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere.
         long _receivedApplicationMessagesCount;
         long _sentApplicationMessagesCount;
-        MqttDisconnectReasonCode _disconnectReason;
+        MqttClientDisconnectReason _disconnectReason;
 
         public MqttClientConnection(MqttConnectPacket connectPacket,
             IMqttChannelAdapter channelAdapter,
@@ -84,12 +85,12 @@ namespace MQTTnet.Server
 
         public MqttClientSession Session { get; }
 
-        public async Task StopAsync(MqttDisconnectReasonCode reason)
+        public async Task StopAsync(MqttClientDisconnectReason reason)
         {
             Status = MqttClientConnectionStatus.Finalizing;
             _disconnectReason = reason;
 
-            if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout)
+            if (reason == MqttClientDisconnectReason.SessionTakenOver || reason == MqttClientDisconnectReason.KeepAliveTimeout)
             {
                 // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
                 // token because the entire connection is closed (disposed) as soon as the cancellation
@@ -97,10 +98,18 @@ namespace MQTTnet.Server
                 // at the client!
                 try
                 {
-                    await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket
+                    var disconnectOptions = new MqttClientDisconnectOptions
                     {
-                        ReasonCode = reason
+                        ReasonCode = reason,
-                    }, CancellationToken.None).ConfigureAwait(false);
+                        ReasonString = reason.ToString()
+                    };
+                    
+                    var disconnectPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateDisconnectPacket(disconnectOptions);
+
+                    using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
+                    {
+                        await _channelAdapter.SendPacketAsync(disconnectPacket, timeout.Token).ConfigureAwait(false);
+                    }
                 }
                 catch (Exception exception)
                 {
@@ -203,6 +212,10 @@ namespace MQTTnet.Server
                     {
                         await SendAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false);
                     }
+                    else if (packet is MqttPingRespPacket)
+                    {
+                        throw new MqttProtocolViolationException("A PINGRESP Packet is sent by the Server to the Client in response to a PINGREQ Packet only.");
+                    }
                     else if (packet is MqttDisconnectPacket)
                     {
                         Session.WillMessage = null;
@@ -213,7 +226,10 @@ namespace MQTTnet.Server
                     }
                     else
                     {
-                        _packetDispatcher.Dispatch(packet);
+                        if (!_packetDispatcher.TryDispatch(packet))
+                        {
+                            throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
+                        }
                     }
                 }
             }
@@ -235,7 +251,7 @@ namespace MQTTnet.Server
             }
             finally
             {
-                if (_disconnectReason == MqttDisconnectReasonCode.SessionTakenOver)
+                if (_disconnectReason == MqttClientDisconnectReason.SessionTakenOver)
                 {
                     disconnectType = MqttClientDisconnectType.Takeover;
                 }
@@ -246,7 +262,7 @@ namespace MQTTnet.Server
                     Session.WillMessage = null;
                 }
 
-                _packetDispatcher.Cancel();
+                _packetDispatcher.CancelAll();
 
                 _logger.Info("Client '{0}': Connection stopped.", ClientId);
 
@@ -277,12 +293,7 @@ namespace MQTTnet.Server
 
         Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
         {
-            var pubCompPacket = new MqttPubCompPacket
+            var pubCompPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            {
-                PacketIdentifier = pubRelPacket.PacketIdentifier,
-                ReasonCode = MqttPubCompReasonCode.Success
-            };
-
             return SendAsync(pubCompPacket, cancellationToken);
         }
 
@@ -327,12 +338,12 @@ namespace MQTTnet.Server
                     }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                     {
-                        var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket);
+                        var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
                         return SendAsync(pubAckPacket, cancellationToken);
                     }
                 case MqttQualityOfServiceLevel.ExactlyOnce:
                     {
-                        var pubRecPacket = _dataConverter.CreatePubRecPacket(publishPacket);
+                        var pubRecPacket = _dataConverter.CreatePubRecPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
                         return SendAsync(pubRecPacket, cancellationToken);
                     }
                 default:
@@ -441,13 +452,10 @@ namespace MQTTnet.Server
                         using (var awaiter2 = _packetDispatcher.AddAwaiter<MqttPubCompPacket>(publishPacket.PacketIdentifier))
                         {
                             await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false);
-                            await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
+                            var pubRecPacket = await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
 
-                            await SendAsync(new MqttPubRelPacket
+                            var pubRelPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-                            {
+                            await SendAsync(pubRelPacket, cancellationToken).ConfigureAwait(false);
-                                PacketIdentifier = publishPacket.PacketIdentifier,
-                                ReasonCode = MqttPubRelReasonCode.Success
-                            }, cancellationToken).ConfigureAwait(false);
 
                             await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
                         }

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -1,4 +1,4 @@
-using MQTTnet.Adapter;
+using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
@@ -12,7 +12,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using OperationCanceledException = System.OperationCanceledException;
+using MQTTnet.Client.Disconnecting;
 
 namespace MQTTnet.Server
 {
@@ -20,7 +20,7 @@ namespace MQTTnet.Server
     {
         readonly BlockingCollection<MqttPendingApplicationMessage> _messageQueue = new BlockingCollection<MqttPendingApplicationMessage>();
 
-        readonly object _createConnectionSyncRoot = new object();
+        readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
         readonly Dictionary<string, MqttClientConnection> _connections = new Dictionary<string, MqttClientConnection>();
         readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
 
@@ -97,7 +97,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                var connection = CreateClientConnection(connectPacket, connectionValidatorContext, channelAdapter);
+                var connection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);
                 await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket.ClientId).ConfigureAwait(false);
                 await connection.RunAsync().ConfigureAwait(false);
             }
@@ -121,7 +121,7 @@ namespace MQTTnet.Server
 
             foreach (var connection in connections)
             {
-                await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false);
             }
         }
 
@@ -211,7 +211,7 @@ namespace MQTTnet.Server
 
             if (connection != null)
             {
-                await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false);
             }
             
             _logger.Verbose("Session for client '{0}' deleted.", clientId);
@@ -236,11 +236,13 @@ namespace MQTTnet.Server
                 }
             }
 
+            var endpoint = channelAdapter.Endpoint;
+
             await SafeCleanupChannelAsync(channelAdapter).ConfigureAwait(false);
 
             if (clientId != null)
             {
-                await _eventDispatcher.SafeNotifyClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyClientDisconnectedAsync(clientId, disconnectType, endpoint).ConfigureAwait(false);
             }
         }
 
@@ -301,7 +303,7 @@ namespace MQTTnet.Server
                         {
                             if (clientConnection != null)
                             {
-                                await clientConnection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                                await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false);
                             }
                         }
 
@@ -386,9 +388,12 @@ namespace MQTTnet.Server
             return context;
         }
 
-        MqttClientConnection CreateClientConnection(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
+        async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
         {
-            lock (_createConnectionSyncRoot)
+            MqttClientConnection existingConnection;
+            MqttClientConnection connection;
+
+            using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false))
             {
                 MqttClientSession session;
                 lock (_sessions)
@@ -414,8 +419,6 @@ namespace MQTTnet.Server
                     _sessions[connectPacket.ClientId] = session;
                 }
 
-                MqttClientConnection existingConnection;
-                MqttClientConnection connection;
                 lock (_connections)
                 {
                     _connections.TryGetValue(connectPacket.ClientId, out existingConnection);
@@ -424,10 +427,13 @@ namespace MQTTnet.Server
                     _connections[connectPacket.ClientId] = connection;
                 }
 
-                existingConnection?.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).GetAwaiter().GetResult();
+                if (existingConnection != null)
-
+                {
-                return connection;
+                    await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false);
+                }
             }
+
+            return connection;
         }
 
         async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(IMqttServerApplicationMessageInterceptor interceptor, MqttClientConnection clientConnection, MqttApplicationMessage applicationMessage)

Source/MQTTnet/Server/MqttServerClientDisconnectedEventArgs.cs

@@ -1,13 +1,14 @@
-using System;
+using System;
 
 namespace MQTTnet.Server
 {
     public class MqttServerClientDisconnectedEventArgs : EventArgs
     {
-        public MqttServerClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType)
+        public MqttServerClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType, string endpoint)
         {
             ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
             DisconnectType = disconnectType;
+            Endpoint = endpoint;
         }
 
         /// <summary>
@@ -17,5 +18,7 @@ namespace MQTTnet.Server
         public string ClientId { get; }
 
         public MqttClientDisconnectType DisconnectType { get; }
+
+        public string Endpoint { get; }
     }
 }

Source/MQTTnet/Server/MqttServerEventDispatcher.cs

@@ -1,4 +1,4 @@
-using MQTTnet.Client.Receiving;
+using MQTTnet.Client.Receiving;
 using MQTTnet.Diagnostics;
 using System;
 using System.Threading.Tasks;
@@ -44,7 +44,7 @@ namespace MQTTnet.Server
             }
         }
 
-        public async Task SafeNotifyClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType)
+        public async Task SafeNotifyClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType, string endpoint)
         {
             try
             {
@@ -54,7 +54,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                await handler.HandleClientDisconnectedAsync(new MqttServerClientDisconnectedEventArgs(clientId, disconnectType)).ConfigureAwait(false);
+                await handler.HandleClientDisconnectedAsync(new MqttServerClientDisconnectedEventArgs(clientId, disconnectType, endpoint)).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -108,7 +108,7 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)).ConfigureAwait(false);
+                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage, null, null)).ConfigureAwait(false);
             }
             catch (Exception exception)
             {

Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs

@@ -3,8 +3,8 @@ using MQTTnet.Internal;
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Client.Disconnecting;
 using MQTTnet.Implementations;
-using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
@@ -108,7 +108,7 @@ namespace MQTTnet.Server
                 // We do not need to wait for the task so no await is needed.
                 // Also the internal state of the connection must be swapped to "Finalizing" because the
                 // next iteration of the keep alive timer happens.
-                var _ = connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout);
+                var _ = connection.StopAsync(MqttClientDisconnectReason.KeepAliveTimeout);
             }
             catch (Exception exception)
             {

Source/MQTTnet/Server/MqttServerOptions.cs

@@ -34,12 +34,10 @@ namespace MQTTnet.Server
 
         public IMqttServerUnsubscriptionInterceptor UnsubscriptionInterceptor { get; set; }
 
+        public IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
+
         public IMqttServerStorage Storage { get; set; }
 
         public IMqttRetainedMessagesManager RetainedMessagesManager { get; set; } = new MqttRetainedMessagesManager();
-
-        public IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
-
-        public IMqttServerClientDisconnectedHandler ClientDisconnectedInterceptor { get; set; }
     }
 }

Source/MQTTnet/Server/MqttServerOptionsBuilder.cs

@@ -164,19 +164,13 @@ namespace MQTTnet.Server
             _options.ConnectionValidator = new MqttServerConnectionValidatorDelegate(value);
             return this;
         }
-
+        
-        public MqttServerOptionsBuilder WithDisconnectedInterceptor(IMqttServerClientDisconnectedHandler value)
+        public MqttServerOptionsBuilder WithConnectionValidator(Func<MqttConnectionValidatorContext, Task> value)
         {
-            _options.ClientDisconnectedInterceptor = value;
+            _options.ConnectionValidator = new MqttServerConnectionValidatorDelegate(value);
             return this;
         }
-
+        
-        public MqttServerOptionsBuilder WithDisconnectedInterceptor(Action<MqttServerClientDisconnectedEventArgs> value)
-        {
-            _options.ClientDisconnectedInterceptor = new MqttServerClientDisconnectedHandlerDelegate(value);
-            return this;
-        }
-
         public MqttServerOptionsBuilder WithApplicationMessageInterceptor(IMqttServerApplicationMessageInterceptor value)
         {
             _options.ApplicationMessageInterceptor = value;

Source/MQTTnet/Server/Status/MqttClientStatus.cs

@@ -1,7 +1,7 @@
 using MQTTnet.Formatter;
 using System;
 using System.Threading.Tasks;
-using MQTTnet.Protocol;
+using MQTTnet.Client.Disconnecting;
 
 namespace MQTTnet.Server.Status
 {
@@ -46,7 +46,7 @@ namespace MQTTnet.Server.Status
 
         public Task DisconnectAsync()
         {
-            return _connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection);
+            return _connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection);
         }
 
         public void ResetStatistics()

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

@@ -9,6 +9,8 @@ using System.Diagnostics;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Extensions.Rpc;
+using MQTTnet.Extensions.Rpc.Options;
 using MQTTnet.LowLevelClient;
 
 namespace MQTTnet.Tests.Mockups
@@ -77,6 +79,11 @@ namespace MQTTnet.Tests.Mockups
             };
         }
 
+        public async Task<IMqttRpcClient> ConnectRpcClientAsync(IMqttRpcClientOptions options)
+        {
+            return new MqttRpcClient(await ConnectClientAsync(), options);
+        }
+
         public IMqttClient CreateClient()
         {
             lock (_clients)

Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs

@@ -109,15 +109,6 @@ namespace MQTTnet.Tests.Mockups
             return Implementation.StartAsync(options);
         }
 
-        public void ConnectionValidator(MqttConnectionValidatorContext ctx)
-        {
-            if (!ctx.ClientId.StartsWith(TestContext.TestName))
-            {
-                TestEnvironment.TrackException(new InvalidOperationException($"Invalid client ID used ({ctx.ClientId}). It must start with UnitTest name."));
-                ctx.ReasonCode = Protocol.MqttConnectReasonCode.ClientIdentifierNotValid;
-            }
-        }
-
         public Task StopAsync()
         {
             return Implementation.StopAsync();
@@ -137,5 +128,14 @@ namespace MQTTnet.Tests.Mockups
         {
             Implementation.Dispose();
         }
+        
+        void ConnectionValidator(MqttConnectionValidatorContext ctx)
+        {
+            if (!ctx.ClientId.StartsWith(TestContext.TestName))
+            {
+                TestEnvironment.TrackException(new InvalidOperationException($"Invalid client ID used ({ctx.ClientId}). It must start with UnitTest name."));
+                ctx.ReasonCode = Protocol.MqttConnectReasonCode.ClientIdentifierNotValid;
+            }
+        }
     }
 }

Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs

@@ -445,6 +445,55 @@ namespace MQTTnet.Tests
             }
         }
 
+        [TestMethod]
+        public async Task Preserve_Message_Order_With_Delayed_Acknowledgement()
+        {
+            // The messages are sent in reverse or to ensure that the delay in the handler
+            // needs longer for the first messages and later messages may be processed earlier (if there
+            // is an issue).
+            const int MessagesCount = 50;
+
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServerAsync();
+
+                var client1 = await testEnvironment.ConnectClientAsync();
+                await client1.SubscribeAsync("x", MqttQualityOfServiceLevel.ExactlyOnce);
+
+                var receivedValues = new List<int>();
+
+                Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs)
+                {
+                    var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString());
+                    eventArgs.AutoAcknowledge = false;
+                    Task.Delay(value).ContinueWith(x => eventArgs.AcknowledgeAsync(CancellationToken.None));
+
+                    System.Diagnostics.Debug.WriteLine($"received {value}");
+                    lock (receivedValues)
+                    {
+                        receivedValues.Add(value);
+                    }
+
+                    return Task.CompletedTask;
+                }
+
+                client1.UseApplicationMessageReceivedHandler(Handler1);
+
+                var client2 = await testEnvironment.ConnectClientAsync();
+                for (var i = MessagesCount; i > 0; i--)
+                {
+                    await client2.PublishAsync("x", i.ToString(), MqttQualityOfServiceLevel.ExactlyOnce);
+                }
+
+                await Task.Delay(5000);
+
+                for (var i = MessagesCount; i > 0; i--)
+                {
+                    Assert.AreEqual(i, receivedValues[MessagesCount - i]);
+                }
+            }
+        }
+
         [TestMethod]
         public async Task Send_Reply_For_Any_Received_Message()
         {

Tests/MQTTnet.Core.Tests/RPC_Tests.cs

@@ -81,18 +81,19 @@ namespace MQTTnet.Tests
 
                 var requestSender = await testEnvironment.ConnectClientAsync();
 
-                var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()) .Build());
+                var rpcClient = await testEnvironment.ConnectRpcClientAsync(new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()) .Build());
+                
                 await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
             }
         }
 
-        private async Task Execute_Success(MqttQualityOfServiceLevel qosLevel, MqttProtocolVersion protocolVersion)
+        async Task Execute_Success(MqttQualityOfServiceLevel qosLevel, MqttProtocolVersion protocolVersion)
         {
             using (var testEnvironment = new TestEnvironment(TestContext))
             {
                 await testEnvironment.StartServerAsync();
                 var responseSender = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(protocolVersion));
-                await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping");
+                await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", qosLevel);
 
                 responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
                 {
@@ -101,14 +102,16 @@ namespace MQTTnet.Tests
 
                 var requestSender = await testEnvironment.ConnectClientAsync();
 
-                var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build());
+                using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
-                var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel);
+                {
-
+                    var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel);
-                Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+                    
+                    Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+                }
             }
         }
 
-        private class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
+        class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
         {
             public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
             {

Tests/MQTTnet.Core.Tests/Server_Tests.cs

@@ -1,4 +1,4 @@
-using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Adapter;
 using MQTTnet.Client;
 using MQTTnet.Client.Connecting;
@@ -6,6 +6,7 @@ using MQTTnet.Client.Disconnecting;
 using MQTTnet.Client.Options;
 using MQTTnet.Client.Receiving;
 using MQTTnet.Client.Subscribing;
+using MQTTnet.Exceptions;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
@@ -25,6 +26,55 @@ namespace MQTTnet.Tests
     {
         public TestContext TestContext { get; set; }
 
+        [TestMethod]
+        [DataRow("", null)]
+        [DataRow("", "")]
+        [DataRow(null, null)]
+        public async Task Use_Admissible_Credentials(string username, string password)
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+
+                var client = testEnvironment.CreateClient();
+
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("localhost", testEnvironment.ServerPort)
+                    .WithCredentials(username, password)
+                    .Build();
+
+                var connectResult = await client.ConnectAsync(clientOptions);
+
+                Assert.IsFalse(connectResult.IsSessionPresent);
+                Assert.IsTrue(client.IsConnected);
+            }
+        }
+
+        [TestMethod]
+        public async Task Use_Username_Null_Password_Empty()
+        {
+            string username = null;
+            string password = string.Empty;
+
+            using (var testEnvironment = new TestEnvironment())
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+
+                await testEnvironment.StartServerAsync();
+
+                var client = testEnvironment.CreateClient();
+
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("localhost", testEnvironment.ServerPort)
+                    .WithCredentials(username, password)
+                    .Build();
+
+                var ex = await Assert.ThrowsExceptionAsync<MqttConnectingFailedException>(async () => await client.ConnectAsync(clientOptions));
+                Assert.IsInstanceOfType(ex.InnerException, typeof(MqttProtocolViolationException));
+                Assert.AreEqual("Error while authenticating. If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22].", ex.Message, false);
+            }
+        }
+
         [TestMethod]
         public async Task Use_Empty_Client_ID()
         {
@@ -416,7 +466,7 @@ namespace MQTTnet.Tests
         public async Task Publish_Multiple_Clients()
         {
             var receivedMessagesCount = 0;
-            
+
             using (var testEnvironment = new TestEnvironment(TestContext))
             {
                 await testEnvironment.StartServerAsync();
@@ -446,7 +496,7 @@ namespace MQTTnet.Tests
                 }
 
                 SpinWait.SpinUntil(() => receivedMessagesCount == 1000, TimeSpan.FromSeconds(20));
-                
+
                 Assert.AreEqual(1000, receivedMessagesCount);
             }
         }