Update MQTTnet.nuspec

Thank you, this is indeed a bug.

.github/workflows/ci.yml

@@ -0,0 +1,34 @@
+name: CI
+
+on: [push]
+
+jobs:
+  build:
+
+    runs-on: windows-latest
+    strategy:
+      matrix:
+        dotnet-version: ['6.0.x']
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Setup .NET Core SDK ${{ matrix.dotnet-version }}
+        uses: actions/setup-dotnet@v1.7.2
+        with:
+          dotnet-version: ${{ matrix.dotnet-version }}
+
+      - name: Build
+        shell: pwsh
+        run: |
+          Set-Location -Path "Build"
+          .\build.ps1 1.2.3 4.5.6-rc1
+
+
+#      - name: Install dependencies
+#        run: dotnet restore MQTTnet.sln
+#      - name: Build
+#        run: dotnet build --configuration Release --no-restore MQTTnet.sln
+#      - name: Test MQTTnet
+#        run: dotnet test --no-restore --verbosity normal Tests\MQTTnet.Core.Tests\MQTTnet.Tests.csproj
+#      - name: Test AspNetCore
+#        run: dotnet test --no-restore --verbosity normal Tests\MQTTnet.AspNetCore.Tests\MQTTnet.AspNetCore.Tests.csproj

.github/workflows/codeql-analysis.yml

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

.github/workflows/deb-packaging.yml

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

Build/MQTTnet.nuspec

@@ -12,26 +12,11 @@
     <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 all builders to the MQTT factory.
