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

fixed deadlock and object disposed exception

Browse Source
This commit is contained in:
JanEggers
2018-07-23 17:58:29 +02:00
parent d30fe61264
commit 043cfdb3f9
2 changed files with 67 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/MqttClientSession.cs
+27 -16
View File
@@ -126,18 +126,11 @@ namespace MQTTnet.Server
_logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId);
}
Stop(MqttClientDisconnectType.NotClean);
Stop(MqttClientDisconnectType.NotClean, true);
}
finally
{
if (_adapter != null)
{
_adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
_adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
await Cleanup().ConfigureAwait(false);
}
_adapter = null;
await Cleanup().ConfigureAwait(false);
_cleanupHandle?.Dispose();
_cleanupHandle = null;
@@ -153,8 +146,18 @@ namespace MQTTnet.Server
{
try
{
await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
_adapter.Dispose();
var adapter = _adapter;
if (adapter == null)
{
return;
}
_adapter = null;
adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
adapter.Dispose();
}
catch (Exception exception)
{
@@ -163,6 +166,11 @@ namespace MQTTnet.Server
}
public void Stop(MqttClientDisconnectType type)
{
Stop(type, false);
}
private void Stop(MqttClientDisconnectType type, bool isInsideSession)
{
try
{
@@ -183,7 +191,10 @@ namespace MQTTnet.Server
_willMessage = null;
_workerTask?.GetAwaiter().GetResult();
if (!isInsideSession)
{
_workerTask?.GetAwaiter().GetResult();
}
}
finally
{
@@ -329,18 +340,18 @@ namespace MQTTnet.Server
if (packet is MqttDisconnectPacket)
{
Stop(MqttClientDisconnectType.Clean);
Stop(MqttClientDisconnectType.Clean, true);
return;
}
if (packet is MqttConnectPacket)
{
Stop(MqttClientDisconnectType.NotClean);
Stop(MqttClientDisconnectType.NotClean, true);
return;
}
_logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
Stop(MqttClientDisconnectType.NotClean);
Stop(MqttClientDisconnectType.NotClean, true);
}
private void EnqueueSubscribedRetainedMessages(ICollection<TopicFilter> topicFilters)
@@ -359,7 +370,7 @@ namespace MQTTnet.Server
if (subscribeResult.CloseConnection)
{
Stop(MqttClientDisconnectType.NotClean);
Stop(MqttClientDisconnectType.NotClean, true);
return;
}
Tests/MQTTnet.Core.Tests/MqttServerTests.cs
+40
View File
@@ -262,6 +262,46 @@ namespace MQTTnet.Core.Tests
Assert.AreEqual(1, disconnectCalled);
}
[TestMethod]
public async Task MqttServer_HandleCleanDisconnect()
{
MqttNetGlobalLogger.LogMessagePublished += (_, e) =>
{
System.Diagnostics.Debug.WriteLine($"[{e.TraceMessage.Timestamp:s}] {e.TraceMessage.Source} {e.TraceMessage.Message}");
};
var serverAdapter = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger());
var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
var clientConnectedCalled = 0;
var clientDisconnectedCalled = 0;
s.ClientConnected += (_, __) => clientConnectedCalled++;
s.ClientDisconnected += (_, __) => clientDisconnectedCalled++;
var clientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("localhost")
.Build();
await s.StartAsync(new MqttServerOptions());
var c1 = new MqttFactory().CreateMqttClient();
await c1.ConnectAsync(clientOptions);
await Task.Delay(100);
await c1.DisconnectAsync();
await Task.Delay(100);
await s.StopAsync();
await Task.Delay(100);
Assert.AreEqual(clientConnectedCalled, clientDisconnectedCalled);
}
[TestMethod]
public async Task MqttServer_RetainedMessagesFlow()
{