Fail all waiting tasks when the client connection gets closed (#1563)

.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)

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);
+            }
+        }
+    }
+}

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()
         {

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("#");
+            }
+        }
+    }
+}

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)
+        {
+        }
+    }
+}

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;

Source/MQTTnet/Client/MqttClient.cs

@@ -32,25 +32,21 @@ namespace MQTTnet.Client
         readonly object _disconnectLock = new object();
         readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
         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<MqttPublishPacket> _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);

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)
         {
         }
 

Source/MQTTnet/MQTTnet.csproj.DotSettings

@@ -2,6 +2,7 @@
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cconnecting/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cdiagnostics/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cdisconnecting/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cexceptions/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cextendedauthenticationexchange/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Coptions/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cpublishing/@EntryIndexedValue">True</s:Boolean>

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<IMqttPacketAwaitable> _awaitables = new List<IMqttPacketAwaitable>();
+        readonly List<IMqttPacketAwaitable> _waiters = new List<IMqttPacketAwaitable>();
 
-        public void FailAll(Exception exception)
+        bool _isDisposed;
+
+        public MqttPacketAwaitable<TResponsePacket> AddAwaitable<TResponsePacket>(ushort packetIdentifier) where TResponsePacket : MqttPacket
         {
-            if (exception == null) throw new ArgumentNullException(nameof(exception));
+            var awaitable = new MqttPacketAwaitable<TResponsePacket>(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<IMqttPacketAwaitable>();
-            
-            lock (_awaitables)
+            var waiters = new List<IMqttPacketAwaitable>();
+
+            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<TResponsePacket> AddAwaitable<TResponsePacket>(ushort packetIdentifier) where TResponsePacket : MqttPacket
-        {
-            var awaitable = new MqttPacketAwaitable<TResponsePacket>(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));
             }
         }
     }