diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 9e402beb..7a1d6fec 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -79,155 +79,11 @@ namespace MQTTnet.Server return _workerTask; } - private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) - { - if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket)); - if (adapter == null) throw new ArgumentNullException(nameof(adapter)); - - try - { - if (_cancellationTokenSource != null) - { - Stop(MqttClientDisconnectType.Clean, true); - } - - adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted; - adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted; - - _cancellationTokenSource = new CancellationTokenSource(); - - //workaround for https://github.com/dotnet/corefx/issues/24430 -#pragma warning disable 4014 - _cleanupHandle = _cancellationTokenSource.Token.Register(() => TryDisposeAdapterAsync(adapter)); -#pragma warning restore 4014 - //end workaround - - _wasCleanDisconnect = false; - _willMessage = connectPacket.WillMessage; - - _pendingPacketsQueue.Start(adapter, _cancellationTokenSource.Token); - _keepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token); - - _adapterEndpoint = adapter.Endpoint; - _adapterProtocolVersion = adapter.PacketSerializer.ProtocolVersion; - - while (!_cancellationTokenSource.IsCancellationRequested) - { - var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false); - if (packet != null) - { - _keepAliveMonitor.PacketReceived(packet); - ProcessReceivedPacket(adapter, packet, _cancellationTokenSource.Token); - } - } - } - catch (OperationCanceledException) - { - } - catch (Exception exception) - { - if (exception is MqttCommunicationException) - { - if (exception is MqttCommunicationClosedGracefullyException) - { - _logger.Verbose("Client '{0}': Connection closed gracefully.", ClientId); - } - else - { - _logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", ClientId); - } - } - else - { - _logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId); - } - - Stop(MqttClientDisconnectType.NotClean, true); - } - finally - { - _adapterEndpoint = null; - _adapterProtocolVersion = null; - - await TryDisposeAdapterAsync(adapter).ConfigureAwait(false); - - _cleanupHandle?.Dispose(); - _cleanupHandle = null; - - _cancellationTokenSource?.Cancel(false); - _cancellationTokenSource?.Dispose(); - _cancellationTokenSource = null; - } - } - - private async Task TryDisposeAdapterAsync(IMqttChannelAdapter adapter) - { - if (adapter == null) - { - return; - } - - try - { - adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; - adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; - - await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); - } - catch (Exception exception) - { - _logger.Error(exception, exception.Message); - } - finally - { - try - { - adapter.Dispose(); - } - catch - { - } - } - } - public void Stop(MqttClientDisconnectType type) { Stop(type, false); } - private void Stop(MqttClientDisconnectType type, bool isInsideSession) - { - try - { - var cts = _cancellationTokenSource; - if (cts == null || cts.IsCancellationRequested) - { - return; - } - - _cancellationTokenSource?.Cancel(false); - - _wasCleanDisconnect = type == MqttClientDisconnectType.Clean; - - if (_willMessage != null && !_wasCleanDisconnect) - { - _sessionsManager.EnqueueApplicationMessage(this, _willMessage.ToPublishPacket()); - } - - _willMessage = null; - - if (!isInsideSession) - { - _workerTask?.GetAwaiter().GetResult(); - } - } - finally - { - _logger.Info("Client '{0}': Disconnected (clean={1}).", ClientId, _wasCleanDisconnect); - _eventDispatcher.OnClientDisconnected(ClientId, _wasCleanDisconnect); - } - } - public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket) { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); @@ -313,6 +169,163 @@ namespace MQTTnet.Server _cancellationTokenSource = null; } + private void Stop(MqttClientDisconnectType type, bool isInsideSession) + { + try + { + var cts = _cancellationTokenSource; + if (cts == null || cts.IsCancellationRequested) + { + return; + } + + _cancellationTokenSource?.Cancel(false); + + _wasCleanDisconnect = type == MqttClientDisconnectType.Clean; + + if (_willMessage != null && !_wasCleanDisconnect) + { + _sessionsManager.EnqueueApplicationMessage(this, _willMessage.ToPublishPacket()); + } + + _willMessage = null; + + if (!isInsideSession) + { + _workerTask?.GetAwaiter().GetResult(); + } + } + finally + { + _logger.Info("Client '{0}': Disconnected (clean={1}).", ClientId, _wasCleanDisconnect); + _eventDispatcher.OnClientDisconnected(ClientId, _wasCleanDisconnect); + } + } + + private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + { + if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket)); + if (adapter == null) throw new ArgumentNullException(nameof(adapter)); + + try + { + if (_cancellationTokenSource != null) + { + Stop(MqttClientDisconnectType.Clean, true); + } + + adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted; + adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted; + + _cancellationTokenSource = new CancellationTokenSource(); + + //workaround for https://github.com/dotnet/corefx/issues/24430 +#pragma warning disable 4014 + _cleanupHandle = _cancellationTokenSource.Token.Register(async () => + { + await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false); + TryDisposeAdapter(adapter); + }); +#pragma warning restore 4014 + //end workaround + + _wasCleanDisconnect = false; + _willMessage = connectPacket.WillMessage; + + _pendingPacketsQueue.Start(adapter, _cancellationTokenSource.Token); + _keepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token); + + _adapterEndpoint = adapter.Endpoint; + _adapterProtocolVersion = adapter.PacketSerializer.ProtocolVersion; + + while (!_cancellationTokenSource.IsCancellationRequested) + { + var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false); + if (packet != null) + { + _keepAliveMonitor.PacketReceived(packet); + ProcessReceivedPacket(adapter, packet, _cancellationTokenSource.Token); + } + } + } + catch (OperationCanceledException) + { + } + catch (Exception exception) + { + if (exception is MqttCommunicationException) + { + if (exception is MqttCommunicationClosedGracefullyException) + { + _logger.Verbose("Client '{0}': Connection closed gracefully.", ClientId); + } + else + { + _logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", ClientId); + } + } + else + { + _logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId); + } + + Stop(MqttClientDisconnectType.NotClean, true); + } + finally + { + _adapterEndpoint = null; + _adapterProtocolVersion = null; + + // Uncomment as soon as the workaround above is no longer needed. + //await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false); + //TryDisposeAdapter(adapter); + + _cleanupHandle?.Dispose(); + _cleanupHandle = null; + + _cancellationTokenSource?.Cancel(false); + _cancellationTokenSource?.Dispose(); + _cancellationTokenSource = null; + } + } + + private void TryDisposeAdapter(IMqttChannelAdapter adapter) + { + if (adapter == null) + { + return; + } + + try + { + adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted; + adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted; + + adapter.Dispose(); + } + catch (Exception exception) + { + _logger.Error(exception, "Error while disposing channel adapter."); + } + } + + private async Task TryDisconnectAdapterAsync(IMqttChannelAdapter adapter) + { + if (adapter == null) + { + return; + } + + try + { + await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, "Error while disconnecting channel adapter."); + } + } + private void ProcessReceivedPacket(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken) { if (packet is MqttPublishPacket publishPacket) diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 39352ec3..457de926 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -301,11 +301,6 @@ namespace MQTTnet.Core.Tests [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()); @@ -338,6 +333,51 @@ namespace MQTTnet.Core.Tests Assert.AreEqual(clientConnectedCalled, clientDisconnectedCalled); } + [TestMethod] + public async Task MqttServer_Client_Disconnect_Without_Errors() + { + var errors = 0; + + MqttNetGlobalLogger.LogMessagePublished += (_, e) => + { + System.Diagnostics.Debug.WriteLine($"[{e.TraceMessage.Timestamp:s}] {e.TraceMessage.Source} {e.TraceMessage.Message}"); + + if (e.TraceMessage.Level == MqttNetLogLevel.Error) + { + errors++; + } + }; + + bool clientWasConnected; + + var server = new MqttFactory().CreateMqttServer(); + try + { + var options = new MqttServerOptionsBuilder().Build(); + await server.StartAsync(options); + + var client = new MqttFactory().CreateMqttClient(); + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost") + .Build(); + + await client.ConnectAsync(clientOptions); + + clientWasConnected = true; + + await client.DisconnectAsync(); + + await Task.Delay(500); + } + finally + { + await server.StopAsync(); + } + + Assert.IsTrue(clientWasConnected); + Assert.AreEqual(0, errors); + } + [TestMethod] public async Task MqttServer_LotsOfRetainedMessages() {