diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index c45abf78..9e17c1aa 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,3 +1,4 @@ * [Client] Fixed _PlatformNotSupportedException_ when using Blazor (#1755, thanks to @Nickztar). * [Server] Fixed _NullReferenceException_ in retained messages management (#1762, thanks to @logicaloud). * [Server] Exposed new option which allows disabling packet fragmentation (#1753). +* [Server] Expired sessions will no longer be used when a client connects (#1756). diff --git a/Source/MQTTnet.Tests/Extension_Tests.cs b/Source/MQTTnet.Tests/Extension_Tests.cs deleted file mode 100644 index fa72dbc3..00000000 --- a/Source/MQTTnet.Tests/Extension_Tests.cs +++ /dev/null @@ -1,98 +0,0 @@ -// Licensed to the .NET Foundation under one or more agreements. -// The .NET Foundation licenses this file to you under the MIT license. -// See the LICENSE file in the project root for more information. - -using System; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using Microsoft.VisualStudio.TestTools.UnitTesting; -using MQTTnet.Exceptions; -using MQTTnet.Internal; - -namespace MQTTnet.Tests -{ - [TestClass] - public class Extension_Tests - { - [ExpectedException(typeof(MqttCommunicationTimedOutException))] - [TestMethod] - public async Task TimeoutAfter() - { - await MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct), - TimeSpan.FromMilliseconds(100), CancellationToken.None); - } - - [ExpectedException(typeof(MqttCommunicationTimedOutException))] - [TestMethod] - public async Task TimeoutAfterWithResult() - { - await MqttTaskTimeout.WaitAsync( - ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct).ContinueWith(t => 5, ct), - TimeSpan.FromMilliseconds(100), CancellationToken.None); - } - - [TestMethod] - public async Task TimeoutAfterWithInnerException() - { - try - { - await MqttTaskTimeout.WaitAsync(ct => Task.Run(() => - { - var iis = new int[0]; - iis[1] = 0; - }, ct), TimeSpan.FromSeconds(1), CancellationToken.None); - - Assert.Fail(); - } - catch (Exception e) - { - Assert.IsTrue(e is IndexOutOfRangeException); - } - } - - [TestMethod] - public async Task TimeoutAfterWithInnerExceptionWithResult() - { - try - { - await MqttTaskTimeout.WaitAsync(ct => Task.Run(() => - { - var iis = new int[0]; - iis[1] = 0; - return iis[0]; - }, ct), TimeSpan.FromSeconds(1), CancellationToken.None); - - Assert.Fail(); - } - catch (Exception e) - { - Assert.IsTrue(e is IndexOutOfRangeException); - } - } - - [TestMethod] - public async Task TimeoutAfterMemoryUsage() - { - var initialMemory = GC.GetTotalMemory(true); - - var tasks = Enumerable.Range(0, 100000) - .Select(i => - { - return MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(1), ct), - TimeSpan.FromMinutes(1), CancellationToken.None); - }); - - await Task.WhenAll(tasks); - AssertIsLess(initialMemory + 2_000_000, GC.GetTotalMemory(true)); - } - - static void AssertIsLess(long bound, long actual) - { - if (bound < actual) - { - Assert.Fail($"value must be less than {bound:N0} but is {actual:N0}"); - } - } - } -} \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs index b3543706..92f2543f 100644 --- a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs +++ b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs @@ -3,6 +3,7 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Linq; @@ -24,10 +25,9 @@ namespace MQTTnet.Tests.Mockups public sealed class TestEnvironment : IDisposable { readonly List _clientErrors = new List(); - readonly List _clients = new List(); - + readonly ConcurrentBag _clients = new ConcurrentBag(); readonly List _exceptions = new List(); - readonly List _lowLevelClients = new List(); + readonly ConcurrentBag _lowLevelClients = new ConcurrentBag(); readonly MqttProtocolVersion _protocolVersion; readonly List _serverErrors = new List(); @@ -65,9 +65,12 @@ namespace MQTTnet.Tests.Mockups if (e.LogMessage.Level == MqttNetLogLevel.Error) { - lock (_clientErrors) + if (!IgnoreClientLogErrors) { - _clientErrors.Add(e.LogMessage.ToString()); + lock (_clientErrors) + { + _clientErrors.Add(e.LogMessage.ToString()); + } } } }; @@ -212,30 +215,27 @@ namespace MQTTnet.Tests.Mockups public IMqttClient CreateClient() { - lock (_clients) + var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : MqttNetNullLogger.Instance; + + var client = Factory.CreateMqttClient(logger); + client.ConnectingAsync += e => { - var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : new MqttNetNullLogger(); - - var client = Factory.CreateMqttClient(logger); - _clients.Add(client); - - client.ConnectingAsync += e => + if (TestContext != null) { - if (TestContext != null) + var clientOptions = e.ClientOptions; + var existingClientId = clientOptions.ClientId; + if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName)) { - var clientOptions = e.ClientOptions; - var existingClientId = clientOptions.ClientId; - if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName)) - { - clientOptions.ClientId = TestContext.TestName + "_" + existingClientId; - } + clientOptions.ClientId = TestContext.TestName + "_" + existingClientId; } + } - return CompletedTask.Instance; - }; - - return client; - } + return CompletedTask.Instance; + }; + + _clients.Add(client); + + return client; } public MqttClientOptions CreateDefaultClientOptions() @@ -250,13 +250,10 @@ namespace MQTTnet.Tests.Mockups public ILowLevelMqttClient CreateLowLevelClient() { - lock (_clients) - { - var client = Factory.CreateLowLevelMqttClient(ClientLogger); - _lowLevelClients.Add(client); + var client = Factory.CreateLowLevelMqttClient(ClientLogger); + _lowLevelClients.Add(client); - return client; - } + return client; } public MqttServer CreateServer(MqttServerOptions options) @@ -383,13 +380,16 @@ namespace MQTTnet.Tests.Mockups } } - lock (_clientErrors) + if (!IgnoreClientLogErrors) { - if (!IgnoreClientLogErrors && _clientErrors.Count > 0) + lock (_clientErrors) { - var message = $"Client(s) had {_clientErrors.Count} errors (${string.Join(Environment.NewLine, _clientErrors)})"; - Console.WriteLine(message); - throw new Exception(message); + if (_clientErrors.Count > 0) + { + var message = $"Client(s) had {_clientErrors.Count} errors (${string.Join(Environment.NewLine, _clientErrors)})"; + Console.WriteLine(message); + throw new Exception(message); + } } } } diff --git a/Source/MQTTnet.Tests/Server/General.cs b/Source/MQTTnet.Tests/Server/General.cs index 2e0061e1..20bc013c 100644 --- a/Source/MQTTnet.Tests/Server/General.cs +++ b/Source/MQTTnet.Tests/Server/General.cs @@ -598,8 +598,7 @@ namespace MQTTnet.Tests.Server // c var c1 = await testEnvironment.ConnectClient(clientOptionsBuilder); - await Task.Delay(500); - + await LongTestDelay(); var flow = string.Join(string.Empty, events); Assert.AreEqual("c", flow); @@ -607,10 +606,8 @@ namespace MQTTnet.Tests.Server // Connect client with same client ID. Should disconnect existing client. var c2 = await testEnvironment.ConnectClient(clientOptionsBuilder); - await Task.Delay(500); - + await LongTestDelay(); flow = string.Join(string.Empty, events); - Assert.AreEqual("cdc", flow); c2.ApplicationMessageReceivedAsync += e => @@ -628,8 +625,7 @@ namespace MQTTnet.Tests.Server // r await c2.PublishStringAsync("topic"); - await Task.Delay(500); - + await LongTestDelay(); flow = string.Join(string.Empty, events); Assert.AreEqual("cdcr", flow); @@ -639,16 +635,17 @@ namespace MQTTnet.Tests.Server await c1.TryDisconnectAsync(); Assert.AreEqual(false, c1.IsConnected); - await Task.Delay(500); + await LongTestDelay(); // d Assert.AreEqual(true, c2.IsConnected); await c2.DisconnectAsync(); - await Task.Delay(500); + await LongTestDelay(); await server.StopAsync(); + await LongTestDelay(); flow = string.Join(string.Empty, events); Assert.AreEqual("cdcrd", flow); } @@ -852,11 +849,11 @@ namespace MQTTnet.Tests.Server return CompletedTask.Instance; }; - await Task.Delay(100); - + // Stopping the server should disconnect the connection with the client + // which, it turn, will fire the disconnected event at the client. await server.StopAsync(); - await Task.Delay(100); + await LongTestDelay(); Assert.AreEqual(1, disconnectCalled); } diff --git a/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs b/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs index ec4aef90..233cd1f4 100644 --- a/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs +++ b/Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs @@ -221,7 +221,10 @@ namespace MQTTnet.Tests.Server var eventContainer = new MqttServerEventContainer(); var clientSessionManager = new MqttClientSessionsManager(options, retainedMessagesManager, eventContainer, logger); - var session = new MqttSession("", false, new ConcurrentDictionary(), options, eventContainer, retainedMessagesManager, clientSessionManager); + var session = new MqttSession(new MqttConnectPacket + { + ClientId = "" + }, new ConcurrentDictionary(), options, eventContainer, retainedMessagesManager, clientSessionManager); _subscriptionsManager = new MqttClientSubscriptionsManager(session, new MqttServerEventContainer(), retainedMessagesManager, clientSessionManager); } diff --git a/Source/MQTTnet.Tests/Server/Session_Tests.cs b/Source/MQTTnet.Tests/Server/Session_Tests.cs index b8c5527f..5a6ab80c 100644 --- a/Source/MQTTnet.Tests/Server/Session_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Session_Tests.cs @@ -70,6 +70,28 @@ namespace MQTTnet.Tests.Server } } + [TestMethod] + public async Task Do_Not_Use_Expired_Session() + { + using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500)) + { + await testEnvironment.StartServer(o => o.WithPersistentSessions()); + + var c1 = await testEnvironment.ConnectClient(o => o.WithClientId("Client1").WithCleanSession(false).WithSessionExpiryInterval(3)); + + // Kill the client connection and ensure that the session will stay there but is expired after 3 seconds! + c1.Dispose(); + + await Task.Delay(TimeSpan.FromSeconds(6)); + + c1 = testEnvironment.CreateClient(); + var options = testEnvironment.CreateDefaultClientOptionsBuilder().WithClientId("Client1").WithCleanSession(false).WithSessionExpiryInterval(3).Build(); + + var connectResult = await c1.ConnectAsync(options); + Assert.AreEqual(false, connectResult.IsSessionPresent); + } + } + [TestMethod] public async Task Fire_Deleted_Event() { @@ -121,7 +143,7 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(true, session.Items["can_subscribe_x"]); } } - + [TestMethod] [DataRow(MqttProtocolVersion.V310)] [DataRow(MqttProtocolVersion.V311)] @@ -134,16 +156,12 @@ namespace MQTTnet.Tests.Server await testEnvironment.StartServer(); - var options = new MqttClientOptionsBuilder().WithClientId("1").WithTimeout(TimeSpan.FromSeconds(1)).WithProtocolVersion(protocolVersion); + var options = new MqttClientOptionsBuilder().WithClientId("1").WithTimeout(TimeSpan.FromSeconds(10)).WithProtocolVersion(protocolVersion); var hasReceive = false; - void OnReceive() { - if (!hasReceive) - { - hasReceive = true; - } + hasReceive = true; } // Try to connect 50 clients at the same time. @@ -152,7 +170,7 @@ namespace MQTTnet.Tests.Server var connectedClients = clients.Where(c => c?.TryPingAsync().GetAwaiter().GetResult() == true).ToList(); await LongTestDelay(); - + Assert.AreEqual(1, connectedClients.Count); var option2 = new MqttClientOptionsBuilder().WithClientId("2").WithKeepAlivePeriod(TimeSpan.FromSeconds(10)); @@ -164,7 +182,7 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(true, hasReceive); } } - + [DataTestMethod] [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)] [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)] @@ -355,8 +373,8 @@ namespace MQTTnet.Tests.Server onReceive(); return CompletedTask.Instance; }; - - using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5))) + + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30))) { await client.SubscribeAsync("aaa", MqttQualityOfServiceLevel.AtMostOnce, timeout.Token).ConfigureAwait(false); } diff --git a/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs b/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs index 2dd9fe5f..24df145d 100644 --- a/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs @@ -557,7 +557,7 @@ namespace MQTTnet.Tests.Server var eventContainer = new MqttServerEventContainer(); var retainedMessagesManager = new MqttRetainedMessagesManager(eventContainer, logger); var sessionManager = new MqttClientSessionsManager(serverOptions, retainedMessagesManager, eventContainer, logger); - _clientSession = new MqttSession(clientId, false, new Dictionary(), serverOptions, eventContainer, retainedMessagesManager, sessionManager); + _clientSession = new MqttSession(new MqttConnectPacket{ ClientId = clientId }, new Dictionary(), serverOptions, eventContainer, retainedMessagesManager, sessionManager); List topicsToSubscribe; diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 12c28202..2b851ea6 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -442,10 +442,18 @@ namespace MQTTnet.Client { result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion); } - else + else if (receivedPacket is MqttAuthPacket) { throw new NotSupportedException("Extended authentication handler is not yet supported"); } + else if (receivedPacket == null) + { + throw new MqttCommunicationException("Connection closed."); + } + else + { + throw new InvalidOperationException($"Received an unexpected MQTT packet ({receivedPacket})."); + } } catch (Exception exception) { @@ -547,7 +555,7 @@ namespace MQTTnet.Client try { - _packetDispatcher.Dispose(new MqttClientDisconnectedException(exception)); + _packetDispatcher?.Dispose(new MqttClientDisconnectedException(exception)); var receiverTask = _packetReceiverTask.WaitAsync(sender, _logger); var publishPacketReceiverTask = _publishPacketReceiverTask.WaitAsync(sender, _logger); diff --git a/Source/MQTTnet/Internal/MqttTaskTimeout.cs b/Source/MQTTnet/Internal/MqttTaskTimeout.cs deleted file mode 100644 index abd05c36..00000000 --- a/Source/MQTTnet/Internal/MqttTaskTimeout.cs +++ /dev/null @@ -1,38 +0,0 @@ -// Licensed to the .NET Foundation under one or more agreements. -// The .NET Foundation licenses this file to you under the MIT license. -// See the LICENSE file in the project root for more information. - -using System; -using System.Threading; -using System.Threading.Tasks; -using MQTTnet.Exceptions; - -namespace MQTTnet.Internal -{ - public static class MqttTaskTimeout - { - public static async Task WaitAsync(Func action, TimeSpan timeout, CancellationToken cancellationToken) - { - if (action == null) throw new ArgumentNullException(nameof(action)); - - using (var timeoutCts = new CancellationTokenSource(timeout)) - using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken)) - { - try - { - await action(linkedCts.Token).ConfigureAwait(false); - } - catch (OperationCanceledException exception) - { - var timeoutReached = timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested; - if (timeoutReached) - { - throw new MqttCommunicationTimedOutException(exception); - } - - throw; - } - } - } - } -} diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 53b49fdc..2cfde887 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -103,6 +103,8 @@ namespace MQTTnet.Server finally { IsRunning = false; + + Session.DisconnectedTimestamp = DateTime.UtcNow; _cancellationToken?.TryCancel(); _cancellationToken?.Dispose(); diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index c0d385d6..22d9e28e 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -34,7 +34,7 @@ namespace MQTTnet.Server // The _sessions dictionary contains all session, the _subscriberSessions hash set contains subscriber sessions only. // See the MqttSubscription object for a detailed explanation. - readonly Dictionary _sessions = new Dictionary(4096); + readonly MqttSessionsStorage _sessionsStorage = new MqttSessionsStorage(); readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim(); readonly HashSet _subscriberSessions = new HashSet(); @@ -78,21 +78,16 @@ namespace MQTTnet.Server _logger.Verbose("Deleting session for client '{0}'.", clientId); MqttClient connection; - lock (_clients) { _clients.TryGetValue(clientId, out connection); } MqttSession session; - _sessionsManagementLock.EnterWriteLock(); try { - _sessions.TryGetValue(clientId, out session); - _sessions.Remove(clientId); - - if (session != null) + if (_sessionsStorage.TryRemoveSession(clientId, out session)) { _subscriberSessions.Remove(session); } @@ -111,7 +106,7 @@ namespace MQTTnet.Server } catch (Exception exception) { - _logger.Error(exception, $"Error while deleting session '{clientId}'."); + _logger.Error(exception, "Error while deleting session '{0}'", clientId); } try @@ -124,12 +119,12 @@ namespace MQTTnet.Server } catch (Exception exception) { - _logger.Error(exception, $"Error while executing session deleted event for session '{clientId}'."); + _logger.Error(exception, "Error while executing session deleted event for session '{0}'", clientId); } session?.Dispose(); - _logger.Verbose("Session for client '{0}' deleted.", clientId); + _logger.Verbose("Session of client '{0}' deleted", clientId); } public async Task DispatchApplicationMessage( @@ -269,20 +264,14 @@ namespace MQTTnet.Server _sessionsManagementLock.EnterWriteLock(); try { - foreach (var sessionItem in _sessions) - { - sessionItem.Value.Dispose(); - } + _sessionsStorage.Dispose(); } finally { _sessionsManagementLock.ExitWriteLock(); } - if (_sessionsManagementLock != null) - { - _sessionsManagementLock.Dispose(); - } + _sessionsManagementLock?.Dispose(); } public MqttClient GetClient(string id) @@ -333,9 +322,9 @@ namespace MQTTnet.Server _sessionsManagementLock.EnterReadLock(); try { - foreach (var sessionItem in _sessions) + foreach (var session in _sessionsStorage.ReadAllSessions()) { - var sessionStatus = new MqttSessionStatus(sessionItem.Value); + var sessionStatus = new MqttSessionStatus(session); result.Add(sessionStatus); } } @@ -484,7 +473,7 @@ namespace MQTTnet.Server { if (!_options.EnablePersistentSessions) { - _sessions.Clear(); + _sessionsStorage.Clear(); } } @@ -584,33 +573,35 @@ namespace MQTTnet.Server MqttSession session; // Create a new session (if required). - if (!_sessions.TryGetValue(connectPacket.ClientId, out oldSession)) + if (!_sessionsStorage.TryGetSession(connectPacket.ClientId, out oldSession)) { - session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist); + session = CreateSession(connectPacket, validatingConnectionEventArgs, sessionShouldPersist); } else { if (connectPacket.CleanSession) { - _logger.Verbose("Deleting existing session of client '{0}' due to clean start.", connectPacket.ClientId); + _logger.Verbose("Deleting existing session of client '{0}' due to clean start", connectPacket.ClientId); _subscriberSessions.Remove(oldSession); - session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist); + session = CreateSession(connectPacket, validatingConnectionEventArgs, sessionShouldPersist); } else { - _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId); + _logger.Verbose("Reusing existing session of client '{0}'", connectPacket.ClientId); session = oldSession; oldSession = null; // Session persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval session.IsPersistent = sessionShouldPersist; - connAckPacket.IsSessionPresent = true; + session.DisconnectedTimestamp = null; session.Recover(); + + connAckPacket.IsSessionPresent = true; } } - _sessions[connectPacket.ClientId] = session; - + _sessionsStorage.UpdateSession(connectPacket.ClientId, session); + // Create a new client (always required). lock (_clients) { @@ -656,11 +647,14 @@ namespace MQTTnet.Server return client; } - MqttSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent) + MqttSession CreateSession(MqttConnectPacket connectPacket, ValidatingConnectionEventArgs validatingConnectionEventArgs, bool isPersistent) { - _logger.Verbose("Created new session for client '{0}'.", clientId); + _logger.Verbose("Created new session for client '{0}'", connectPacket.ClientId); - return new MqttSession(clientId, isPersistent, sessionItems, _options, _eventContainer, _retainedMessagesManager, this); + return new MqttSession(connectPacket, validatingConnectionEventArgs.SessionItems, _options, _eventContainer, _retainedMessagesManager, this) + { + IsPersistent = isPersistent + }; } async Task FireApplicationMessageNotConsumedEvent(MqttApplicationMessage applicationMessage, string senderId) @@ -679,7 +673,7 @@ namespace MQTTnet.Server _sessionsManagementLock.EnterReadLock(); try { - if (!_sessions.TryGetValue(clientId, out var session)) + if (!_sessionsStorage.TryGetSession(clientId, out var session)) { throw new InvalidOperationException($"Client session '{clientId}' is unknown."); } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 037d6489..8cacf560 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -162,10 +162,7 @@ namespace MQTTnet.Server public void Dispose() { - if (_subscriptionsLock != null) - { - _subscriptionsLock.Dispose(); - } + _subscriptionsLock?.Dispose(); } public async Task Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs index b40c2645..e16b6534 100644 --- a/Source/MQTTnet/Server/Internal/MqttSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttSession.cs @@ -21,6 +21,7 @@ namespace MQTTnet.Server readonly MqttPacketBus _packetBus = new MqttPacketBus(); readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); + readonly MqttConnectPacket _connectPacket; readonly MqttServerOptions _serverOptions; readonly MqttClientSubscriptionsManager _subscriptionsManager; @@ -31,18 +32,16 @@ namespace MQTTnet.Server HashSet _subscribedTopics; public MqttSession( - string clientId, - bool isPersistent, + MqttConnectPacket connectPacket, 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)); + _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket)); _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager)); @@ -50,10 +49,14 @@ namespace MQTTnet.Server } public DateTime CreatedTimestamp { get; } = DateTime.UtcNow; + + public DateTime? DisconnectedTimestamp { get; set; } public bool HasSubscribedTopics => _subscribedTopics != null && _subscribedTopics.Count > 0; - public string Id { get; } + public uint ExpiryInterval => _connectPacket.SessionExpiryInterval; + + public string Id => _connectPacket.ClientId; /// /// Session should persist if CleanSession was set to false (Mqtt3) or if SessionExpiryInterval != 0 (Mqtt5) diff --git a/Source/MQTTnet/Server/Internal/MqttSessionsStorage.cs b/Source/MQTTnet/Server/Internal/MqttSessionsStorage.cs new file mode 100644 index 00000000..bbb10e43 --- /dev/null +++ b/Source/MQTTnet/Server/Internal/MqttSessionsStorage.cs @@ -0,0 +1,93 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections.Generic; +using System.Collections.ObjectModel; +using System.Linq; + +namespace MQTTnet.Server +{ + public sealed class MqttSessionsStorage + { + readonly Dictionary _sessions = new Dictionary(4096); + + public void Clear() + { + // Make sure that the sessions are also getting disposed! + // This object can be reused after disposing. + Dispose(); + } + + public void Dispose() + { + foreach (var sessionItem in _sessions) + { + sessionItem.Value.Dispose(); + } + + _sessions.Clear(); + } + + public IReadOnlyCollection ReadAllSessions() + { + return new ReadOnlyCollection(_sessions.Values.ToList()); + } + + public bool TryGetSession(string id, out MqttSession session) + { + if (id == null) + { + throw new ArgumentNullException(nameof(id)); + } + + if (!_sessions.TryGetValue(id, out session)) + { + return false; + } + + // If the Session Expiry Interval is 0xFFFFFFFF (UINT_MAX), the Session does not expire. + if (session.ExpiryInterval > 0 && session.ExpiryInterval != uint.MaxValue) + { + if (session.DisconnectedTimestamp.HasValue) + { + if (DateTime.UtcNow > session.DisconnectedTimestamp.Value.AddSeconds(session.ExpiryInterval)) + { + TryRemoveSession(id, out var oldSession); + oldSession?.Dispose(); + return false; + } + } + } + + return true; + } + + public bool TryRemoveSession(string id, out MqttSession session) + { + if (id == null) + { + throw new ArgumentNullException(nameof(id)); + } + + if (_sessions.TryGetValue(id, out session)) + { + _sessions.Remove(id); + return true; + } + + return false; + } + + public void UpdateSession(string id, MqttSession session) + { + if (id == null) + { + throw new ArgumentNullException(nameof(id)); + } + + _sessions[id] = session; + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index 3a4b193b..9ba6692f 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -21,6 +21,10 @@ namespace MQTTnet.Server public DateTime CreatedTimestamp => _session.CreatedTimestamp; + public DateTime? DisconnectedTimestamp => _session.DisconnectedTimestamp; + + public uint ExpiryInterval => _session.ExpiryInterval; + public string Id => _session.Id; public IDictionary Items => _session.Items;