From 85c3c0b352bae08c984a826eb5a9a6b2e02de8a9 Mon Sep 17 00:00:00 2001 From: SeppPenner Date: Mon, 8 Nov 2021 08:50:59 +0100 Subject: [PATCH 1/9] MQTTnet.noUWP.sln: Removed no longer present MQTTnet.Server project. --- MQTTnet.noUWP.sln | 2 -- 1 file changed, 2 deletions(-) diff --git a/MQTTnet.noUWP.sln b/MQTTnet.noUWP.sln index 3d842a5e..5fcfbb87 100644 --- a/MQTTnet.noUWP.sln +++ b/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 From 258df6bb1434e6fa98114da52e55a5d092e15ace Mon Sep 17 00:00:00 2001 From: HansM Date: Thu, 11 Nov 2021 20:10:35 +0100 Subject: [PATCH 2/9] Removed IDisposable from MqttClientConnection class. (#1288) --- Build/MQTTnet.nuspec | 21 +------------------ .../Server/Internal/MqttClientConnection.cs | 10 ++------- .../Internal/MqttClientSessionsManager.cs | 2 -- 3 files changed, 3 insertions(+), 30 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index df9fb118..b5003326 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -12,26 +12,7 @@ true 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. -* [Core] Added all builders to the MQTT factory. -* [Core] Removed global logger and refactored logging (BREAKING CHANGE!). -* [Client] Renamed _MqttClientConnectResult_ to _MqttClientConnectResult_ (BREAKING CHANGE!). -* [ManagedClient] Extended ReconnectAsync (thanks to @nvsnkv, #1202). -* [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. +* [Core] Removed IDisposable from MqttClientConnection.cs (Fixes #1284). Git commit: $gitCommit Copyright Christian Kratky 2016-2021 diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 9143c5c3..6081f273 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/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 _topicAlias = new Dictionary(); 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); diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 9793d868..8dccde9a 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -177,8 +177,6 @@ namespace MQTTnet.Server.Internal } } - clientConnection?.Dispose(); - await channelAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None) .ConfigureAwait(false); } From 47cd5aa3cdde0bad1487748a719e483cd5342a8c Mon Sep 17 00:00:00 2001 From: Volker Milbrandt Date: Sat, 13 Nov 2021 10:02:55 +0100 Subject: [PATCH 3/9] test ResultCode for invalid credentials (no userName but password) (#1270) fix order of parameters in exceptions AreEqual(expected, actual) --- Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs index 3cddcf12..bc1e3bd1 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs +++ b/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(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(async () => await client.ConnectAsync(clientOptions)); - Assert.AreEqual(ex.Result.ResultCode, MqttClientConnectResultCode.BadUserNameOrPassword); + Assert.AreEqual(MqttClientConnectResultCode.BadUserNameOrPassword, ex.Result.ResultCode); } } } From f0ea13c07f783ed6de401b0c7ba3de411b505b9e Mon Sep 17 00:00:00 2001 From: logicaloud <47704763+logicaloud@users.noreply.github.com> Date: Sat, 13 Nov 2021 22:05:54 +1300 Subject: [PATCH 4/9] Fix MQTT 3.1.1 session persistence when CleanSession=1 (#1294) --- .../Server/Internal/MqttClientSession.cs | 14 ++++- .../Internal/MqttClientSessionsManager.cs | 41 +++++++++++--- Tests/MQTTnet.Core.Tests/Server/General.cs | 4 +- .../Server/MqttSubscriptionsManager_Tests.cs | 5 +- .../Server/Session_Tests.cs | 54 ++++++++++++++++++- .../MQTTnet.Core.Tests/Server/Status_Tests.cs | 6 +-- 6 files changed, 107 insertions(+), 17 deletions(-) diff --git a/Source/MQTTnet/Server/Internal/MqttClientSession.cs b/Source/MQTTnet/Server/Internal/MqttClientSession.cs index 3f68fdfd..291b3e5a 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSession.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using MQTTnet.Server.Status; @@ -8,24 +8,34 @@ namespace MQTTnet.Server.Internal { readonly DateTime _createdTimestamp = DateTime.UtcNow; + /// + /// Session should persist if CleanSession was set to false (Mqtt3) or if SessionExpiryInterval != 0 (Mqtt5) + /// + readonly bool _isPersistent; + public MqttClientSession( string clientId, IDictionary 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)); SubscriptionsManager = new MqttClientSubscriptionsManager(this, serverOptions, eventDispatcher, retainedMessagesManager); ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions); + _isPersistent = isPersistent; } public string ClientId { get; } public bool IsCleanSession { get; set; } = true; + public bool IsPersistent => _isPersistent; + public MqttApplicationMessage WillMessage { get; set; } public MqttClientSubscriptionsManager SubscriptionsManager { get; } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 8dccde9a..8831a05b 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/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); } @@ -491,10 +491,35 @@ namespace MQTTnet.Server.Internal MqttConnectPacket connectPacket, MqttConnAckPacket connAckPacket, IMqttChannelAdapter channelAdapter, - IDictionary 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 != 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; @@ -503,14 +528,14 @@ 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 { @@ -604,14 +629,16 @@ namespace MQTTnet.Server.Internal _rootLogger); } - MqttClientSession CreateSession(string clientId, IDictionary sessionItems) + MqttClientSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent) { return new MqttClientSession( clientId, sessionItems, _eventDispatcher, _options, - _retainedMessagesManager); + _retainedMessagesManager, + isPersistent + ); } } } \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index f39b5215..7d0ba934 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/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"); diff --git a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs index 4ef18b4d..2a6f58f9 100644 --- a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs +++ b/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(), new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions(), - new MqttRetainedMessagesManager()); + new MqttRetainedMessagesManager(), + false); } } } diff --git a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs index e9000cdc..c502edff 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; @@ -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 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 c3f9f389..2995859e 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; @@ -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 c2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("client2")); + var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("client1").WithCleanSession(false)); + var c2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("client2").WithCleanSession(false)); await c1.DisconnectAsync(); From 2d94abdde371a3d0c50978f9bfcf30833cc7e5eb Mon Sep 17 00:00:00 2001 From: Christian Date: Sat, 13 Nov 2021 10:09:59 +0100 Subject: [PATCH 5/9] Update MQTTnet.nuspec Update docs. --- Build/MQTTnet.nuspec | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index b5003326..6a36380b 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -12,7 +12,8 @@ true 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. -* [Core] Removed IDisposable from MqttClientConnection.cs (Fixes #1284). +* [Core] Removed IDisposable from MqttClientConnection.cs (#1284). +* [Server] Improved session management for MQTT v5 (#1294, thanks to @logicaloud). Git commit: $gitCommit Copyright Christian Kratky 2016-2021 From 25ddfc83a8c8a5f459969bae901c43ebc792ac3d Mon Sep 17 00:00:00 2001 From: Michi Date: Sun, 28 Nov 2021 12:41:56 +0100 Subject: [PATCH 6/9] MqttRpcClient now adds ResponseTopic to MqttApplicationMessage when using MQTT v5 (#1295) * 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 (TaskCreationOptions.RunContinuationsAsynchronously); #endif - + if (!_waitingCalls.TryAdd(responseTopic, awaitable)) { throw new InvalidOperationException(); diff --git a/Tests/MQTTnet.Core.Tests/RPC_Tests.cs b/Tests/MQTTnet.Core.Tests/RPC_Tests.cs index 1d472b6d..09f55613 100644 --- a/Tests/MQTTnet.Core.Tests/RPC_Tests.cs +++ b/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) From 731c89d2e90b4f409daa54116fbce42ebd911ccd Mon Sep 17 00:00:00 2001 From: logicaloud <47704763+logicaloud@users.noreply.github.com> Date: Mon, 29 Nov 2021 00:51:29 +1300 Subject: [PATCH 7/9] Fix MQTT5 'IsPersisted' session flag handling (#1300) --- .../Server/Internal/MqttClientSession.cs | 13 +- .../Internal/MqttClientSessionsManager.cs | 4 +- .../MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs | 129 +++++++++++++++--- 3 files changed, 118 insertions(+), 28 deletions(-) diff --git a/Source/MQTTnet/Server/Internal/MqttClientSession.cs b/Source/MQTTnet/Server/Internal/MqttClientSession.cs index 291b3e5a..a943f5ad 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSession.cs @@ -8,11 +8,6 @@ namespace MQTTnet.Server.Internal { readonly DateTime _createdTimestamp = DateTime.UtcNow; - /// - /// Session should persist if CleanSession was set to false (Mqtt3) or if SessionExpiryInterval != 0 (Mqtt5) - /// - readonly bool _isPersistent; - public MqttClientSession( string clientId, IDictionary items, @@ -24,17 +19,19 @@ namespace MQTTnet.Server.Internal { 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); - _isPersistent = isPersistent; } public string ClientId { get; } public bool IsCleanSession { get; set; } = true; - public bool IsPersistent => _isPersistent; + /// + /// Session should persist if CleanSession was set to false (Mqtt3) or if SessionExpiryInterval != 0 (Mqtt5) + /// + public bool IsPersistent { get; set; } public MqttApplicationMessage WillMessage { get; set; } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 8831a05b..471e9df8 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -507,7 +507,7 @@ namespace MQTTnet.Server.Internal // in each time it connects. // Persist if SessionExpiryInterval != 0, but may start with a clean session - sessionShouldPersist = context.SessionExpiryInterval != 0; + sessionShouldPersist = context.SessionExpiryInterval.GetValueOrDefault() != 0; } else { @@ -540,6 +540,8 @@ namespace MQTTnet.Server.Internal 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; } } diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs index 18fde518..ce015b18 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs +++ b/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 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 @@ -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() - .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 +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(); + } } } From 39e637795ecb07a2aadf5ae541ea2a78100f3c8b Mon Sep 17 00:00:00 2001 From: patagona Date: Sun, 28 Nov 2021 12:56:32 +0100 Subject: [PATCH 8/9] improve exception message when handling connection errors (#1302) --- Source/MQTTnet/Exceptions/MqttCommunicationTimedOutException.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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) { } } From 72c7a8959c58c8d9f2c8c937988fd8c8352052fe Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 28 Nov 2021 12:57:22 +0100 Subject: [PATCH 9/9] Update MQTTnet.nuspec --- Build/MQTTnet.nuspec | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 6a36380b..a698c06c 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -13,7 +13,10 @@ 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. * [Core] Removed IDisposable from MqttClientConnection.cs (#1284). -* [Server] Improved session management for MQTT v5 (#1294, thanks to @logicaloud). +* [Core] Improved message of timeout exception (#1302, thanks to @patagonaa). +* [RpcClient] Filling response topic when using MQTTv5. (#1295, thanks to @MD-V). +* [Server] Fixed 'SessionIsPresent' handling for MQTTv5 (#1300, thanks to @logicaloud). +* [Server] Improved session management for MQTTv5 (#1294, thanks to @logicaloud). Git commit: $gitCommit Copyright Christian Kratky 2016-2021