+* [Client] Increased delay for keep alive checks do decrease CPU load.
-* [Core] Removed global logger and refactored logging (BREAKING CHANGE!).
+* [Core] Decreased object allocations (#1324, thanks to @gfoidl).
-* [Client] Renamed _MqttClientConnectResult_ to _MqttClientConnectResult_ (BREAKING CHANGE!).
+* [Core] Decreased object allocations when logging is not active (thanks to @gfoidl, @Tymoniden).
-* [ManagedClient] Extended ReconnectAsync (thanks to @nvsnkv, #1202).
+* [Client] Fixed issue in _MqttApplicationMessageBuilder.WithPayload_ (#1322, thanks to @gfoidl).
-* [ManagedClient] Improved Amazon AWS support (thanks to @scottbrogden-iheartmedia, #1209). 
+
-* [ManagedClient] Fixed bug that allowed invalid subscriptions (Thanks to @marcelwinh).
-* [Server] Added support for _RetainHandling_ (MQTTv5, BREAKING CHANGE!).
-* [Server] Added support for _NoLocal_ (MQTTv5, BREAKING CHANGE!).
-* [Server] Added support for _SubscriptionIdentifier_ (MQTTv5, BREAKING CHANGE!).
-* [Server] Server now reports supported features properly after successful connection (MQTTv5, BREAKING CHANGE!).
-* [Server] Fixed a memory/performance leak when using QoS Level 1.
-* [Server] Exposed connection timestamp in client status. 
-* [Server] Refactored connection management code. 
-* [Server] Exposed more details in _MqttServerClientConnectedEventArgs_. 
-* [Server] Processing all pending messages before stopping (thanks to @AblEdge, #1234).
-* [Server] Added support for a custom exception handler in _MqttServerMultiThreadedApplicationMessageInterceptorDelegate_.
-* [Server] Removed logger from _MqttServerMultiThreadedApplicationMessageInterceptorDelegate_ (BREAKING CHANGE!).
-* [Server] Fixed a memory leak when deleting sessions.
-* [Server] Fixed timestamp check in server keep-alive monitor (thanks to @logicaloud, #1277).
-* [MQTTnet.Server] Moved server project to a dedicated GitHub repository.
 Git commit: $gitCommit
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2021</copyright>

Build/build.ps1

@@ -5,6 +5,7 @@ if ([string]::IsNullOrEmpty($nugetVersion)) {$nugetVersion = "0.0.1"}
 
 $vswhere = ${Env:\ProgramFiles(x86)} + '\Microsoft Visual Studio\Installer\vswhere'
 $msbuild = &$vswhere -products * -requires Microsoft.Component.MSBuild -latest -find MSBuild\**\Bin\MSBuild.exe
+$vstest = &$vswhere -products * -latest -find **\TestWindow\**\vstest.console.exe
 
 Write-Host
 Write-Host "Assembly version = $assemblyVersion"
@@ -20,8 +21,8 @@ Invoke-WebRequest -Uri "https://dist.nuget.org/win-x86-commandline/latest/nuget.
 &$msbuild ..\Tests\MQTTnet.Core.Tests\MQTTnet.Tests.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netcoreapp3.1" /verbosity:m
 &$msbuild ..\Tests\MQTTnet.AspNetCore.Tests\MQTTnet.AspNetCore.Tests.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netcoreapp3.1" /verbosity:m
 
-vstest.console.exe ..\Tests\MQTTnet.Core.Tests\bin\Release\netcoreapp3.1\MQTTnet.Tests.dll
+&$vstest ..\Tests\MQTTnet.Core.Tests\bin\Release\netcoreapp3.1\MQTTnet.Tests.dll
-vstest.console.exe ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.Tests.dll
+&$vstest ..\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.Tests.dll
 
 # Build the core library
 &$msbuild ..\Source\MQTTnet\MQTTnet.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="net452" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"

Build/deb-meta/DEBIAN/control

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

Build/deb-meta/DEBIAN/copyright

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

Build/deb-meta/DEBIAN/postinst

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

Build/deb-meta/DEBIAN/postrm

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

Build/deb-meta/DEBIAN/prerm

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

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

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

CODE-OF-CONDUCT.md

@@ -0,0 +1,6 @@
+# Code of Conduct
+
+This project has adopted the code of conduct defined by the Contributor Covenant
+to clarify expected behavior in our community.
+
+For more information, see the [.NET Foundation Code of Conduct](https://dotnetfoundation.org/code-of-conduct).

MQTTnet.noUWP.sln

@@ -45,8 +45,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore.Tests",
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Tests", "Tests\MQTTnet.Core.Tests\MQTTnet.Tests.csproj", "{9C7106CA-96B8-4ABE-B3B4-9357AB8ACB41}"
 EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Server", "Source\MQTTnet.Server\MQTTnet.Server.csproj", "{DADEA836-5CC3-474C-A2D8-4D0F31118CD7}"
-EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.WebSocket4Net", "Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj", "{9AE2F8DA-4178-470B-B8C4-88BE1DAE7D5F}"
 EndProject
 Global

README.md

@@ -5,10 +5,9 @@
 </p>
 
 [![NuGet Badge](https://buildstats.info/nuget/MQTTnet)](https://www.nuget.org/packages/MQTTnet)
-[![Build status](https://ci.appveyor.com/api/projects/status/ycit86voxfevm2aa/branch/master?svg=true)](https://ci.appveyor.com/project/chkr1011/mqttnet)
 [![BCH compliance](https://bettercodehub.com/edge/badge/chkr1011/MQTTnet?branch=master)](https://bettercodehub.com/)
 [![Join the chat at https://gitter.im/MQTTnet/community](https://badges.gitter.im/MQTTnet/community.svg)](https://gitter.im/MQTTnet/community?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
-[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://raw.githubusercontent.com/chkr1011/MQTTnet/master/LICENSE)
+[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](https://raw.githubusercontent.com/chkr1011/MQTTnet/master/LICENSE)
 
 # MQTTnet
 

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -1,4 +1,4 @@
-using MQTTnet.Client;
+using MQTTnet.Client;
 using MQTTnet.Exceptions;
 using MQTTnet.Extensions.Rpc.Options;
 using MQTTnet.Extensions.Rpc.Options.TopicGeneration;
@@ -90,6 +90,7 @@ namespace MQTTnet.Extensions.Rpc
                 .WithTopic(requestTopic)
                 .WithPayload(payload)
                 .WithQualityOfServiceLevel(qualityOfServiceLevel)
+                .WithResponseTopic(responseTopic)
                 .Build();
 
             try
@@ -99,7 +100,7 @@ namespace MQTTnet.Extensions.Rpc
 #else
                 var awaitable = new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously);
 #endif
-                
+
                 if (!_waitingCalls.TryAdd(responseTopic, awaitable))
                 {
                     throw new InvalidOperationException();

Source/MQTTnet/Client/MqttClient.cs

@@ -455,7 +455,7 @@ namespace MQTTnet.Client
                 _logger.Verbose("Start sending keep alive packets.");
 
                 var keepAlivePeriod = Options.KeepAlivePeriod;
-
+     
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     // Values described here: [MQTT-3.1.2-24].
@@ -470,7 +470,7 @@ namespace MQTTnet.Client
                     // 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);
+                    await Task.Delay(250, cancellationToken).ConfigureAwait(false);
                 }
             }
             catch (Exception exception)

Source/MQTTnet/Diagnostics/Logger/IMqttNetLogger.cs

@@ -4,6 +4,8 @@ namespace MQTTnet.Diagnostics.Logger
 {
     public interface IMqttNetLogger
     {
+        bool IsEnabled { get; }
+        
         void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception);
     }
 }

Source/MQTTnet/Diagnostics/Logger/MqttNetEventLogger.cs

@@ -16,6 +16,8 @@ namespace MQTTnet.Diagnostics.Logger
 
         public string LogId { get; }
 
+        public bool IsEnabled { get; set; } = true;
+
         public void Publish(MqttNetLogLevel level, string source, string message, object[] parameters, Exception exception)
         {
             var eventHandler = LogMessagePublished;

Source/MQTTnet/Diagnostics/Logger/MqttNetNullLogger.cs

@@ -7,6 +7,8 @@ namespace MQTTnet.Diagnostics.Logger
     /// </summary>
     public sealed class MqttNetNullLogger : IMqttNetLogger
     {
+        public bool IsEnabled { get; }
+
         public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
         {
         }

Source/MQTTnet/Diagnostics/Logger/MqttNetSourceLogger.cs

@@ -13,6 +13,8 @@ namespace MQTTnet.Diagnostics.Logger
             _source = source;
         }
 
+        public bool IsEnabled => _logger.IsEnabled;
+        
         public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception)
         {
             _logger.Publish(logLevel, _source, message, parameters, exception);

Source/MQTTnet/Diagnostics/Logger/MqttNetSourceLoggerExtensions.cs

@@ -4,66 +4,176 @@ namespace MQTTnet.Diagnostics.Logger
 {
     public static class MqttNetSourceLoggerExtensions
     {
+        /*
+         * The logger uses generic parameters in order to avoid boxing of parameter values like integers etc.
+         */
+
         public static MqttNetSourceLogger WithSource(this IMqttNetLogger logger, string source)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
-            
+
             return new MqttNetSourceLogger(logger, source);
         }
-        
+
-        public static void Verbose(this MqttNetSourceLogger logger, string message, params object[] parameters)
+        public static void Verbose<TParameter1>(this MqttNetSourceLogger logger, string message, TParameter1 parameter1)
         {
-            logger.Publish(MqttNetLogLevel.Verbose, message, parameters, null);
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+
+            logger.Publish(MqttNetLogLevel.Verbose, message, new object[] {parameter1}, null);
         }
-        
+
+        public static void Verbose<TParameter1, TParameter2>(this MqttNetSourceLogger logger, string message, TParameter1 parameter1, TParameter2 parameter2)
+        {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+
+            logger.Publish(MqttNetLogLevel.Verbose, message, new object[] {parameter1, parameter2}, null);
+        }
+
+        public static void Verbose<TParameter1, TParameter2, TParameter3>(this MqttNetSourceLogger logger, string message, TParameter1 parameter1, TParameter2 parameter2,
+            TParameter3 parameter3)
+        {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+
+            logger.Publish(MqttNetLogLevel.Verbose, message, new object[] {parameter1, parameter2, parameter3}, null);
+        }
+
         public static void Verbose(this MqttNetSourceLogger logger, string message)
         {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+
             logger.Publish(MqttNetLogLevel.Verbose, message, null, null);
         }
 
-        public static void Info(this MqttNetSourceLogger logger, string message, params object[] parameters)
+        public static void Info<TParameter1>(this MqttNetSourceLogger logger, string message, TParameter1 parameter1)
         {
-            logger.Publish(MqttNetLogLevel.Info, message, parameters, null);
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+
+            logger.Publish(MqttNetLogLevel.Info, message, new object[] {parameter1}, null);
         }
         
+        public static void Info<TParameter1, TParameter2>(this MqttNetSourceLogger logger, string message, TParameter1 parameter1, TParameter2 parameter2)
+        {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+
+            logger.Publish(MqttNetLogLevel.Info, message, new object[] {parameter1, parameter2}, null);
+        }
+
         public static void Info(this MqttNetSourceLogger logger, string message)
         {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
             logger.Publish(MqttNetLogLevel.Info, message, null, null);
         }
 
-        public static void Warning(this MqttNetSourceLogger logger, Exception exception, string message, params object[] parameters)
+        public static void Warning<TParameter1>(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1)
         {
-            logger.Publish(MqttNetLogLevel.Warning, message, parameters, exception);
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
+            logger.Publish(MqttNetLogLevel.Warning, message, new object[] {parameter1}, exception);
+        }
+        
+        public static void Warning<TParameter1, TParameter2>(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1, TParameter2 parameter2)
+        {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
+            logger.Publish(MqttNetLogLevel.Warning, message, new object[] {parameter1, parameter2}, exception);
         }
         
         public static void Warning(this MqttNetSourceLogger logger, Exception exception, string message)
         {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
             logger.Publish(MqttNetLogLevel.Warning, message, null, exception);
         }
         
-        public static void Warning(this MqttNetSourceLogger logger, string message, params object[] parameters)
+        public static void Warning<TParameter1>(this MqttNetSourceLogger logger, string message, TParameter1 parameter1)
         {
-            logger.Publish(MqttNetLogLevel.Warning, message, parameters, null);
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
+            logger.Publish(MqttNetLogLevel.Warning, message, new object[] {parameter1}, null);
         }
         
         public static void Warning(this MqttNetSourceLogger logger, string message)
         {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
             logger.Publish(MqttNetLogLevel.Warning, message, null, null);
         }
 
-        public static void Error(this MqttNetSourceLogger logger, Exception exception, string message, params object[] parameters)
+        public static void Error<TParameter1>(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1)
         {
-            logger.Publish(MqttNetLogLevel.Error, message, parameters, exception);
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
+            logger.Publish(MqttNetLogLevel.Error, message, new object[] {parameter1}, exception);
         }
         
+        public static void Error<TParameter1, TParameter2>(this MqttNetSourceLogger logger, Exception exception, string message, TParameter1 parameter1, TParameter2 parameter2)
+        {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
+            logger.Publish(MqttNetLogLevel.Error, message, new object[] {parameter1, parameter2}, exception);
+        }
+
         public static void Error(this MqttNetSourceLogger logger, Exception exception, string message)
         {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
             logger.Publish(MqttNetLogLevel.Error, message, null, exception);
         }
-        
+
         public static void Error(this MqttNetSourceLogger logger, string message)
         {
+            if (!logger.IsEnabled)
+            {
+                return;
+            }
+            
             logger.Publish(MqttNetLogLevel.Error, message, null, null);
         }
     }
-}
+}

Source/MQTTnet/Exceptions/MqttCommunicationTimedOutException.cs

@@ -8,7 +8,7 @@ namespace MQTTnet.Exceptions
         {
         }
 
-        public MqttCommunicationTimedOutException(Exception innerException) : base(innerException)
+        public MqttCommunicationTimedOutException(Exception innerException) : base("The operation has timed out.", innerException)
         {
         }
     }

Source/MQTTnet/Implementations/CrossPlatformSocket.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.IO;
 using System.Net;
 using System.Net.Sockets;
@@ -11,12 +11,14 @@ namespace MQTTnet.Implementations
     public sealed class CrossPlatformSocket : IDisposable
     {
         readonly Socket _socket;
+        readonly Action _socketDisposeAction;
 
         NetworkStream _networkStream;
 
         public CrossPlatformSocket(AddressFamily addressFamily)
         {
             _socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp);
+            _socketDisposeAction = _socket.Dispose;
         }
 
         public CrossPlatformSocket()
@@ -24,12 +26,16 @@ namespace MQTTnet.Implementations
             // Having this constructor is important because avoiding the address family as parameter
             // will make use of dual mode in the .net framework.
             _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+
+            _socketDisposeAction = _socket.Dispose;
         }
 
         public CrossPlatformSocket(Socket socket)
         {
             _socket = socket ?? throw new ArgumentNullException(nameof(socket));
             _networkStream = new NetworkStream(socket, true);
+
+            _socketDisposeAction = _socket.Dispose;
         }
 
         public bool NoDelay
@@ -112,7 +118,7 @@ namespace MQTTnet.Implementations
                 _networkStream?.Dispose();
 
                 // Workaround for: https://github.com/dotnet/corefx/issues/24430
-                using (cancellationToken.Register(() => _socket.Dispose()))
+                using (cancellationToken.Register(_socketDisposeAction))
                 {
                     cancellationToken.ThrowIfCancellationRequested();
 

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -18,10 +18,16 @@ namespace MQTTnet.Implementations
     {
         readonly IMqttClientOptions _clientOptions;
         readonly MqttClientTcpOptions _tcpOptions;
+        readonly Action _disposeAction;
 
         Stream _stream;
 
-        public MqttTcpChannel(IMqttClientOptions clientOptions)
+        public MqttTcpChannel()
+        {
+            _disposeAction = Dispose;
+        }
+
+        public MqttTcpChannel(IMqttClientOptions clientOptions) : this()
         {
             _clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
             _tcpOptions = (MqttClientTcpOptions)clientOptions.ChannelOptions;
@@ -29,7 +35,7 @@ namespace MQTTnet.Implementations
             IsSecureConnection = clientOptions.ChannelOptions?.TlsOptions?.UseTls == true;
         }
 
-        public MqttTcpChannel(Stream stream, string endpoint, X509Certificate2 clientCertificate)
+        public MqttTcpChannel(Stream stream, string endpoint, X509Certificate2 clientCertificate) : this()
         {
             _stream = stream ?? throw new ArgumentNullException(nameof(stream));
 
@@ -149,11 +155,15 @@ namespace MQTTnet.Implementations
                     return 0;
                 }
 
+#if NETCOREAPP3_0_OR_GREATER || NETSTANDARD2_1_OR_GREATER
+                return await stream.ReadAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false);
+#else
                 // Workaround for: https://github.com/dotnet/corefx/issues/24430
-                using (cancellationToken.Register(Dispose))
+                using (cancellationToken.Register(_disposeAction))
                 {
                     return await stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
                 }
+#endif
             }
             catch (ObjectDisposedException)
             {
@@ -177,18 +187,22 @@ namespace MQTTnet.Implementations
 
             try
             {
-                // Workaround for: https://github.com/dotnet/corefx/issues/24430
+                var stream = _stream;
-                using (cancellationToken.Register(Dispose))
+
+                if (stream == null)
                 {
-                    var stream = _stream;
+                    throw new MqttCommunicationException("The TCP connection is closed.");
-
+                }
-                    if (stream == null)
-                    {
-                        throw new MqttCommunicationException("The TCP connection is closed.");
-                    }
 
+#if NETCOREAPP3_0_OR_GREATER || NETSTANDARD2_1_OR_GREATER
+                await stream.WriteAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false);
+#else
+                // Workaround for: https://github.com/dotnet/corefx/issues/24430
+                using (cancellationToken.Register(_disposeAction))
+                {
                     await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
                 }
+#endif
             }
             catch (ObjectDisposedException)
             {

Source/MQTTnet/MqttApplicationMessageBuilder.cs

@@ -159,7 +159,18 @@ namespace MQTTnet
             else
             {
                 _payload = new byte[length];
-                payload.Read(_payload, 0, _payload.Length);
+
+                var totalRead = 0;
+                do
+                {
+                    var bytesRead = payload.Read(_payload, totalRead, _payload.Length - totalRead);
+                    if (bytesRead == 0)
+                    {
+                        break;
+                    }
+                    totalRead += bytesRead;
+                }
+                while (totalRead < length);
             }
 
             return this;

Source/MQTTnet/Server/Internal/MqttClientConnection.cs

@@ -1,11 +1,10 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Client;
 using MQTTnet.Client.Disconnecting;
-using MQTTnet.Diagnostics;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
@@ -18,7 +17,7 @@ using MQTTnet.Server.Status;
 
 namespace MQTTnet.Server.Internal
 {
-    public sealed class MqttClientConnection : IDisposable
+    public sealed class MqttClientConnection
     {
         readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
@@ -107,11 +106,6 @@ namespace MQTTnet.Server.Internal
             Statistics.FillClientStatus(clientStatus);
         }
 
-        public void Dispose()
-        {
-            _cancellationToken.Dispose();
-        }
-
         public async Task RunAsync()
         {
             _logger.Info("Client '{0}': Session started.", ClientId);

Source/MQTTnet/Server/Internal/MqttClientSession.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Generic;
 using MQTTnet.Server.Status;
 
@@ -13,11 +13,13 @@ namespace MQTTnet.Server.Internal
             IDictionary<object, object> items,
             MqttServerEventDispatcher eventDispatcher,
             IMqttServerOptions serverOptions,
-            IMqttRetainedMessagesManager retainedMessagesManager)
+            IMqttRetainedMessagesManager retainedMessagesManager,
+            bool isPersistent
+            )
         {
             ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
             Items = items ?? throw new ArgumentNullException(nameof(items));
-
+            IsPersistent = isPersistent;
             SubscriptionsManager = new MqttClientSubscriptionsManager(this, serverOptions, eventDispatcher, retainedMessagesManager);
             ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);
         }
@@ -26,6 +28,11 @@ namespace MQTTnet.Server.Internal
 
         public bool IsCleanSession { get; set; } = true;
 
+        /// <summary>
+        /// Session should persist if CleanSession was set to false (Mqtt3) or if SessionExpiryInterval != 0 (Mqtt5)
+        /// </summary>
+        public bool IsPersistent { get; set; }
+
         public MqttApplicationMessage WillMessage { get; set; }
 
         public MqttClientSubscriptionsManager SubscriptionsManager { get; }

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -126,7 +126,7 @@ namespace MQTTnet.Server.Internal
 
                 // Pass connAckPacket so that IsSessionPresent flag can be set if the client session already exists
                 clientConnection = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter,
-                    connectionValidatorContext.SessionItems).ConfigureAwait(false);
+                    connectionValidatorContext).ConfigureAwait(false);
 
                 await channelAdapter.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
 
@@ -156,7 +156,7 @@ namespace MQTTnet.Server.Internal
                                 _clientConnections.Remove(clientConnection.ClientId);
                             }
 
-                            if (!_options.EnablePersistentSessions)
+                            if ((!_options.EnablePersistentSessions) || (!clientConnection.Session.IsPersistent))
                             {
                                 await DeleteSessionAsync(clientConnection.ClientId).ConfigureAwait(false);
                             }
@@ -177,8 +177,6 @@ namespace MQTTnet.Server.Internal
                     }
                 }
 
-                clientConnection?.Dispose();
-
                 await channelAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None)
                     .ConfigureAwait(false);
             }
@@ -493,10 +491,35 @@ namespace MQTTnet.Server.Internal
             MqttConnectPacket connectPacket,
             MqttConnAckPacket connAckPacket,
             IMqttChannelAdapter channelAdapter,
-            IDictionary<object, object> sessionItems)
+            MqttConnectionValidatorContext context)
         {
             MqttClientConnection connection;
 
+            bool sessionShouldPersist;
+
+            if (context.ProtocolVersion == MqttProtocolVersion.V500)
+            {
+                // MQTT 5.0 section 3.1.2.11.2
+                // The Client and Server MUST store the Session State after the Network Connection is closed if the Session Expiry Interval is greater than 0 [MQTT-3.1.2-23].
+                //
+                // A Client that only wants to process messages while connected will set the Clean Start to 1 and set the Session Expiry Interval to 0.
+                // It will not receive Application Messages published before it connected and has to subscribe afresh to any topics that it is interested
+                // in each time it connects.
+
+                // Persist if SessionExpiryInterval != 0, but may start with a clean session
+                sessionShouldPersist = context.SessionExpiryInterval.GetValueOrDefault() != 0;
+            }
+            else
+            {
+                // MQTT 3.1.1 section 3.1.2.4: persist only if 'not CleanSession'
+                //
+                // If CleanSession is set to 1, the Client and Server MUST discard any previous Session and start a new one.
+                // This Session lasts as long as the Network Connection. State data associated with this Session MUST NOT be
+                // reused in any subsequent Session [MQTT-3.1.2-6].
+
+                sessionShouldPersist = !connectPacket.CleanSession;
+            }
+
             using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false))
             {
                 MqttClientSession session;
@@ -505,18 +528,20 @@ namespace MQTTnet.Server.Internal
                     if (!_clientSessions.TryGetValue(connectPacket.ClientId, out session))
                     {
                         _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
-                        session = CreateSession(connectPacket.ClientId, sessionItems);
+                        session = CreateSession(connectPacket.ClientId, context.SessionItems, sessionShouldPersist);
                     }
                     else
                     {
                         if (connectPacket.CleanSession)
                         {
                             _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
-                            session = CreateSession(connectPacket.ClientId, sessionItems);
+                            session = CreateSession(connectPacket.ClientId, context.SessionItems, sessionShouldPersist);
                         }
                         else
                         {
                             _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
+                            // Session persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval
+                            session.IsPersistent = sessionShouldPersist;
                             connAckPacket.IsSessionPresent = true;
                         }
                     }
@@ -606,14 +631,16 @@ namespace MQTTnet.Server.Internal
                 _rootLogger);
         }
 
