fixed blocking code in networkstrem and reduced sleep in tests

This commit is contained in:
JanEggers
2018-07-19 21:13:28 +02:00
parent b55f5ff298
commit 53bd2a0960
2 changed files with 24 additions and 12 deletions
+22 -10
View File
@@ -30,6 +30,7 @@ namespace MQTTnet.Server
private bool _wasCleanDisconnect; private bool _wasCleanDisconnect;
private IMqttChannelAdapter _adapter; private IMqttChannelAdapter _adapter;
private Task<bool> _run; private Task<bool> _run;
private IDisposable _cleanupHandle;
public MqttClientSession( public MqttClientSession(
string clientId, string clientId,
@@ -84,6 +85,10 @@ namespace MQTTnet.Server
adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted; adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted;
_cancellationTokenSource = new CancellationTokenSource(); _cancellationTokenSource = new CancellationTokenSource();
//woraround for https://github.com/dotnet/corefx/issues/24430
_cleanupHandle = _cancellationTokenSource.Token.Register(() => Cleanup());
//endworkaround
_wasCleanDisconnect = false; _wasCleanDisconnect = false;
_willMessage = connectPacket.WillMessage; _willMessage = connectPacket.WillMessage;
@@ -129,20 +134,14 @@ namespace MQTTnet.Server
{ {
_adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; _adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
_adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; _adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
await Cleanup();
try
{
await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
_adapter.Dispose();
}
catch (Exception exception)
{
_logger.Error(exception, exception.Message);
}
} }
_adapter = null; _adapter = null;
_cleanupHandle?.Dispose();
_cleanupHandle = null;
_cancellationTokenSource?.Dispose(); _cancellationTokenSource?.Dispose();
_cancellationTokenSource = null; _cancellationTokenSource = null;
} }
@@ -150,6 +149,19 @@ namespace MQTTnet.Server
return _wasCleanDisconnect; return _wasCleanDisconnect;
} }
private async Task Cleanup()
{
try
{
await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
_adapter.Dispose();
}
catch (Exception exception)
{
_logger.Error(exception, exception.Message);
}
}
public void Stop(MqttClientDisconnectType type) public void Stop(MqttClientDisconnectType type)
{ {
try try
+2 -2
View File
@@ -253,11 +253,11 @@ namespace MQTTnet.Core.Tests
await c1.ConnectAsync(clientOptions); await c1.ConnectAsync(clientOptions);
await Task.Delay(500); await Task.Delay(100);
await s.StopAsync(); await s.StopAsync();
await Task.Delay(1000); await Task.Delay(100);
Assert.IsTrue(disconnectCalled); Assert.IsTrue(disconnectCalled);
} }