From 58ad599783daa522fabb4e928f955c089dfa2713 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 30 Oct 2020 00:23:45 +0100 Subject: [PATCH] Improve disconnect code of client. --- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 63 +++++++++---------- Source/MQTTnet/Client/MqttClient.cs | 18 +++--- .../MQTTnet/Implementations/MqttTcpChannel.cs | 22 ++++--- 3 files changed, 50 insertions(+), 53 deletions(-) diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 5cf7fc71..20dc952d 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -116,49 +116,42 @@ namespace MQTTnet.Adapter cancellationToken.ThrowIfCancellationRequested(); ThrowIfDisposed(); - try + using (await _syncRoot.WaitAsync(cancellationToken).ConfigureAwait(false)) { - using (await _syncRoot.WaitAsync(cancellationToken).ConfigureAwait(false)) + // Check for cancellation here again because "WaitAsync" might take some time. + cancellationToken.ThrowIfCancellationRequested(); + + try { - // Check for cancellation here again because "WaitAsync" might take some time. - cancellationToken.ThrowIfCancellationRequested(); + var packetData = PacketFormatterAdapter.Encode(packet); - try + if (timeout == TimeSpan.Zero) { - var packetData = PacketFormatterAdapter.Encode(packet); - - if (timeout == TimeSpan.Zero) - { - await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false); - } - else - { - await MqttTaskTimeout.WaitAsync( - t => _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, t), timeout, cancellationToken).ConfigureAwait(false); - } - - Interlocked.Add(ref _bytesReceived, packetData.Count); - - _logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet); + await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false); } - catch (Exception exception) + else { - if (IsWrappedException(exception)) - { - throw; - } + await MqttTaskTimeout.WaitAsync( + t => _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, t), timeout, cancellationToken).ConfigureAwait(false); + } - WrapAndThrowException(exception); - } - finally - { - PacketFormatterAdapter.FreeBuffer(); - } + Interlocked.Add(ref _bytesReceived, packetData.Count); + + _logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet); + } + catch (Exception exception) + { + if (IsWrappedException(exception)) + { + throw; + } + + WrapAndThrowException(exception); + } + finally + { + PacketFormatterAdapter.FreeBuffer(); } - } - catch (ObjectDisposedException) - { - throw new OperationCanceledException(); } } diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 5263c1e8..d8df4a5f 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -77,7 +77,7 @@ namespace MQTTnet.Client Options = options; _packetIdentifierProvider.Reset(); - _packetDispatcher.Reset(); + _packetDispatcher.Cancel(); _backgroundCancellationTokenSource = new CancellationTokenSource(); var backgroundCancellationToken = _backgroundCancellationTokenSource.Token; @@ -266,8 +266,6 @@ namespace MQTTnet.Client if (disposing) { Cleanup(); - - DisconnectedHandler = null; } base.Dispose(disposing); @@ -464,6 +462,7 @@ namespace MQTTnet.Client if (exception is OperationCanceledException) { + return; } else if (exception is MqttCommunicationException) { @@ -567,15 +566,15 @@ namespace MQTTnet.Client { await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false); } - else if (packet is MqttDisconnectPacket disconnectPacket) + else if (packet is MqttDisconnectPacket) { // Also dispatch disconnect to waiting threads to generate a proper exception. _packetDispatcher.Dispatch(packet); - await DisconnectAsync(new MqttClientDisconnectOptions() + if (!DisconnectIsPending()) { - ReasonCode = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.UnspecifiedError) - }, cancellationToken).ConfigureAwait(false); + await DisconnectInternalAsync(_packetReceiverTask, null, null); + } } else if (packet is MqttAuthPacket authPacket) { @@ -626,7 +625,7 @@ namespace MQTTnet.Client } catch (Exception exception) { - _logger.Error(exception, "Error while enqueueing application message."); + _logger.Error(exception, "Error while queueing application message."); } } @@ -636,8 +635,7 @@ namespace MQTTnet.Client { try { - var publishPacketDequeueResult = - await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken); + var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken); if (!publishPacketDequeueResult.IsSuccess) { return; diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index 4f0a2d02..e3959088 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -138,16 +138,21 @@ namespace MQTTnet.Implementations try { + var stream = _stream; + + if (stream == null) + { + return 0; + } + + if (!stream.CanRead) + { + return 0; + } + // Workaround for: https://github.com/dotnet/corefx/issues/24430 using (cancellationToken.Register(Dispose)) { - var stream = _stream; - - if (stream == null) - { - return 0; - } - return await stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); } } @@ -182,7 +187,7 @@ namespace MQTTnet.Implementations if (stream == null) { - throw new ObjectDisposedException(nameof(stream)); + throw new MqttCommunicationException("The TCP connection is closed."); } await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); @@ -194,6 +199,7 @@ namespace MQTTnet.Implementations } catch (ObjectDisposedException) { + throw new MqttCommunicationException("The TCP connection is closed."); } catch (IOException exception) {