-        MqttClientSession CreateSession(string clientId, IDictionary<object, object> sessionItems)
+        MqttClientSession CreateSession(string clientId, IDictionary<object, object> sessionItems, bool isPersistent)
         {
             return new MqttClientSession(
                 clientId,
                 sessionItems,
                 _eventDispatcher,
                 _options,
-                _retainedMessagesManager);
+                _retainedMessagesManager,
+                isPersistent
+                );
         }
     }
 }

Tests/MQTTnet.Core.Tests/Logger/Logger_Tests.cs

@@ -35,7 +35,7 @@ namespace MQTTnet.Tests.Logger
 
             childLogger.Verbose("Verbose");
             childLogger.Info("Info");
-            childLogger.Warning(null, "Warning");
+            childLogger.Warning((Exception)null, "Warning");
             childLogger.Error(null, "Error");
 
             Assert.AreEqual(4, logMessagesCount);
@@ -55,7 +55,7 @@ namespace MQTTnet.Tests.Logger
             
             childLogger.Verbose("Verbose");
             childLogger.Info("Info");
-            childLogger.Warning(null, "Warning");
+            childLogger.Warning((Exception)null, "Warning");
             childLogger.Error(null, "Error");
         }
     }

Tests/MQTTnet.Core.Tests/Logger/SourceLogger_Tests.cs

