Treat a connection handover as a clean disconnect.

This commit is contained in:
Christian
2018-05-03 20:29:53 +02:00
parent 9fdcb08364
commit d9501de252
4 changed files with 30 additions and 19 deletions
@@ -0,0 +1,8 @@
namespace MQTTnet.Server
{
public enum MqttClientDisconnectType
{
Clean,
NotClean
}
}
@@ -128,7 +128,7 @@ namespace MQTTnet.Server
if (!cancellationToken.IsCancellationRequested) if (!cancellationToken.IsCancellationRequested)
{ {
await _clientSession.StopAsync().ConfigureAwait(false); await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
} }
} }
} }
@@ -90,7 +90,7 @@ namespace MQTTnet.Server
return _wasCleanDisconnect; return _wasCleanDisconnect;
} }
public async Task StopAsync(bool wasCleanDisconnect = false) public async Task StopAsync(MqttClientDisconnectType type)
{ {
try try
{ {
@@ -99,7 +99,7 @@ namespace MQTTnet.Server
return; return;
} }
_wasCleanDisconnect = wasCleanDisconnect; _wasCleanDisconnect = type == MqttClientDisconnectType.Clean;
_cancellationTokenSource?.Cancel(false); _cancellationTokenSource?.Cancel(false);
@@ -118,7 +118,7 @@ namespace MQTTnet.Server
var willMessage = _willMessage; var willMessage = _willMessage;
_willMessage = null; // clear willmessage so it is send just once _willMessage = null; // clear willmessage so it is send just once
if (willMessage != null && !wasCleanDisconnect) if (willMessage != null && !_wasCleanDisconnect)
{ {
await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false); await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
} }
@@ -181,7 +181,7 @@ namespace MQTTnet.Server
private Task StopDueToKeepAliveTimeoutAsync() private Task StopDueToKeepAliveTimeoutAsync()
{ {
_logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId); _logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId);
return StopAsync(); return StopAsync(MqttClientDisconnectType.NotClean);
} }
private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken) private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -198,15 +198,18 @@ namespace MQTTnet.Server
catch (OperationCanceledException) catch (OperationCanceledException)
{ {
} }
catch (MqttCommunicationException exception)
{
_logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
await StopAsync().ConfigureAwait(false);
}
catch (Exception exception) catch (Exception exception)
{ {
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); if (exception is MqttCommunicationException)
await StopAsync().ConfigureAwait(false); {
_logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
}
else
{
_logger.Error<MqttClientSession>(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) if (packet is MqttDisconnectPacket)
{ {
return StopAsync(true); return StopAsync(MqttClientDisconnectType.Clean);
} }
if (packet is MqttConnectPacket) if (packet is MqttConnectPacket)
{ {
return StopAsync(); return StopAsync(MqttClientDisconnectType.NotClean);
} }
_logger.Warning<MqttClientSession>("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); _logger.Warning<MqttClientSession>("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) private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
@@ -274,7 +277,7 @@ namespace MQTTnet.Server
if (subscribeResult.CloseConnection) if (subscribeResult.CloseConnection)
{ {
await StopAsync().ConfigureAwait(false); await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
} }
await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false); await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
@@ -124,7 +124,7 @@ namespace MQTTnet.Server
{ {
foreach (var session in _sessions) foreach (var session in _sessions)
{ {
await session.Value.StopAsync().ConfigureAwait(false); await session.Value.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
} }
_sessions.Clear(); _sessions.Clear();
@@ -162,7 +162,7 @@ namespace MQTTnet.Server
var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage); var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage);
if (interceptorContext.CloseConnection) if (interceptorContext.CloseConnection)
{ {
await senderClientSession.StopAsync().ConfigureAwait(false); await senderClientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
} }
if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
@@ -283,7 +283,7 @@ namespace MQTTnet.Server
{ {
_sessions.Remove(connectPacket.ClientId); _sessions.Remove(connectPacket.ClientId);
await clientSession.StopAsync().ConfigureAwait(false); await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false);
clientSession.Dispose(); clientSession.Dispose();
clientSession = null; clientSession = null;