Removing the use of a BlockingCollection and replacing with ConcurrentQueue + SemaphoreSlim to get rid of a blocking wait for better concurrency.

Author: Stefán J. Sigurðarson

projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringConnection.cs

@@ -476,10 +476,8 @@ private void Init(IFrameHandler fh)
             {
                 if (ShouldTriggerConnectionRecovery(args))
                 {
-                    if (!_recoveryLoopCommandQueue.TryAdd(RecoveryCommand.BeginAutomaticRecovery))
-                    {
-                        ESLog.Warn("Failed to notify RecoveryLoop to BeginAutomaticRecovery.");
-                    }
+                    _recoveryLoopCommandQueue.Enqueue(RecoveryCommand.BeginAutomaticRecovery);
+                    _semaphore.Release();
                 }
             };
             lock (_eventLock)
@@ -963,7 +961,8 @@ private enum RecoveryConnectionState
         private Task _recoveryTask;
         private RecoveryConnectionState _recoveryLoopState = RecoveryConnectionState.Connected;
 
-        private readonly BlockingCollection<RecoveryCommand> _recoveryLoopCommandQueue = new BlockingCollection<RecoveryCommand>();
+        private readonly ConcurrentQueue<RecoveryCommand> _recoveryLoopCommandQueue = new ConcurrentQueue<RecoveryCommand>();
+        readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
         private readonly CancellationTokenSource _recoveryCancellationToken = new CancellationTokenSource();
         private readonly TaskCompletionSource<int> _recoveryLoopComplete = new TaskCompletionSource<int>();
 
@@ -974,19 +973,31 @@ private async Task MainRecoveryLoop()
         {
             try
             {
-                while (_recoveryLoopCommandQueue.TryTake(out RecoveryCommand command, -1, _recoveryCancellationToken.Token))
+                while (!_recoveryCancellationToken.IsCancellationRequested)
                 {
-                    switch (_recoveryLoopState)
+                    try
+                    {
+                        await _semaphore.WaitAsync(_recoveryCancellationToken.Token).ConfigureAwait(false);
+                    }
+                    catch (TaskCanceledException)
                     {
-                        case RecoveryConnectionState.Connected:
-                            await RecoveryLoopConnectedHandler(command).ConfigureAwait(false);
-                            break;
-                        case RecoveryConnectionState.Recovering:
-                            await RecoveryLoopRecoveringHandler(command).ConfigureAwait(false);
-                            break;
-                        default:
-                            ESLog.Warn("RecoveryLoop state is out of range.");
-                            break;
+                        // Swallowing the task cancellation in case we are stopping work.
+                    }
+
+                    if (!_recoveryCancellationToken.IsCancellationRequested && _recoveryLoopCommandQueue.TryDequeue(out RecoveryCommand command))
+                    {
+                        switch (_recoveryLoopState)
+                        {
+                            case RecoveryConnectionState.Connected:
+                                await RecoveryLoopConnectedHandler(command).ConfigureAwait(false);
+                                break;
+                            case RecoveryConnectionState.Recovering:
+                                await RecoveryLoopRecoveringHandler(command).ConfigureAwait(false);
+                                break;
+                            default:
+                                ESLog.Warn("RecoveryLoop state is out of range.");
+                                break;
+                        }
                     }
                 }
             }
@@ -1034,7 +1045,8 @@ private async Task RecoveryLoopRecoveringHandler(RecoveryCommand command)
                     else
                     {
                         await Task.Delay(_factory.NetworkRecoveryInterval);
-                        _recoveryLoopCommandQueue.TryAdd(RecoveryCommand.PerformAutomaticRecovery);
+                        _recoveryLoopCommandQueue.Enqueue(RecoveryCommand.PerformAutomaticRecovery);
+                        _semaphore.Release();
                     }
 
                     break;
@@ -1058,7 +1070,8 @@ private async Task RecoveryLoopConnectedHandler(RecoveryCommand command)
                 case RecoveryCommand.BeginAutomaticRecovery:
                     _recoveryLoopState = RecoveryConnectionState.Recovering;
                     await Task.Delay(_factory.NetworkRecoveryInterval).ConfigureAwait(false);
-                    _recoveryLoopCommandQueue.TryAdd(RecoveryCommand.PerformAutomaticRecovery);
+                    _recoveryLoopCommandQueue.Enqueue(RecoveryCommand.PerformAutomaticRecovery);
+                    _semaphore.Release();
                     break;
                 default:
                     ESLog.Warn($"RecoveryLoop command {command} is out of range.");