diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs index cdf83291..3c24170a 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs +++ b/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; @@ -74,6 +74,7 @@ namespace MQTTnet.Extensions.Rpc .WithTopic(requestTopic) .WithPayload(payload) .WithQualityOfServiceLevel(qualityOfServiceLevel) + .WithResponseTopic(responseTopic) .Build(); try @@ -83,7 +84,7 @@ namespace MQTTnet.Extensions.Rpc #else var awaitable = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); #endif - + if (!_waitingCalls.TryAdd(responseTopic, awaitable)) { throw new InvalidOperationException(); diff --git a/Source/MQTTnet/Exceptions/MqttCommunicationTimedOutException.cs b/Source/MQTTnet/Exceptions/MqttCommunicationTimedOutException.cs index df3c4255..38ee182e 100644 --- a/Source/MQTTnet/Exceptions/MqttCommunicationTimedOutException.cs +++ b/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) { } } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 88a198ba..01fe66aa 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -431,6 +431,31 @@ namespace MQTTnet.Server sessionShouldPersist = !connectPacket.CleanSession; } + 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)) { MqttSession session; @@ -450,6 +475,8 @@ namespace MQTTnet.Server 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; session.Recover(); } diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs index c9485900..75b2087c 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs @@ -52,15 +52,8 @@ namespace MQTTnet.Tests.MQTTv5 const string ClientId = "Client1"; // Create client without clean session and long session expiry interval - - var client1 = await testEnvironment.ConnectClient(o => o - .WithProtocolVersion(MqttProtocolVersion.V500) - .WithTcpServer("127.0.0.1", testEnvironment.ServerPort) - .WithSessionExpiryInterval(9999) - .WithCleanSession(false) - .WithClientId(ClientId) - .Build() - ); + var options1 = CreateClientOptions(testEnvironment, ClientId, false, 9999); + var client1 = await testEnvironment.ConnectClient(options1); // Disconnect; empty session should remain on server @@ -73,15 +66,8 @@ namespace MQTTnet.Tests.MQTTv5 // Reconnect the same client ID to existing session var client2 = testEnvironment.CreateClient(); - var options = testEnvironment.Factory.CreateClientOptionsBuilder() - .WithProtocolVersion(MqttProtocolVersion.V500) - .WithTcpServer("127.0.0.1", testEnvironment.ServerPort) - .WithSessionExpiryInterval(9999) - .WithCleanSession(false) - .WithClientId(ClientId) - .Build(); - - var result = await client2.ConnectAsync(options).ConfigureAwait(false); + var options2 = CreateClientOptions(testEnvironment, ClientId, false, 9999); + var result = await client2.ConnectAsync(options2).ConfigureAwait(false); await client2.DisconnectAsync(); @@ -90,5 +76,112 @@ 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(); + + var result = await client2.ConnectAsync(options).ConfigureAwait(false); + // 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(); + } } } diff --git a/Tests/MQTTnet.Core.Tests/RPC_Tests.cs b/Tests/MQTTnet.Core.Tests/RPC_Tests.cs index 2a14fb09..d4ecadf0 100644 --- a/Tests/MQTTnet.Core.Tests/RPC_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/RPC_Tests.cs @@ -54,6 +54,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() @@ -61,7 +80,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()); @@ -79,8 +98,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); } } @@ -100,12 +119,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) diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index a930c62e..e8ca0b0e 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -941,6 +941,12 @@ namespace MQTTnet.Tests.Server var client = testEnvironment.CreateClient(); var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build()); + + // Create the session including the subscription. + var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false)); + await client1.SubscribeAsync("x"); + await client1.DisconnectAsync(); + await Task.Delay(500); Assert.IsFalse(connectResult.IsSessionPresent); } @@ -952,6 +958,11 @@ namespace MQTTnet.Tests.Server using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); + 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"); + await client2.DisconnectAsync(); var client = testEnvironment.CreateClient(); diff --git a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs index ca07aff0..5c21e0c2 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs +++ b/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; @@ -263,6 +263,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 TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) { try diff --git a/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs index ba6f918a..a94714a2 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs +++ b/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;