Improve RPC client handling.

Update codeql-analysis.yml
Merge remote-tracking branch 'origin/master'
2026-10-15 17:00:03 +08:00 · 2021-02-26 10:55:26 +01:00 · 2021-02-23 12:05:24 +01:00 · 2021-02-23 11:46:38 +01:00 · 2021-02-23 11:46:16 +01:00 · 2021-02-23 11:45:22 +01:00
37 changed files with 610 additions and 392 deletions
@@ -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
@@ -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\" />
@@ -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>
@@ -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\" />

@@ -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="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>
+      <dependency id="MQTTnet" version="$nugetVersion" />
+      <dependency id="WebSocket4Net" version="0.15.2" />
    </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\" />

@@ -12,54 +12,31 @@
    <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] Added support for TLS 1.3 (requires .NET Core 3.1+) (thanks to @Dvergatal).
-* [Core] Added support for _MaximumQoS_ transmit when using MQTTv5 (thanks to @nayato).
-* [Core] Aligned the format of some log messages.
-* [Client] Fixed an issue with connection state detection (thanks to @tobiasfrick).
-* [Client] Fixed an issue when receiving partial QoS 2 packets after a reconnect.
-* [Client] Added new API for raw packet inspection.
-* [ManagedClient] Removed _AutoReconnect_ since this does no longer work (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.
+* [Core] Fixed some issues in nuget packages.
+* [Client] Changed exception types so that proper exceptions are thrown when connecting (#1082).
+* [Client] Exposed Dup flag in application messages.
+* [Client] Improved keep alive message sending.
+* [RpcClient] Fixed an issue when using a custom application message reveived handler (#1093).
+* [RpcClient] Fixed a race condition when using timeout and cancellation token at the same time (BREAKING CHANGE!).
+* [Server] Improved keep alive message sending.

 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="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="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" />
+      <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>
@@ -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"
@@ -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>
@@ -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);
    }
 }
@@ -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)
-        {
-            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)
+        public async Task<byte[]> ExecuteAsync(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 (!_waitingCalls.TryAdd(responseTopic, tcs))
+#if NET452
+                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);
-                await _mqttClient.PublishAsync(requestMessage).ConfigureAwait(false);
+                var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
+                    .WithTopicFilter(responseTopic, qualityOfServiceLevel)
+                    .Build();

-                using (var timeoutCts = new CancellationTokenSource(timeout))
-                using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token))
+                await _mqttClient.SubscribeAsync(subscribeOptions, cancellationToken).ConfigureAwait(false);
+                await _mqttClient.PublishAsync(requestMessage, cancellationToken).ConfigureAwait(false);
+
+                using (cancellationToken.Register(() => { promise.TrySetCanceled(); }))
                {
-                    linkedCts.Token.Register(() =>
-                    {
-                        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;
-                        }
-                    }
+                    return await promise.Task.ConfigureAwait(false);
                }
            }
            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();
        }
    }
-}
+}
@@ -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);
+        }
+    }
+}
@@ -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);
        }
    }
 }
@@ -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)
-            : base($"Connecting with MQTT server failed ({result.ResultCode.ToString()}).")
+        public MqttConnectingFailedException(string message, Exception innerException, MqttClientAuthenticateResult authenticateResult)
+            : base(message, innerException)
        {
-            Result = result;
+            Result = authenticateResult;
        }

        public MqttClientAuthenticateResult Result { get; }
-        public MqttClientConnectResultCode ResultCode => Result.ResultCode;
+
+        public MqttClientConnectResultCode ResultCode => Result?.ResultCode ?? MqttClientConnectResultCode.UnspecifiedError;
    }
 }
@@ -12,7 +12,7 @@
        ServerBusy = 137,
        ServerShuttingDown = 139,
        BadAuthenticationMethod = 140,
-        KeepaliveTimeout = 141,
+        KeepAliveTimeout = 141,
        SessionTakenOver = 142,
        TopicFilterInvalid = 143,
        TopicNameInvalid = 144,
@@ -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;
@@ -44,6 +41,8 @@ namespace MQTTnet.Client
        long _isDisconnectPending;
        bool _isConnected;
        MqttClientDisconnectReason _disconnectReason;