@@ -18,7 +18,7 @@ namespace MQTTnet.Tests.Logger
             };
             
             var sourceLogger = logger.WithSource("The_Source");
-            sourceLogger.Info("MESSAGE", null, null);
+            sourceLogger.Info("MESSAGE", (object)null, (object)null);
          
             Assert.AreEqual("The_Source", logMessage.Source);
         }

Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs

@@ -51,15 +51,8 @@ namespace MQTTnet.Tests.MQTTv5
                 const string ClientId = "Client1";
 
                 // Create client without clean session and long session expiry interval
-
+                var options1 = CreateClientOptions(testEnvironment, ClientId, false, 9999);
-                var client1 = await testEnvironment.ConnectClient(o => o
+                var client1 = await testEnvironment.ConnectClient(options1);
-                    .WithProtocolVersion(MqttProtocolVersion.V500)
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
-                    .WithSessionExpiryInterval(9999)
-                    .WithCleanSession(false)
-                    .WithClientId(ClientId)
-                    .Build()
-                );
 
                 // Disconnect; empty session should remain on server
 
@@ -72,16 +65,8 @@ namespace MQTTnet.Tests.MQTTv5
                 // Reconnect the same client ID to existing session
 
                 var client2 = testEnvironment.CreateClient();
-                var options = testEnvironment.Factory.CreateClientOptionsBuilder()
+                var options2 = CreateClientOptions(testEnvironment, ClientId, false, 9999);
-                    .WithProtocolVersion(MqttProtocolVersion.V500)
+                var result = await client2.ConnectAsync(options2).ConfigureAwait(false);
-                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
-                    .WithSessionExpiryInterval(9999)
-                    .WithCleanSession(false)
-                    .WithClientId(ClientId)
-                    .Build();
-
-
-                var result = await client2.ConnectAsync(options).ConfigureAwait(false);
 
                 await client2.DisconnectAsync();
 
