diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index d3fdeb50..2c200893 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -12,34 +12,14 @@ 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!). * [Core] Improved memory management when working with large payloads. -* [Client] Renamed _MqttClientConnectResult_ to _MqttClientConnectResult_ (BREAKING CHANGE!). * [Client] The OS will now choose the best TLS version to use. It is no longer fixed to 1.3 etc. (thanks to @patagonaa, #1271). * [Client] Exposed user properties and reason string in subscribe result. * [Client] Exposed user properties and reason string in unsubscribe result. -* [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). * [Server] Added support for returning individual subscription errors (#80 thanks to @jimch) * [Server] Improved topic filter comparisons (support for $). -* [Server] Fixed wrong "IsSessionPresent" result (thanks to @logicaloud, #1268). * [Server] Added more MQTTv5 response information to all interceptors (BREAKING CHANGE!). -* [MQTTnet.Server] Moved server project to a dedicated GitHub repository. +* [Server] Improved session management for MQTT v5 (#1294, thanks to @logicaloud). Git commit: $gitCommit Copyright Christian Kratky 2016-2021 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 diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 6f5d5f90..00b92d1b 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -96,11 +96,6 @@ namespace MQTTnet.Server ChannelAdapter.ResetStatistics(); } - public void Dispose() - { - _cancellationToken.Dispose(); - } - public async Task RunAsync() { _logger.Info("Client '{0}': Session started.", Id); diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index c722a45f..c4fe2c57 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -103,7 +103,7 @@ namespace MQTTnet.Server connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext); // Pass connAckPacket so that IsSessionPresent flag can be set if the client session already exists - client = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, connectionValidatorContext.SessionItems).ConfigureAwait(false); + client = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, connectionValidatorContext).ConfigureAwait(false); await client.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false); @@ -138,7 +138,7 @@ namespace MQTTnet.Server _clients.Remove(client.Id); } - if (!_options.EnablePersistentSessions) + if ((!_options.EnablePersistentSessions) || (!clientConnection.Session.IsPersistent)) { await DeleteSessionAsync(client.Id).ConfigureAwait(false); } @@ -422,11 +422,36 @@ namespace MQTTnet.Server async Task CreateClientConnection( MqttConnectPacket connectPacket, MqttConnAckPacket connAckPacket, - IMqttChannelAdapter channelAdapter, - IDictionary sessionItems) + IMqttChannelAdapter channelAdapter, + MqttConnectionValidatorContext context) { MqttClient 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)) { MqttSession session; @@ -434,14 +459,14 @@ namespace MQTTnet.Server { if (!_sessions.TryGetValue(connectPacket.ClientId, out session)) { - session = CreateSession(connectPacket.ClientId, sessionItems); + session = CreateSession(connectPacket.ClientId, sessionItems, 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 { @@ -548,7 +573,7 @@ namespace MQTTnet.Server _rootLogger); } - MqttSession CreateSession(string clientId, IDictionary sessionItems) + MqttClientSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent) { _logger.Verbose("Created a new session for client '{0}'.", clientId); diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs index c82755f8..82142b70 100644 --- a/Source/MQTTnet/Server/Internal/MqttSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttSession.cs @@ -20,23 +20,31 @@ namespace MQTTnet.Server readonly MqttClientSessionsManager _clientSessionsManager; public MqttSession(string clientId, + bool isPersistent, IDictionary items, MqttServerOptions serverOptions, MqttServerEventContainer eventContainer, MqttRetainedMessagesManager retainedMessagesManager, MqttClientSessionsManager clientSessionsManager) { + Id = clientId ?? throw new ArgumentNullException(nameof(clientId)); + IsPersistent = isPersistent; + Items = items ?? throw new ArgumentNullException(nameof(items)); + _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager)); - Id = clientId ?? throw new ArgumentNullException(nameof(clientId)); - Items = items ?? throw new ArgumentNullException(nameof(items)); - + SubscriptionsManager = new MqttClientSubscriptionsManager(this, serverOptions, eventContainer, retainedMessagesManager); } public event EventHandler Deleted; public string Id { get; } + + /// + /// Session should persist if CleanSession was set to false (Mqtt3) or if SessionExpiryInterval != 0 (Mqtt5) + /// + public bool IsPersistent { get; } public MqttPacketIdentifierProvider PacketIdentifierProvider { get; } = new MqttPacketIdentifierProvider(); diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index f597b184..05c8e3bb 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -1021,7 +1021,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); @@ -1029,7 +1029,7 @@ namespace MQTTnet.Tests.Server var clientStatus = await server.GetClientsAsync(); 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/Security_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs index 6e4ffa9b..3f534ea1 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; @@ -32,6 +32,7 @@ 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); } } @@ -53,7 +54,7 @@ namespace MQTTnet.Tests.Server return TestCredentials("x", "x"); } - async Task TestCredentials(string userName, string password) + private async Task TestCredentials(string userName, string password) { using (var testEnvironment = CreateTestEnvironment()) { @@ -84,7 +85,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); } } } diff --git a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs index 798890aa..5bdd3a86 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs @@ -1,4 +1,3 @@ -using System; using System.Linq; using System.Text; using System.Threading.Tasks; @@ -213,6 +212,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 05eac932..d8ed4233 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; @@ -80,8 +80,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();