diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index cb58d55b..3ef3a30e 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,4 +1,5 @@ * [Core] Fixed several dead locks by implementing a new async signal (#1552) +* [Client] All waiting tasks will now fail when the client connection gets closed. This fixes several freezes (#1561). * [Server] Fixed duplicated invocation of the event _ClientAcknowledgedPublishPacketAsync_ for QoS level 2 (#1550) * [Server] Fixed issue in upgrading and downgrading of QoS levels for subscriptions and retained messages (#1560) * [Server] Fixed memory leak when old sessions are discarded (#1553) \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Server/Publishing_Tests.cs b/Source/MQTTnet.Tests/Server/Publishing_Tests.cs new file mode 100644 index 00000000..c5df97f7 --- /dev/null +++ b/Source/MQTTnet.Tests/Server/Publishing_Tests.cs @@ -0,0 +1,31 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Protocol; + +namespace MQTTnet.Tests.Server +{ + [TestClass] + public sealed class Publishing_Tests : BaseTestClass + { + [TestMethod] + [ExpectedException(typeof(MqttClientDisconnectedException))] + public async Task Disconnect_While_Publishing() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + // The client will be disconnect directly after subscribing! + server.InterceptingPublishAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection); + + var client = await testEnvironment.ConnectClient(); + await client.PublishStringAsync("test", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs index b67adea9..d4631bc2 100644 --- a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs @@ -46,6 +46,22 @@ namespace MQTTnet.Tests.Server } } + [TestMethod] + [ExpectedException(typeof(MqttClientDisconnectedException))] + public async Task Disconnect_While_Subscribing() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + // The client will be disconnect directly after subscribing! + server.ClientSubscribedTopicAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection); + + var client = await testEnvironment.ConnectClient(); + await client.SubscribeAsync("#"); + } + } + [TestMethod] public async Task Enqueue_Message_After_Subscription() { diff --git a/Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs b/Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs new file mode 100644 index 00000000..5692fa8d --- /dev/null +++ b/Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs @@ -0,0 +1,32 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Protocol; + +namespace MQTTnet.Tests.Server +{ + [TestClass] + public sealed class Unsubscribe_Tests : BaseTestClass + { + [TestMethod] + [ExpectedException(typeof(MqttClientDisconnectedException))] + public async Task Disconnect_While_Unsubscribing() + { + using (var testEnvironment = CreateTestEnvironment()) + { + var server = await testEnvironment.StartServer(); + + // The client will be disconnect directly after subscribing! + server.ClientUnsubscribedTopicAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection); + + var client = await testEnvironment.ConnectClient(); + await client.SubscribeAsync("#"); + await client.UnsubscribeAsync("#"); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Exceptions/MqttClientDisconnectedException.cs b/Source/MQTTnet/Client/Exceptions/MqttClientDisconnectedException.cs new file mode 100644 index 00000000..ce548428 --- /dev/null +++ b/Source/MQTTnet/Client/Exceptions/MqttClientDisconnectedException.cs @@ -0,0 +1,16 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using MQTTnet.Exceptions; + +namespace MQTTnet.Client +{ + public sealed class MqttClientDisconnectedException : MqttCommunicationException + { + public MqttClientDisconnectedException(Exception innerException) : base("The MQTT client is disconnected.", innerException) + { + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Exceptions/MqttUnexpectedDisconnectReceivedException.cs b/Source/MQTTnet/Client/Exceptions/MqttClientUnexpectedDisconnectReceivedException.cs similarity index 79% rename from Source/MQTTnet/Exceptions/MqttUnexpectedDisconnectReceivedException.cs rename to Source/MQTTnet/Client/Exceptions/MqttClientUnexpectedDisconnectReceivedException.cs index eca736fc..5501e39c 100644 --- a/Source/MQTTnet/Exceptions/MqttUnexpectedDisconnectReceivedException.cs +++ b/Source/MQTTnet/Client/Exceptions/MqttClientUnexpectedDisconnectReceivedException.cs @@ -3,14 +3,15 @@ // See the LICENSE file in the project root for more information. using System.Collections.Generic; +using MQTTnet.Exceptions; using MQTTnet.Packets; using MQTTnet.Protocol; -namespace MQTTnet.Exceptions +namespace MQTTnet.Client { - public class MqttUnexpectedDisconnectReceivedException : MqttCommunicationException + public sealed class MqttClientUnexpectedDisconnectReceivedException : MqttCommunicationException { - public MqttUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket) + public MqttClientUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket) : base($"Unexpected DISCONNECT (Reason code={disconnectPacket.ReasonCode}) received.") { ReasonCode = disconnectPacket.ReasonCode; diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 48c32bf4..50e9223a 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -32,25 +32,21 @@ namespace MQTTnet.Client readonly object _disconnectLock = new object(); readonly AsyncEvent _inspectPacketEvent = new AsyncEvent(); readonly MqttNetSourceLogger _logger; - readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); readonly IMqttNetLogger _rootLogger; IMqttChannelAdapter _adapter; - CancellationTokenSource _backgroundCancellationTokenSource; bool _cleanDisconnectInitiated; volatile int _connectionStatus; - MqttClientDisconnectReason _disconnectReason; string _disconnectReasonString; Task _keepAlivePacketsSenderTask; - DateTime _lastPacketSentTimestamp; + MqttPacketDispatcher _packetDispatcher; Task _packetReceiverTask; - AsyncQueue _publishPacketReceiverQueue; Task _publishPacketReceiverTask; @@ -128,7 +124,7 @@ namespace MQTTnet.Client } _packetIdentifierProvider.Reset(); - _packetDispatcher.CancelAll(); + _packetDispatcher = new MqttPacketDispatcher(); _backgroundCancellationTokenSource = new CancellationTokenSource(); var backgroundCancellationToken = _backgroundCancellationTokenSource.Token; @@ -414,6 +410,10 @@ namespace MQTTnet.Client _publishPacketReceiverQueue = null; _adapter?.Dispose(); + _adapter = null; + + _packetDispatcher?.Dispose(); + _packetDispatcher = null; } } @@ -447,6 +447,8 @@ namespace MQTTnet.Client try { + _packetDispatcher.Dispose(new MqttClientDisconnectedException(exception)); + var receiverTask = WaitForTaskAsync(_packetReceiverTask, sender); var publishPacketReceiverTask = WaitForTaskAsync(_publishPacketReceiverTask, sender); var keepAliveTask = WaitForTaskAsync(_keepAlivePacketsSenderTask, sender); @@ -558,7 +560,7 @@ namespace MQTTnet.Client _disconnectReasonString = disconnectPacket.ReasonString; // Also dispatch disconnect to waiting threads to generate a proper exception. - _packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket)); + _packetDispatcher.Dispose(new MqttClientUnexpectedDisconnectReceivedException(disconnectPacket)); return DisconnectInternalAsync(_packetReceiverTask, null, null); } @@ -776,7 +778,7 @@ namespace MQTTnet.Client } else { - _logger.Error(exception, "Error while receiving packets."); + _logger.Error(exception, $"Error while processing received packet ({packet.GetType().Name})."); } _packetDispatcher.FailAll(exception); diff --git a/Source/MQTTnet/Exceptions/MqttCommunicationException.cs b/Source/MQTTnet/Exceptions/MqttCommunicationException.cs index 6c6ba053..a78c1f7c 100644 --- a/Source/MQTTnet/Exceptions/MqttCommunicationException.cs +++ b/Source/MQTTnet/Exceptions/MqttCommunicationException.cs @@ -8,12 +8,8 @@ namespace MQTTnet.Exceptions { public class MqttCommunicationException : Exception { - protected MqttCommunicationException() - { - } - public MqttCommunicationException(Exception innerException) - : base(innerException.Message, innerException) + : base(innerException?.Message ?? "MQTT communication failed.", innerException) { } diff --git a/Source/MQTTnet/MQTTnet.csproj.DotSettings b/Source/MQTTnet/MQTTnet.csproj.DotSettings index 4af65895..a756552c 100644 --- a/Source/MQTTnet/MQTTnet.csproj.DotSettings +++ b/Source/MQTTnet/MQTTnet.csproj.DotSettings @@ -2,6 +2,7 @@ True True True + True True True True diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs index dae92f73..85413a2c 100644 --- a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs @@ -2,48 +2,103 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. -using MQTTnet.Packets; using System; using System.Collections.Generic; +using MQTTnet.Packets; namespace MQTTnet.PacketDispatcher { - public sealed class MqttPacketDispatcher + public sealed class MqttPacketDispatcher : IDisposable { - readonly List _awaitables = new List(); + readonly List _waiters = new List(); - public void FailAll(Exception exception) + bool _isDisposed; + + public MqttPacketAwaitable AddAwaitable(ushort packetIdentifier) where TResponsePacket : MqttPacket { - if (exception == null) throw new ArgumentNullException(nameof(exception)); + var awaitable = new MqttPacketAwaitable(packetIdentifier, this); - lock (_awaitables) + lock (_waiters) { - foreach (var awaitable in _awaitables) - { - awaitable.Fail(exception); - } - - _awaitables.Clear(); + _waiters.Add(awaitable); } + + return awaitable; } public void CancelAll() { - lock (_awaitables) + lock (_waiters) { - foreach (var awaitable in _awaitables) + foreach (var awaitable in _waiters) { awaitable.Cancel(); } - _awaitables.Clear(); + _waiters.Clear(); } } - + + public void Dispose() + { + Dispose(new ObjectDisposedException(nameof(MqttPacketDispatcher))); + } + + public void Dispose(Exception exception) + { + if (exception == null) + { + throw new ArgumentNullException(nameof(exception)); + } + + lock (_waiters) + { + FailAll(exception); + + // Make sure that no task can start waiting after this instance is already disposed. + // This will prevent unexpected freezes. + _isDisposed = true; + } + } + + public void FailAll(Exception exception) + { + if (exception == null) + { + throw new ArgumentNullException(nameof(exception)); + } + + lock (_waiters) + { + foreach (var awaitable in _waiters) + { + awaitable.Fail(exception); + } + + _waiters.Clear(); + } + } + + public void RemoveAwaitable(IMqttPacketAwaitable awaitable) + { + if (awaitable == null) + { + throw new ArgumentNullException(nameof(awaitable)); + } + + lock (_waiters) + { + _waiters.Remove(awaitable); + } + } + public bool TryDispatch(MqttPacket packet) { - if (packet == null) throw new ArgumentNullException(nameof(packet)); - + if (packet == null) + { + throw new ArgumentNullException(nameof(packet)); + } + ushort identifier = 0; if (packet is MqttPacketWithIdentifier packetWithIdentifier) { @@ -51,13 +106,15 @@ namespace MQTTnet.PacketDispatcher } var packetType = packet.GetType(); - var awaitables = new List(); - - lock (_awaitables) + var waiters = new List(); + + lock (_waiters) { - for (var i = _awaitables.Count - 1; i >= 0; i--) + ThrowIfDisposed(); + + for (var i = _waiters.Count - 1; i >= 0; i--) { - var entry = _awaitables[i]; + var entry = _waiters[i]; // Note: The PingRespPacket will also arrive here and has NO identifier but there // is code which waits for it. So the code must be able to deal with filters which @@ -66,39 +123,25 @@ namespace MQTTnet.PacketDispatcher { continue; } - - awaitables.Add(entry); - _awaitables.RemoveAt(i); + + waiters.Add(entry); + _waiters.RemoveAt(i); } } - - foreach (var matchingEntry in awaitables) + + foreach (var matchingEntry in waiters) { matchingEntry.Complete(packet); } - return awaitables.Count > 0; - } - - public MqttPacketAwaitable AddAwaitable(ushort packetIdentifier) where TResponsePacket : MqttPacket - { - var awaitable = new MqttPacketAwaitable(packetIdentifier, this); - - lock (_awaitables) - { - _awaitables.Add(awaitable); - } - - return awaitable; + return waiters.Count > 0; } - public void RemoveAwaitable(IMqttPacketAwaitable awaitable) + void ThrowIfDisposed() { - if (awaitable == null) throw new ArgumentNullException(nameof(awaitable)); - - lock (_awaitables) + if (_isDisposed) { - _awaitables.Remove(awaitable); + throw new ObjectDisposedException(nameof(MqttPacketDispatcher)); } } }