@@ -90,5 +75,111 @@ namespace MQTTnet.Tests.MQTTv5
                 Assert.IsTrue(result.IsSessionPresent, "Session not present");
             }
         }
+
+        [TestMethod]
+        public async Task Connect_with_Undefined_SessionExpiryInterval()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                // Create server with persistent sessions enabled
+
+                await testEnvironment.StartServer(o => o.WithPersistentSessions());
+
+                const string ClientId = "Client1";
+
+                // Create client without clean session and NO session expiry interval,
+                // that means, the session should not persist
+
+                var options1 = CreateClientOptions(testEnvironment, ClientId, false, null);
+                var client1 = await testEnvironment.ConnectClient(options1);
+
+                // Disconnect; no session should remain on server because the session expiry interval was undefined
+
+                await client1.DisconnectAsync();
+
+                // Simulate some time delay between connections
+
+                await Task.Delay(1000);
+
+                // Reconnect the same client ID to existing session
+
+                var client2 = testEnvironment.CreateClient();
+                var options2 = CreateClientOptions(testEnvironment, ClientId, false, 9999);
+                var result = await client2.ConnectAsync(options2).ConfigureAwait(false);
+
+                await client2.DisconnectAsync();
+
+                // Session should not be present
+
+                Assert.IsTrue(!result.IsSessionPresent, "Session is present when it should not");
+            }
+        }
+
+
+        [TestMethod]
+        public async Task Reconnect_with_different_SessionExpiryInterval()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                // Create server with persistent sessions enabled
+
+                await testEnvironment.StartServer(o => o.WithPersistentSessions());
+
+                const string ClientId = "Client1";
+
+                // Create client with clean session and session expiry interval > 0
+
+                var options = CreateClientOptions(testEnvironment, ClientId, true, 9999);
+                var client1 = await testEnvironment.ConnectClient(options);
+
+                // Disconnect; session should remain on server
+
+                await client1.DisconnectAsync();
+
+                // Simulate some time delay between connections
+
+                await Task.Delay(1000);
+
+                // Reconnect the same client ID to the existing session but leave session expiry interval undefined this time.
+                // Session should be present because the client1 connection had SessionExpiryInterval > 0
+
+                var client2 = testEnvironment.CreateClient();
+                var options2 = CreateClientOptions(testEnvironment, ClientId, false, null);
+                var result2 = await client2.ConnectAsync(options2).ConfigureAwait(false);
+
+                await client2.DisconnectAsync();
+
+                Assert.IsTrue(result2.IsSessionPresent, "Session is not present when it should");
+
+                // Simulate some time delay between connections
+
+                await Task.Delay(1000);
+
+                // Reconnect the same client ID.
+                // No session should be present because the previous session expiry interval was undefined for the client2 connection
+
+                var client3 = testEnvironment.CreateClient();
+                var options3 = CreateClientOptions(testEnvironment, ClientId, false, null);
+                var result3 = await client2.ConnectAsync(options3).ConfigureAwait(false);
+
+                await client3.DisconnectAsync();
+
+                // Session should be present
+
+                Assert.IsTrue(!result3.IsSessionPresent, "Session is present when it should not");
+
+            }
+        }
+
+        IMqttClientOptions CreateClientOptions(TestEnvironment testEnvironment, string clientId, bool cleanSession, uint? sessionExpiryInterval)
+        {
+            return testEnvironment.Factory.CreateClientOptionsBuilder()
+                .WithProtocolVersion(MqttProtocolVersion.V500)
+                .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                .WithSessionExpiryInterval(sessionExpiryInterval)
+                .WithCleanSession(cleanSession)
+                .WithClientId(clientId)
+                .Build();
+        }
     }
 }