+        
+        DateTime _lastPacketSentTimestamp;

        public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
        {
@@ -79,7 +78,7 @@ namespace MQTTnet.Client
                Options = options;

                _packetIdentifierProvider.Reset();
-                _packetDispatcher.Cancel();
+                _packetDispatcher.CancelAll();

                _backgroundCancellationTokenSource = new CancellationTokenSource();
                var backgroundCancellationToken = _backgroundCancellationTokenSource.Token;
@@ -102,8 +101,7 @@ namespace MQTTnet.Client
                    authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false);
                }

-                _sendTracker.Restart();
-                _receiveTracker.Restart();
+                _lastPacketSentTimestamp = DateTime.UtcNow;

                if (Options.KeepAlivePeriod != TimeSpan.Zero)
                {
@@ -167,7 +165,7 @@ namespace MQTTnet.Client

        public Task PingAsync(CancellationToken cancellationToken)
        {
-            return SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken);
+            return SendAndReceiveAsync<MqttPingRespPacket>(MqttPingReqPacket.Instance, cancellationToken);
        }

        public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken)
@@ -280,16 +278,25 @@ namespace MQTTnet.Client

        async Task<MqttClientAuthenticateResult> AuthenticateAsync(IMqttChannelAdapter channelAdapter, MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken)
        {
-            var connectPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnectPacket(
-                willApplicationMessage,
-                Options);
+            MqttClientAuthenticateResult result;

-            var connAckPacket = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket, cancellationToken).ConfigureAwait(false);
-            var result = channelAdapter.PacketFormatterAdapter.DataConverter.CreateClientConnectResult(connAckPacket);
+            try
+            {
+                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.");
@@ -382,8 +389,8 @@ namespace MQTTnet.Client
        {
            cancellationToken.ThrowIfCancellationRequested();

-            _sendTracker.Restart();
-
+            _lastPacketSentTimestamp = DateTime.UtcNow;
+            
            return _adapter.SendPacketAsync(packet, cancellationToken);
        }

@@ -391,23 +398,22 @@ namespace MQTTnet.Client
        {
            cancellationToken.ThrowIfCancellationRequested();

-            ushort identifier = 0;
-            if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier > 0)
+            ushort 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 _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
+                    await SendAsync(requestPacket, cancellationToken).ConfigureAwait(false);
                }
-                catch (Exception e)
+                catch (Exception exception)
                {
-                    _logger.Warning(e, "Error when sending request packet ({0}).", requestPacket.GetType().Name);
-                    packetAwaiter.Cancel();
+                    _logger.Warning(exception, "Error when sending request packet ({0}).", requestPacket.GetType().Name);
+                    packetAwaiter.Fail(exception);
                }

                try
@@ -437,15 +443,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);
@@ -529,7 +535,7 @@ namespace MQTTnet.Client
                    _logger.Error(exception, "Error while receiving packets.");
                }

-                _packetDispatcher.Dispatch(exception);
+                _packetDispatcher.FailAll(exception);

                if (!DisconnectIsPending())
                {
@@ -546,8 +552,6 @@ namespace MQTTnet.Client
        {
            try
            {
-                _receiveTracker.Restart();
-
                if (packet is MqttPublishPacket publishPacket)
                {
                    EnqueueReceivedPublishPacket(publishPacket);
@@ -560,10 +564,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 +572,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,7 +607,7 @@ namespace MQTTnet.Client
                    _logger.Error(exception, "Error while receiving packets.");
                }

-                _packetDispatcher.Dispatch(exception);
+                _packetDispatcher.FailAll(exception);

                if (!DisconnectIsPending())
                {
@@ -641,11 +652,8 @@ namespace MQTTnet.Client

                        if (!eventArgs.ProcessingFailed)
                        {
-                            await SendAsync(new MqttPubAckPacket
-                            {
-                                PacketIdentifier = publishPacket.PacketIdentifier,
-                                ReasonCode = (MqttPubAckReasonCode)eventArgs.ReasonCode
-                            }, cancellationToken).ConfigureAwait(false);
+                            var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, eventArgs.ReasonCode);
+                            await SendAsync(pubAckPacket, cancellationToken).ConfigureAwait(false);
                        }
                    }
                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -654,11 +662,8 @@ namespace MQTTnet.Client

                        if (!eventArgs.ProcessingFailed)
                        {
-                            await SendAsync(new MqttPubRecPacket
-                            {
-                                PacketIdentifier = publishPacket.PacketIdentifier,
-                                ReasonCode = (MqttPubRecReasonCode)eventArgs.ReasonCode
-                            }, cancellationToken).ConfigureAwait(false);
+                            var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, eventArgs.ReasonCode);
+                            await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
                        }
                    }
                    else
