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

Fix Session_Tests.Manage_Session_MaxParallel

Browse Source
This commit is contained in:
SilverFox
2021-04-02 17:54:41 +08:00
parent 9b10eadb24
commit f99daee4d1
1 changed files with 14 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+14 -10
View File
@@ -1,4 +1,4 @@
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Formatter; using MQTTnet.Formatter;
@@ -21,7 +21,7 @@ namespace MQTTnet.Server
{ {
readonly BlockingCollection<MqttPendingApplicationMessage> _messageQueue = new BlockingCollection<MqttPendingApplicationMessage>(); readonly BlockingCollection<MqttPendingApplicationMessage> _messageQueue = new BlockingCollection<MqttPendingApplicationMessage>();
readonly object _createConnectionSyncRoot = new object(); readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
readonly Dictionary<string, MqttClientConnection> _connections = new Dictionary<string, MqttClientConnection>(); readonly Dictionary<string, MqttClientConnection> _connections = new Dictionary<string, MqttClientConnection>();
readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>(); readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
@@ -98,7 +98,7 @@ namespace MQTTnet.Server
return; return;
} }
var connection = CreateClientConnection(connectPacket, connectionValidatorContext, channelAdapter); var connection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);
await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket.ClientId).ConfigureAwait(false); await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket.ClientId).ConfigureAwait(false);
await connection.RunAsync().ConfigureAwait(false); await connection.RunAsync().ConfigureAwait(false);
} }
@@ -387,9 +387,12 @@ namespace MQTTnet.Server
return context; return context;
} }
MqttClientConnection CreateClientConnection(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter) async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
{ {
lock (_createConnectionSyncRoot) MqttClientConnection existingConnection;
MqttClientConnection connection;
using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false))
{ {
MqttClientSession session; MqttClientSession session;
lock (_sessions) lock (_sessions)
@@ -415,8 +418,6 @@ namespace MQTTnet.Server
_sessions[connectPacket.ClientId] = session; _sessions[connectPacket.ClientId] = session;
} }
MqttClientConnection existingConnection;
MqttClientConnection connection;
lock (_connections) lock (_connections)
{ {
_connections.TryGetValue(connectPacket.ClientId, out existingConnection); _connections.TryGetValue(connectPacket.ClientId, out existingConnection);
@@ -425,10 +426,13 @@ namespace MQTTnet.Server
_connections[connectPacket.ClientId] = connection; _connections[connectPacket.ClientId] = connection;
} }
existingConnection?.StopAsync(MqttClientDisconnectReason.SessionTakenOver).GetAwaiter().GetResult(); if (existingConnection != null)
{
return connection; await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false);
}
} }
return connection;
} }
async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(IMqttServerApplicationMessageInterceptor interceptor, MqttClientConnection clientConnection, MqttApplicationMessage applicationMessage) async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(IMqttServerApplicationMessageInterceptor interceptor, MqttClientConnection clientConnection, MqttApplicationMessage applicationMessage)