Tests/MQTTnet.Core.Tests/Mockups/TestLogger.cs

@@ -6,7 +6,9 @@ namespace MQTTnet.Tests.Mockups
     public sealed class TestLogger : IMqttNetLogger
     {
         public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
-        
+
+        public bool IsEnabled { get; } = true;
+
         public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
         {
             LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(new MqttNetLogMessage

Tests/MQTTnet.Core.Tests/RPC_Tests.cs

@@ -56,6 +56,25 @@ namespace MQTTnet.Tests
             return Execute_Success(MqttQualityOfServiceLevel.ExactlyOnce, MqttProtocolVersion.V500);
         }
 
+        [TestMethod]
+        public Task Execute_Success_With_QoS_0_MQTT_V5_Use_ResponseTopic()
+        {
+            return Execute_Success_MQTT_V5(MqttQualityOfServiceLevel.AtMostOnce);
+        }
+
+        [TestMethod]
+        public Task Execute_Success_With_QoS_1_MQTT_V5_Use_ResponseTopic()
+        {
+            return Execute_Success_MQTT_V5(MqttQualityOfServiceLevel.AtLeastOnce);
+        }
+
+        [TestMethod]
+        public Task Execute_Success_With_QoS_2_MQTT_V5_Use_ResponseTopic()
+        {
+            return Execute_Success_MQTT_V5(MqttQualityOfServiceLevel.ExactlyOnce);
+        }
+
+
         [TestMethod]
         [ExpectedException(typeof(MqttCommunicationTimedOutException))]
         public async Task Execute_Timeout()
@@ -63,7 +82,7 @@ namespace MQTTnet.Tests
             using (var testEnvironment = new TestEnvironment(TestContext))
             {
                 await testEnvironment.StartServer();
-                
+
                 var requestSender = await testEnvironment.ConnectClient();
 
                 var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build());
@@ -81,8 +100,8 @@ namespace MQTTnet.Tests
 
                 var requestSender = await testEnvironment.ConnectClient();
 
-                var rpcClient = await testEnvironment.ConnectRpcClient(new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()) .Build());
+                var rpcClient = await testEnvironment.ConnectRpcClient(new MqttRpcClientOptionsBuilder().WithTopicGenerationStrategy(new TestTopicStrategy()).Build());
-                
+
                 await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
             }
         }