@@ -682,11 +687,8 @@ namespace MQTTnet.Client
            {
                // 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
-                {
-                    PacketIdentifier = pubRecPacket.PacketIdentifier,
-                    ReasonCode = MqttPubRelReasonCode.PacketIdentifierNotFound
-                }, cancellationToken);
+                var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
+                return SendAsync(pubRelPacket, cancellationToken);
            }

            return PlatformAbstractionLayer.CompletedTask;
@@ -694,11 +696,8 @@ namespace MQTTnet.Client

        Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
        {
-            return SendAsync(new MqttPubCompPacket
-            {
-                PacketIdentifier = pubRelPacket.PacketIdentifier,
-                ReasonCode = MqttPubCompReasonCode.Success
-            }, cancellationToken);
+            var pubCompPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
+            return SendAsync(pubCompPacket, cancellationToken);
        }

        Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket)
@@ -706,8 +705,8 @@ namespace MQTTnet.Client
            _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);
@@ -731,14 +730,14 @@ 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);
+            return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(response);
        }

        async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
@@ -747,15 +746,11 @@ namespace MQTTnet.Client

            var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);

-            var pubRelPacket = new MqttPubRelPacket
-            {
-                PacketIdentifier = publishPacket.PacketIdentifier,
-                ReasonCode = MqttPubRelReasonCode.Success
-            };
-
+            var pubRelPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.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)
@@ -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);
-
-        MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket);
+        MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode);
+        
+        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);
    }
 }
@@ -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,
-                ReasonCode = MqttPubAckReasonCode.Success
+                PacketIdentifier = publishPacket.PacketIdentifier
            };
        }

-        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
+                PacketIdentifier = publishPacket.PacketIdentifier
+            };
+        }
+
+        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,
@@ -223,7 +252,7 @@ namespace MQTTnet.Formatter.V3
            return new MqttDisconnectPacket();
        }

-        public MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket)
+        public MqttClientPublishResult CreateClientPublishResult(MqttPubAckPacket pubAckPacket)
        {
            return new MqttClientPublishResult
            {
@@ -232,7 +261,7 @@ namespace MQTTnet.Formatter.V3
            };
        }

-        public MqttClientPublishResult CreatePublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
+        public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
        {
            if (pubRecPacket == null || pubCompPacket == null)
            {
@@ -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;
        }
    }
-}
+}
@@ -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>
@@ -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.
@@ -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[];
-
-            if (_payload == null)
-            {
-                _payload = payload.ToArray();
-            }
+            _payload = payload as byte[] ?? 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,
@@ -21,5 +21,13 @@ 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; }
+        
+        public object Tag { get; set; }
    }
 }
@@ -9,6 +9,7 @@
        NotAuthorized = 135,
        TopicNameInvalid = 144,
        PacketIdentifierInUse = 145,
+        PacketIdentifierNotFound = 146,
        QuotaExceeded = 151,
        PayloadFormatInvalid = 153
    }
@@ -5,6 +5,8 @@ namespace MQTTnet.PacketDispatcher
 {
    public interface IMqttPacketAwaiter : IDisposable
    {
+        MqttPacketAwaiterPacketFilter PacketFilter { get; }
+        
        void Complete(MqttBasePacket packet);

        void Fail(Exception exception);
@@ -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);
        }
    }
 }
@@ -0,0 +1,11 @@
+using System;
+
+namespace MQTTnet.PacketDispatcher
+{
+    public sealed class MqttPacketAwaiterPacketFilter
+    {
+        public Type Type { get; set; }
+        
+        public ushort Identifier { get; set; }
+    }
+}
@@ -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)
-            {
-                awaiter.Value.Fail(exception);
-            }
+            if (exception == null) throw new ArgumentNullException(nameof(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 key = new Tuple<ushort, Type>(identifier, type);
-
-            if (_awaiters.TryRemove(key, out var awaiter))
+            var packetType = packet.GetType();
+            var matchingAwaiters = new List<IMqttPacketAwaiter>();
+            
+            lock (_awaiters)
            {
-                awaiter.Complete(packet);
-                return true;
+                for (var i = _awaiters.Count - 1; i >= 0; i--)
+                {
+                    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 _);
        }
    }
 }
