* Remove channels that do not recover successfully

Author: lukebakken

projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs

@@ -144,7 +144,7 @@ public IAsyncBasicConsumer? DefaultConsumer
 
         public string? CurrentQueue => InnerChannel.CurrentQueue;
 
-        internal async Task AutomaticallyRecoverAsync(AutorecoveringConnection conn, bool recoverConsumers,
+        internal async Task<bool> AutomaticallyRecoverAsync(AutorecoveringConnection conn, bool recoverConsumers,
             bool recordedEntitiesSemaphoreHeld, CancellationToken cancellationToken)
         {
             if (false == recordedEntitiesSemaphoreHeld)
@@ -154,7 +154,7 @@ internal async Task AutomaticallyRecoverAsync(AutorecoveringConnection conn, boo
 
             if (_disposed)
             {
-                return;
+                return false;
             }
 
             _connection = conn;
@@ -198,6 +198,7 @@ await newChannel.TxSelectAsync(cancellationToken)
             {
                 await newChannel.AbortAsync()
                     .ConfigureAwait(false);
+                return false;
             }
             else
             {
@@ -210,6 +211,8 @@ await _connection.RecoverConsumersAsync(this, newChannel, recordedEntitiesSemaph
                 }
 
                 _innerChannel.RunRecoveryEventHandlers(this);
+
+                return true;
             }
         }
 

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

@@ -46,7 +46,7 @@ internal sealed partial class AutorecoveringConnection
         private readonly Dictionary<string, RecordedQueue> _recordedQueues = new Dictionary<string, RecordedQueue>();
         private readonly HashSet<RecordedBinding> _recordedBindings = new HashSet<RecordedBinding>();
         private readonly Dictionary<string, RecordedConsumer> _recordedConsumers = new Dictionary<string, RecordedConsumer>();
-        private readonly List<AutorecoveringChannel> _channels = new List<AutorecoveringChannel>();
+        private List<AutorecoveringChannel> _channels = new List<AutorecoveringChannel>();
 
         internal int RecordedExchangesCount => _recordedExchanges.Count;
 

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

@@ -30,6 +30,7 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -583,12 +584,29 @@ private async ValueTask RecoverChannelsAndItsConsumersAsync(bool recordedEntitie
                 throw new InvalidOperationException("recordedEntitiesSemaphore must be held");
             }
 
-            foreach (AutorecoveringChannel channel in _channels)
+            var recoveredChannels = new List<AutorecoveringChannel>();
+            await _channelsSemaphore.WaitAsync(cancellationToken)
+                .ConfigureAwait(false);
+            try
             {
-                await channel.AutomaticallyRecoverAsync(this, _config.TopologyRecoveryEnabled,
-                    recordedEntitiesSemaphoreHeld: recordedEntitiesSemaphoreHeld,
-                    cancellationToken: cancellationToken)
-                    .ConfigureAwait(false);
+                foreach (AutorecoveringChannel channel in _channels)
+                {
+                    bool recovered = await channel.AutomaticallyRecoverAsync(this, _config.TopologyRecoveryEnabled,
+                        recordedEntitiesSemaphoreHeld: recordedEntitiesSemaphoreHeld,
+                        cancellationToken: cancellationToken)
+                        .ConfigureAwait(false);
+
+                    if (recovered)
+                    {
+                        recoveredChannels.Add(channel);
+                    }
+                }
+            }
+            finally
+            {
+                _channels.Clear();
+                _channels = recoveredChannels;
+                _channelsSemaphore.Release();
             }
         }
     }

projects/Test/Applications/GH-1647/GH-1647.csproj

@@ -9,7 +9,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\..\..\RabbitMQ.Client\RabbitMQ.Client.csproj" />
+    <ProjectReference Include="../../../RabbitMQ.Client\RabbitMQ.Client.csproj" />
   </ItemGroup>
 
 </Project>