@@ -105,12 +124,86 @@ namespace MQTTnet.Tests
                 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));
                 }
             }
         }
 
+        async Task Execute_Success_MQTT_V5(MqttQualityOfServiceLevel qosLevel)
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServer();
+                var responseSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+                await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", qosLevel);
+
+                responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
+                {
+                    await responseSender.PublishAsync(e.ApplicationMessage.ResponseTopic, "pong");
+                });
+
+                var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+
+                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));
+                }
+            }
+        }
+
+        [TestMethod]
+        public async Task Execute_Success_MQTT_V5_Mixed_Clients()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServer();
+                var responseSender = await testEnvironment.ConnectClient();
+                await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", MqttQualityOfServiceLevel.AtMostOnce);
+
+                responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(async e =>
+                {
+                    Assert.IsNull(e.ApplicationMessage.ResponseTopic);
+                    await responseSender.PublishAsync(e.ApplicationMessage.Topic + "/response", "pong");
+                });
+
+                var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+
+                using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
+                {
+                    var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(5), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+
+                    Assert.AreEqual("pong", Encoding.UTF8.GetString(response));
+                }
+            }
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(MqttCommunicationTimedOutException))]
+        public async Task Execute_Timeout_MQTT_V5_Mixed_Clients()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServer();
+                var responseSender = await testEnvironment.ConnectClient();
+                await responseSender.SubscribeAsync("MQTTnet.RPC/+/ping", MqttQualityOfServiceLevel.AtMostOnce);
+
+                responseSender.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e =>
+                {
+                    Assert.IsNull(e.ApplicationMessage.ResponseTopic);
+                });
+
+                var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+
+                using (var rpcClient = new MqttRpcClient(requestSender, new MqttRpcClientOptionsBuilder().Build()))
+                {
+                    var response = await rpcClient.ExecuteAsync(TimeSpan.FromSeconds(2), "ping", "", MqttQualityOfServiceLevel.AtMostOnce);
+                }
+            }
+        }
+
         class TestTopicStrategy : IMqttRpcClientTopicGenerationStrategy
         {
             public MqttRpcTopicPair CreateRpcTopics(TopicGenerationContext context)

Tests/MQTTnet.Core.Tests/Server/General.cs

@@ -1167,7 +1167,7 @@ namespace MQTTnet.Tests.Server
                 var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithPersistentSessions());
 
                 // Create the session including the subscription.
-                var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a"));
+                var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false));
                 await client1.SubscribeAsync("x");
                 await client1.DisconnectAsync();
                 await Task.Delay(500);
@@ -1175,7 +1175,7 @@ namespace MQTTnet.Tests.Server
                 var clientStatus = await server.GetClientStatusAsync();
                 Assert.AreEqual(0, clientStatus.Count);
 