@@ -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()
        {
@@ -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()
        {
@@ -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
-                    }, CancellationToken.None).ConfigureAwait(false);
+                        ReasonCode = reason,
+                        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
-            {
-                PacketIdentifier = pubRelPacket.PacketIdentifier,
-                ReasonCode = MqttPubCompReasonCode.Success
-            };
-
+            var pubCompPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.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
-                            {
-                                PacketIdentifier = publishPacket.PacketIdentifier,
-                                ReasonCode = MqttPubRelReasonCode.Success
-                            }, cancellationToken).ConfigureAwait(false);
+                            var pubRelPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
+                            await SendAsync(pubRelPacket, cancellationToken).ConfigureAwait(false);

                            await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
                        }
@@ -12,6 +12,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Client.Disconnecting;
 using OperationCanceledException = System.OperationCanceledException;

 namespace MQTTnet.Server
@@ -121,7 +122,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 +212,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);
@@ -301,7 +302,7 @@ namespace MQTTnet.Server
                        {
                            if (clientConnection != null)
                            {
-                                await clientConnection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false);
+                                await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false);
                            }
                        }

@@ -424,7 +425,7 @@ namespace MQTTnet.Server
                    _connections[connectPacket.ClientId] = connection;
                }

-                existingConnection?.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).GetAwaiter().GetResult();
+                existingConnection?.StopAsync(MqttClientDisconnectReason.SessionTakenOver).GetAwaiter().GetResult();

                return connection;
            }
@@ -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)
            {
@@ -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()
@@ -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)
@@ -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;
+            }
+        }
    }
 }
@@ -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());
-                var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel);
-
-                Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+                using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
+                {
+                    var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", qosLevel);
+                    
+                    Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+                }
            }
        }

-        private class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
+        class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
        {
            public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)
            {
@@ -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);
            }
        }
Author	SHA1	Message	Date
Christian	12f346115f	Improve RPC client handling.	2021-02-26 10:55:26 +01:00
Christian	05827341d4	Update codeql-analysis.yml	2021-02-23 12:05:24 +01:00
Christian	1e88d6a05b	Merge remote-tracking branch 'origin/master'	2021-02-23 11:46:38 +01:00
Christian	8261ba9dc1	Update docs.	2021-02-23 11:46:16 +01:00
Christian	b766c1bebb	Refactor keep alive message handling in order to fix issues with sudden disconnects.	2021-02-23 11:45:22 +01:00
Christian	6c188899fb	Create codeql-analysis.yml	2021-02-21 11:33:51 +01:00
Christian	2d4ea8ee9b	Improve nuget packages.	2021-02-19 21:52:42 +01:00
Christian	15d3761843	Merge pull request #1097 from yyjdelete/nuspec Update nuspecs	2021-02-19 21:50:49 +01:00
Christian	ed520455d0	Use the data converter only when creating packets.	2021-02-19 21:45:23 +01:00
Christian	cfc4ebe6a3	Expose Dup flag in application messages (#1096 ).	2021-02-19 21:39:23 +01:00
Christian	120b7fb7d1	Fix RPC issue #1093 by calling both message handlers.	2021-02-18 18:56:15 +01:00
SilverFox	8b164a4846	Update nuspecs Manually sync with the file generated by `msbuild /t:pack` or `dotnet pack`, and remove some unnessary dependencies	2021-02-18 11:26:54 +08:00
Christian	fda4272611	Update docs.	2021-02-11 12:31:13 +01:00
Christian	88cc79d52f	Fix issues in nuspecs.	2021-02-11 12:25:58 +01:00
Christian	8ba960343c	Fix exception handling issues (#1082 ).	2021-02-08 13:29:15 +01:00
Christian	0976fa682a	Merge pull request #1083 from milbrandt/bugfix/noUsername Test for Regression: No exception on ConnectAsync(null, password) [MQTT-3.1.2-22]	2021-02-06 13:38:04 +01:00
Christian	18cc454842	Fix build script for .net5.0.	2021-02-06 13:36:23 +01:00
Volker Milbrandt	e342a51d50	Merge branch 'master'	2021-02-04 21:23:02 +01:00
Volker Milbrandt	57652c76cf	tests with MqttClient.ConnectAsync with credentials for all four combinations of null/non-null for username and password	2021-02-04 21:09:36 +01:00