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

Also cleanup MqttClientSessionsManager._subscriberSessions when session is taken over by a new one with CleanSession=true, fix #1552 (#1553)

Browse Source
This commit is contained in:
SilverFox
2022-11-06 10:35:15 +01:00
committed by GitHub
parent 2842f27522
commit 5b1cc7d713
2 changed files with 9 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Tests/Internal/AsyncSignal_Tests.cs
+1 -1
View File
@@ -90,7 +90,7 @@ namespace MQTTnet.Tests.Internal
Assert.IsTrue(stopwatch.ElapsedMilliseconds > 900); Assert.IsTrue(stopwatch.ElapsedMilliseconds > 900);
} }
} }
[TestMethod] [TestMethod]
[ExpectedException(typeof(InvalidOperationException))] [ExpectedException(typeof(InvalidOperationException))]
public async Task Fail_For_Two_Waiters() public async Task Fail_For_Two_Waiters()
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+8 -2
View File
@@ -485,9 +485,10 @@ namespace MQTTnet.Server
using (await _createConnectionSyncRoot.EnterAsync().ConfigureAwait(false)) using (await _createConnectionSyncRoot.EnterAsync().ConfigureAwait(false))
{ {
MqttSession session; MqttSession session;
MqttSession oldSession;
lock (_sessionsManagementLock) lock (_sessionsManagementLock)
{ {
if (!_sessions.TryGetValue(connectPacket.ClientId, out session)) if (!_sessions.TryGetValue(connectPacket.ClientId, out oldSession))
{ {
session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist); session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
} }
@@ -496,11 +497,14 @@ namespace MQTTnet.Server
if (connectPacket.CleanSession) 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.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
} }
else 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 persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval
session.IsPersistent = sessionShouldPersist; session.IsPersistent = sessionShouldPersist;
connAckPacket.IsSessionPresent = true; connAckPacket.IsSessionPresent = true;
@@ -533,7 +537,7 @@ namespace MQTTnet.Server
existingClient.IsTakenOver = true; existingClient.IsTakenOver = true;
await existingClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false); await existingClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
if (_eventContainer.ClientConnectedEvent.HasHandlers) if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
{ {
var eventArgs = new ClientDisconnectedEventArgs( var eventArgs = new ClientDisconnectedEventArgs(
existingClient.Id, existingClient.Id,
@@ -544,6 +548,8 @@ namespace MQTTnet.Server
await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
} }
} }
oldSession?.Dispose();
} }
return client; return client;