-                var client2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("b"));
+                var client2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("b").WithCleanSession(false));
                 await client2.PublishAsync("x", "1");
                 await client2.PublishAsync("x", "2");
                 await client2.PublishAsync("x", "3");

Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs

@@ -1,4 +1,4 @@
-using System.Collections.Concurrent;
+using System.Collections.Concurrent;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Packets;
@@ -107,7 +107,8 @@ namespace MQTTnet.Tests.Server
                 new ConcurrentDictionary<object, object>(),
                 new MqttServerEventDispatcher(new TestLogger()),
                 new MqttServerOptions(),
-                new MqttRetainedMessagesManager());
+                new MqttRetainedMessagesManager(),
+                false);
         }
     }
 }

Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs

@@ -1,4 +1,4 @@
-using System.Threading.Tasks;
+using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Adapter;
 using MQTTnet.Client;
@@ -34,28 +34,29 @@ namespace MQTTnet.Tests.Server
                 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);
+                Assert.AreEqual(MqttClientConnectResultCode.UnspecifiedError, ex.ResultCode);
             }
         }
-        
+
         [TestMethod]
         public Task Handle_Wrong_UserName()
         {
             return TestCredentials("x", "Password1");
         }
-        
+
         [TestMethod]
         public Task Handle_Wrong_Password()
         {
             return TestCredentials("UserName", "x");
         }
-        
+
         [TestMethod]
         public Task Handle_Wrong_UserName_And_Password()
         {
             return TestCredentials("x", "x");
         }
 
-        async Task TestCredentials(string userName, string password)
+        private async Task TestCredentials(string userName, string password)
         {
             using (var testEnvironment = CreateTestEnvironment())
             {
@@ -68,7 +69,7 @@ namespace MQTTnet.Tests.Server
                         {
                             c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                         }
-                        
+
                         if (c.Password != "Password1")
                         {
                             c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
@@ -83,7 +84,7 @@ namespace MQTTnet.Tests.Server
                     .Build();
 
                 var ex = await Assert.ThrowsExceptionAsync<MqttConnectingFailedException>(async () => await client.ConnectAsync(clientOptions));
-                Assert.AreEqual(ex.Result.ResultCode, MqttClientConnectResultCode.BadUserNameOrPassword);
+                Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, ex.Result.ResultCode);
             }
         }
     }

Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs

@@ -1,4 +1,4 @@
-using System.Linq;
+using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -107,6 +107,58 @@ namespace MQTTnet.Tests.Server
             }
         }
 
+        [TestMethod]
+        public async Task Clean_Session_Persistence()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                // Create server with persistent sessions enabled
+
+                await testEnvironment.StartServer(o => o.WithPersistentSessions());
+
+                const string ClientId = "Client1";
+
+                // Create client with clean session and long session expiry interval
+
+                var client1 = await testEnvironment.ConnectClient(o => o
+                    .WithProtocolVersion(Formatter.MqttProtocolVersion.V311)
+                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                    .WithSessionExpiryInterval(9999) // not relevant for v311 but testing impact
+                    .WithCleanSession(true) // start and end with clean session
+                    .WithClientId(ClientId)
+                    .Build()
+                );
+
+                // Disconnect; empty session should be removed from server
+
+                await client1.DisconnectAsync();
+
+                // Simulate some time delay between connections
+
+                await Task.Delay(1000);
+
+                // Reconnect the same client ID without clean session
+
+                var client2 = testEnvironment.CreateClient();
+                var options = testEnvironment.Factory.CreateClientOptionsBuilder()
+                    .WithProtocolVersion(Formatter.MqttProtocolVersion.V311)
+                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                    .WithSessionExpiryInterval(9999) // not relevant for v311 but testing impact
+                    .WithCleanSession(false) // see if there is a session
+                    .WithClientId(ClientId)
+                    .Build();
+
+
+                var result = await client2.ConnectAsync(options).ConfigureAwait(false);
+
+                await client2.DisconnectAsync();
+
+                // Session should NOT be present for MQTT v311 and initial CleanSession == true
+
+                Assert.IsTrue(!result.IsSessionPresent, "Session present");
+            }
+        }
+
         async Task<IMqttClient> TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
         {
             try

Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs

@@ -1,4 +1,4 @@
-using System.Linq;
+using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
@@ -81,8 +81,8 @@ namespace MQTTnet.Tests.Server
             {
                 var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithPersistentSessions());
 
-                var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("client1"));
+                var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("client1").WithCleanSession(false));
-                var c2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("client2"));
+                var c2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("client2").WithCleanSession(false));
 
                 await c1.DisconnectAsync();
 

Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs

@@ -94,12 +94,12 @@ namespace MQTTnet.TestApp.NetCore
                      .WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
             await ExecuteTestAsync("broker.emqx.io TCP TLS12",
-                new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8083)
+                new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883)
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls12).Build());
 
 #if NET5_0_OR_GREATER
             await ExecuteTestAsync("broker.emqx.io TCP TLS13",
-                new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8083)
+                new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883)
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTls(unsafeTls13).Build());
 #endif
 

_config.yml

@@ -1 +0,0 @@
-theme: jekyll-theme-hacker

appveyor.yml

@@ -1,15 +0,0 @@
-version: 1.0.{build}
-image: Visual Studio 2019
-configuration: Release
-before_build:
-- cmd: >-
-    msbuild /t:restore MQTTnet.sln
-
-    cd Tests/MQTTnet.TestApp.AspNetCore2/
-
-    npm install
-
-build:
-  project: MQTTnet.sln
-  verbosity: minimal
-