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

* MqttRpcClient now adds ResponseTopic to MqttApplicationMessage when using MQTT v5

* No need for conditional protocol check (ResponseTopic property won't be added by PacketFormatter in <MQTT v5)

* Cleanup: Remove using

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.
+* [Core] Removed IDisposable from MqttClientConnection.cs (#1284).
-* [Core] Removed global logger and refactored logging (BREAKING CHANGE!).
+* [Core] Improved message of timeout exception (#1302, thanks to @patagonaa).
-* [Client] Renamed _MqttClientConnectResult_ to _MqttClientConnectResult_ (BREAKING CHANGE!).
+* [RpcClient] Filling response topic when using MQTTv5. (#1295, thanks to @MD-V).
-* [ManagedClient] Extended ReconnectAsync (thanks to @nvsnkv, #1202).
+* [Server] Fixed 'SessionIsPresent' handling for MQTTv5 (#1300, thanks to @logicaloud).
-* [ManagedClient] Improved Amazon AWS support (thanks to @scottbrogden-iheartmedia, #1209). 
+* [Server] Improved session management for MQTTv5 (#1294, thanks to @logicaloud).
-* [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>

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

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/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/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/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/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(async 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();