YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-26 16:24:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

[Server] Add initial support for session expiration (#1776)

Browse Source
This commit is contained in:
Christian
2023-06-27 19:32:08 +02:00
committed by GitHub
parent 18614d4e81
commit 9295626503
15 changed files with 224 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+10 -2
View File
@@ -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);
Source/MQTTnet/Internal/MqttTaskTimeout.cs
-38
View File
@@ -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<CancellationToken, Task> 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;
}
}
}
}
}
Source/MQTTnet/Server/Internal/MqttClient.cs
+2
View File
@@ -103,6 +103,8 @@ namespace MQTTnet.Server
finally
{
IsRunning = false;
Session.DisconnectedTimestamp = DateTime.UtcNow;
_cancellationToken?.TryCancel();
_cancellationToken?.Dispose();
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+27 -33
View File
@@ -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<string, MqttSession> _sessions = new Dictionary<string, MqttSession>(4096);
readonly MqttSessionsStorage _sessionsStorage = new MqttSessionsStorage();
readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>();
@@ -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<DispatchApplicationMessageResult> 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.");
}
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+1 -4
View File
@@ -162,10 +162,7 @@ namespace MQTTnet.Server
public void Dispose()
{
if (_subscriptionsLock != null)
{
_subscriptionsLock.Dispose();
}
_subscriptionsLock?.Dispose();
}
public async Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
Source/MQTTnet/Server/Internal/MqttSession.cs
+8 -5
View File
@@ -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<string> _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;
/// <summary>
/// Session should persist if CleanSession was set to false (Mqtt3) or if SessionExpiryInterval != 0 (Mqtt5)
Source/MQTTnet/Server/Internal/MqttSessionsStorage.cs
+93
View File
@@ -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<string, MqttSession> _sessions = new Dictionary<string, MqttSession>(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<MqttSession> ReadAllSessions()
{
return new ReadOnlyCollection<MqttSession>(_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;
}
}
}
Source/MQTTnet/Server/Status/MqttSessionStatus.cs
+4
View File
@@ -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;