diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 400045b4..8bc800ea 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -62,7 +62,7 @@ namespace MQTTnet.Server public Task StartSession(IMqttChannelAdapter clientAdapter) { - return Task.Run(() => RunSession(clientAdapter, _cancellationToken), _cancellationToken); + return Task.Run(() => RunSessionAsync(clientAdapter, _cancellationToken), _cancellationToken); } public IList GetClientStatus() @@ -205,7 +205,7 @@ namespace MQTTnet.Server } } - private async Task RunSession(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken) + private async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken) { var clientId = string.Empty; @@ -241,7 +241,8 @@ namespace MQTTnet.Server } var result = PrepareClientSession(connectPacket); - var clientSession = result.Session; + + Server.OnClientConnected(clientId); await clientAdapter.SendPacketAsync( new MqttConnAckPacket @@ -251,9 +252,7 @@ namespace MQTTnet.Server }, cancellationToken).ConfigureAwait(false); - Server.OnClientConnected(clientId); - - await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false); + await result.Session.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false); } catch (OperationCanceledException) { diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 4836f4de..2a48e993 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -660,16 +660,22 @@ namespace MQTTnet.Core.Tests { var s = new MqttFactory().CreateMqttServer(); - var events = new ConcurrentBag(); + var events = new List(); s.ClientConnected += (_, __) => { - events.Add("c"); + lock (events) + { + events.Add("c"); + } }; s.ClientDisconnected += (_, __) => { - events.Add("d"); + lock (events) + { + events.Add("d"); + } }; var clientOptions = new MqttClientOptionsBuilder() @@ -691,17 +697,17 @@ namespace MQTTnet.Core.Tests await Task.Delay(250); await c1.DisconnectAsync(); + + await Task.Delay(250); + await c2.DisconnectAsync(); await Task.Delay(250); - var flow = string.Join(string.Empty, events); - await s.StopAsync(); + var flow = string.Join(string.Empty, events); Assert.AreEqual("cdcd", flow); - - //Assert.IsFalse(connecteCalledBeforeConnectedClients, "ClientConnected was called before ClientDisconnect was called"); }