From d9501de252da2aa52e822c0892041a81f08af016 Mon Sep 17 00:00:00 2001 From: Christian Date: Thu, 3 May 2018 20:29:53 +0200 Subject: [PATCH] Treat a connection handover as a clean disconnect. --- .../Server/MqttClientDisconnectType.cs | 8 +++++ .../Server/MqttClientPendingMessagesQueue.cs | 2 +- .../Server/MqttClientSession.cs | 33 ++++++++++--------- .../Server/MqttClientSessionsManager.cs | 6 ++-- 4 files changed, 30 insertions(+), 19 deletions(-) create mode 100644 Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs new file mode 100644 index 00000000..19e9da8d --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectType.cs @@ -0,0 +1,8 @@ +namespace MQTTnet.Server +{ + public enum MqttClientDisconnectType + { + Clean, + NotClean + } +} diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs index 99493874..1602a4d1 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs @@ -128,7 +128,7 @@ namespace MQTTnet.Server if (!cancellationToken.IsCancellationRequested) { - await _clientSession.StopAsync().ConfigureAwait(false); + await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs index 8514da99..f86f72a1 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs @@ -90,7 +90,7 @@ namespace MQTTnet.Server return _wasCleanDisconnect; } - public async Task StopAsync(bool wasCleanDisconnect = false) + public async Task StopAsync(MqttClientDisconnectType type) { try { @@ -99,7 +99,7 @@ namespace MQTTnet.Server return; } - _wasCleanDisconnect = wasCleanDisconnect; + _wasCleanDisconnect = type == MqttClientDisconnectType.Clean; _cancellationTokenSource?.Cancel(false); @@ -118,7 +118,7 @@ namespace MQTTnet.Server var willMessage = _willMessage; _willMessage = null; // clear willmessage so it is send just once - if (willMessage != null && !wasCleanDisconnect) + if (willMessage != null && !_wasCleanDisconnect) { await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false); } @@ -181,7 +181,7 @@ namespace MQTTnet.Server private Task StopDueToKeepAliveTimeoutAsync() { _logger.Info("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId); - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken) @@ -198,15 +198,18 @@ namespace MQTTnet.Server catch (OperationCanceledException) { } - catch (MqttCommunicationException exception) - { - _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); - await StopAsync().ConfigureAwait(false); - } catch (Exception exception) { - _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); - await StopAsync().ConfigureAwait(false); + if (exception is MqttCommunicationException) + { + _logger.Warning(exception, "Client '{0}': Communication exception while processing client packets.", ClientId); + } + else + { + _logger.Error(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); + } + + await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } } @@ -255,16 +258,16 @@ namespace MQTTnet.Server if (packet is MqttDisconnectPacket) { - return StopAsync(true); + return StopAsync(MqttClientDisconnectType.Clean); } if (packet is MqttConnectPacket) { - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } _logger.Warning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); - return StopAsync(); + return StopAsync(MqttClientDisconnectType.NotClean); } private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) @@ -274,7 +277,7 @@ namespace MQTTnet.Server if (subscribeResult.CloseConnection) { - await StopAsync().ConfigureAwait(false); + await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false); diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index 8dd9d062..5f6cf15e 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -124,7 +124,7 @@ namespace MQTTnet.Server { foreach (var session in _sessions) { - await session.Value.StopAsync().ConfigureAwait(false); + await session.Value.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } _sessions.Clear(); @@ -162,7 +162,7 @@ namespace MQTTnet.Server var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage); if (interceptorContext.CloseConnection) { - await senderClientSession.StopAsync().ConfigureAwait(false); + await senderClientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); } if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) @@ -283,7 +283,7 @@ namespace MQTTnet.Server { _sessions.Remove(connectPacket.ClientId); - await clientSession.StopAsync().ConfigureAwait(false); + await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false); clientSession.Dispose(); clientSession = null;