From f773257c86fd6cb66cc82d4dbb7abdb1d0590c37 Mon Sep 17 00:00:00 2001
From: Christian Kratky <christian.kratky@googlemail.com>
Date: Fri, 30 Oct 2020 20:43:39 +0100
Subject: [PATCH] Improve exception handling in AsyncQueue.

---
 Source/MQTTnet/Internal/AsyncQueue.cs | 48 +++++++++++++++++++--------
 1 file changed, 35 insertions(+), 13 deletions(-)

diff --git a/Source/MQTTnet/Internal/AsyncQueue.cs b/Source/MQTTnet/Internal/AsyncQueue.cs
index 868ed617..4e8d7a93 100644
--- a/Source/MQTTnet/Internal/AsyncQueue.cs
+++ b/Source/MQTTnet/Internal/AsyncQueue.cs
@@ -7,15 +7,19 @@ namespace MQTTnet.Internal
 {
     public sealed class AsyncQueue<TItem> : IDisposable
     {
+        readonly object _syncRoot = new object();
         SemaphoreSlim _semaphore = new SemaphoreSlim(0);
         ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
-        
+
         public int Count => _queue.Count;
 
         public void Enqueue(TItem item)
         {
-            _queue.Enqueue(item);
-            _semaphore?.Release();
+            lock (_syncRoot)
+            {
+                _queue.Enqueue(item);
+                _semaphore?.Release();
+            }
         }
 
         public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
@@ -24,23 +28,38 @@ namespace MQTTnet.Internal
             {
                 try
                 {
-                    if (_semaphore == null)
+                    Task task;
+                    lock (_syncRoot)
+                    {
+                        if (_semaphore == null)
+                        {
+                            return new AsyncQueueDequeueResult<TItem>(false, default);
+                        }
+
+                        task = _semaphore.WaitAsync(cancellationToken);
+                    }
+                    
+                    await task.ConfigureAwait(false);
+
+                    if (cancellationToken.IsCancellationRequested)
                     {
                         return new AsyncQueueDequeueResult<TItem>(false, default);
                     }
 
-                    await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
-                    cancellationToken.ThrowIfCancellationRequested();
+                    if (_queue.TryDequeue(out var item))
+                    {
+                        return new AsyncQueueDequeueResult<TItem>(true, item);
+                    }
+                }
+                catch (ArgumentNullException)
+                {
+                    // The semaphore throws this internally sometimes.
+                    return new AsyncQueueDequeueResult<TItem>(false, default);
                 }
                 catch (OperationCanceledException)
                 {
                     return new AsyncQueueDequeueResult<TItem>(false, default);
                 }
-
-                if (_queue.TryDequeue(out var item))
-                {
-                    return new AsyncQueueDequeueResult<TItem>(true, item);
-                }
             }
 
             return new AsyncQueueDequeueResult<TItem>(false, default);
@@ -63,8 +82,11 @@ namespace MQTTnet.Internal
 
         public void Dispose()
         {
-            _semaphore?.Dispose();
-            _semaphore = null;
+            lock (_syncRoot)
+            {
+                _semaphore?.Dispose();
+                _semaphore = null;
+            }
         }
     }
 }