Merge pull request #730 from rabbitmq/rabbitmq-dotnet-client-510

Address topology recovery issue

Author: michaelklishin

RabbitMQDotNetClient.sln

@@ -1,6 +1,6 @@
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio Version 16
-VisualStudioVersion = 16.0.29709.97
+# Visual Studio 15
+VisualStudioVersion = 15.0.28307.1022
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "projects", "projects", "{068D7DC3-8E6E-4951-B9E3-272C641BF0DE}"
 EndProject

projects/client/RabbitMQ.Client/RabbitMQ.Client.csproj

@@ -24,6 +24,7 @@
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>
     <AllowedOutputExtensionsInPackageBuildOutputFolder>$(AllowedOutputExtensionsInPackageBuildOutputFolder);.pdb</AllowedOutputExtensionsInPackageBuildOutputFolder>
+    <LangVersion>latest</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>

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

@@ -64,15 +64,17 @@ internal sealed class AutorecoveringConnection : IConnection
 
         private readonly object _recordedEntitiesLock = new object();
 
-        private readonly List<AutorecoveringModel> _models = new List<AutorecoveringModel>();
+        private readonly IDictionary<string, RecordedExchange> _recordedExchanges = new Dictionary<string, RecordedExchange>();
 
-        private readonly ConcurrentDictionary<RecordedBinding, byte> _recordedBindings =
-            new ConcurrentDictionary<RecordedBinding, byte>();
+        private readonly IDictionary<string, RecordedQueue> _recordedQueues = new Dictionary<string, RecordedQueue>();
 
-        private EventHandler<ConnectionBlockedEventArgs> _recordedBlockedEventHandlers;
+        private readonly IDictionary<RecordedBinding, byte> _recordedBindings = new Dictionary<RecordedBinding, byte>();
+
+        private readonly IDictionary<string, RecordedConsumer> _recordedConsumers = new Dictionary<string, RecordedConsumer>();
 
-        private readonly IDictionary<string, RecordedConsumer> _recordedConsumers =
-            new ConcurrentDictionary<string, RecordedConsumer>();
+        private readonly ICollection<AutorecoveringModel> _models = new List<AutorecoveringModel>();
+
+        private EventHandler<ConnectionBlockedEventArgs> _recordedBlockedEventHandlers;
         private EventHandler<ShutdownEventArgs> _recordedShutdownEventHandlers;
         private EventHandler<EventArgs> _recordedUnblockedEventHandlers;
 
@@ -301,7 +303,7 @@ public void DeleteRecordedBinding(RecordedBinding rb)
         {
             lock (_recordedEntitiesLock)
             {
-                ((IDictionary<RecordedBinding, byte>)_recordedBindings).Remove(rb);
+                _recordedBindings.Remove(rb);
             }
         }
 
@@ -324,11 +326,11 @@ public void DeleteRecordedExchange(string name)
         {
             lock (_recordedEntitiesLock)
             {
-                RecordedExchanges.Remove(name);
+                _recordedExchanges.Remove(name);
 
                 // find bindings that need removal, check if some auto-delete exchanges
                 // might need the same
-                IEnumerable<RecordedBinding> bs = _recordedBindings.Keys.Where(b => name.Equals(b.Destination));
+                IEnumerable<RecordedBinding> bs = _recordedBindings.Keys.Where(b => name.Equals(b.Destination)).ToArray();
                 foreach (RecordedBinding b in bs)
                 {
                     DeleteRecordedBinding(b);
@@ -341,10 +343,10 @@ public void DeleteRecordedQueue(string name)
         {
             lock (_recordedEntitiesLock)
             {
-                RecordedQueues.Remove(name);
+                _recordedQueues.Remove(name);
                 // find bindings that need removal, check if some auto-delete exchanges
                 // might need the same
-                IEnumerable<RecordedBinding> bs = _recordedBindings.Keys.Where(b => name.Equals(b.Destination));
+                IEnumerable<RecordedBinding> bs = _recordedBindings.Keys.Where(b => name.Equals(b.Destination)).ToArray();
                 foreach (RecordedBinding b in bs)
                 {
                     DeleteRecordedBinding(b);
@@ -371,13 +373,13 @@ public void MaybeDeleteRecordedAutoDeleteExchange(string exchange)
             {
                 if (!HasMoreDestinationsBoundToExchange(_recordedBindings.Keys, exchange))
                 {
-                    RecordedExchanges.TryGetValue(exchange, out RecordedExchange rx);
+                    _recordedExchanges.TryGetValue(exchange, out RecordedExchange rx);
                     // last binding where this exchange is the source is gone,
                     // remove recorded exchange
                     // if it is auto-deleted. See bug 26364.
                     if ((rx != null) && rx.IsAutoDelete)
                     {
-                        RecordedExchanges.Remove(exchange);
+                        _recordedExchanges.Remove(exchange);
                     }
                 }
             }
@@ -389,12 +391,12 @@ public void MaybeDeleteRecordedAutoDeleteQueue(string queue)
             {
                 if (!HasMoreConsumersOnQueue(_recordedConsumers.Values, queue))
                 {
-                    RecordedQueues.TryGetValue(queue, out RecordedQueue rq);
+                    _recordedQueues.TryGetValue(queue, out RecordedQueue rq);
                     // last consumer on this connection is gone, remove recorded queue
                     // if it is auto-deleted. See bug 26364.
                     if ((rq != null) && rq.IsAutoDelete)
                     {
-                        RecordedQueues.Remove(queue);
+                        _recordedQueues.Remove(queue);
                     }
                 }
             }
@@ -404,7 +406,10 @@ public void RecordBinding(RecordedBinding rb)
         {
             lock (_recordedEntitiesLock)
             {
-                _recordedBindings.TryAdd(rb, 0);
+                if (!_recordedBindings.ContainsKey(rb))
+                {
+                    _recordedBindings.Add(rb, 0);
+                }
             }
         }
 
@@ -423,15 +428,15 @@ public void RecordExchange(string name, RecordedExchange x)
         {
             lock (_recordedEntitiesLock)
             {
-                RecordedExchanges[name] = x;
+                _recordedExchanges[name] = x;
             }
         }
 
         public void RecordQueue(string name, RecordedQueue q)
         {
             lock (_recordedEntitiesLock)
             {
-                RecordedQueues[name] = q;
+                _recordedQueues[name] = q;
             }
         }
 
@@ -497,15 +502,19 @@ public void Abort()
         {
             StopRecoveryLoop();
             if (_delegate.IsOpen)
+            {
                 _delegate.Abort();
+            }
         }
 
         ///<summary>API-side invocation of connection abort.</summary>
         public void Abort(ushort reasonCode, string reasonText)
         {
             StopRecoveryLoop();
             if (_delegate.IsOpen)
+            {
                 _delegate.Abort(reasonCode, reasonText);
+            }
         }
 
         ///<summary>API-side invocation of connection abort with timeout.</summary>
@@ -523,23 +532,29 @@ public void Abort(ushort reasonCode, string reasonText, TimeSpan timeout)
         {
             StopRecoveryLoop();
             if (_delegate.IsOpen)
+            {
                 _delegate.Abort(reasonCode, reasonText, timeout);
+            }
         }
 
         ///<summary>API-side invocation of connection.close.</summary>
         public void Close()
         {
             StopRecoveryLoop();
             if (_delegate.IsOpen)
+            {
                 _delegate.Close();
+            }
         }
 
         ///<summary>API-side invocation of connection.close.</summary>
         public void Close(ushort reasonCode, string reasonText)
         {
             StopRecoveryLoop();
             if (_delegate.IsOpen)
+            {
                 _delegate.Close(reasonCode, reasonText);
+            }
         }
 
         ///<summary>API-side invocation of connection.close with timeout.</summary>
@@ -565,16 +580,19 @@ public void Close(ushort reasonCode, string reasonText, TimeSpan timeout)
         public IModel CreateModel()
         {
             EnsureIsOpen();
-            AutorecoveringModel m;
-            m = new AutorecoveringModel(this,
-                CreateNonRecoveringModel());
+            AutorecoveringModel m = new AutorecoveringModel(this, CreateNonRecoveringModel());
             lock (_models)
             {
                 _models.Add(m);
             }
             return m;
         }
 
+        void IDisposable.Dispose()
+        {
+            Dispose(true);
+        }
+
         public void HandleConnectionBlocked(string reason)
         {
             _delegate.HandleConnectionBlocked(reason);
@@ -585,9 +603,26 @@ public void HandleConnectionUnblocked()
             _delegate.HandleConnectionUnblocked();
         }
 
-        void IDisposable.Dispose()
+        internal int RecordedExchangesCount
         {
-            Dispose(true);
+            get
+            {
+                lock (_recordedEntitiesLock)
+                {
+                    return _recordedExchanges.Count;
+                }
+            }
+        }
+
+        internal int RecordedQueuesCount
+        {
+            get
+            {
+                lock (_recordedEntitiesLock)
+                {
+                    return _recordedExchanges.Count;
+                }
+            }
         }
 
         private void Dispose(bool disposing)
@@ -639,7 +674,7 @@ private void PropagateQueueNameChangeToBindings(string oldName, string newName)
 
         private void PropagateQueueNameChangeToConsumers(string oldName, string newName)
         {
-            lock (_recordedBindings)
+            lock (_recordedConsumers)
             {
                 IEnumerable<KeyValuePair<string, RecordedConsumer>> cs = _recordedConsumers.
                     Where(pair => pair.Value.Queue.Equals(oldName));
@@ -652,7 +687,13 @@ private void PropagateQueueNameChangeToConsumers(string oldName, string newName)
 
         private void RecoverBindings()
         {
-            foreach (RecordedBinding b in _recordedBindings.Keys)
+            IDictionary<RecordedBinding, byte> recordedBindingsCopy = null;
+            lock (_recordedBindings)
+            {
+                recordedBindingsCopy = _recordedBindings.ToDictionary(e => e.Key, e => e.Value);
+            }
+
+            foreach (RecordedBinding b in recordedBindingsCopy.Keys)
             {
                 try
                 {
@@ -717,7 +758,13 @@ private void RecoverConnectionUnblockedHandlers()
 
         private void RecoverConsumers()
         {
-            foreach (KeyValuePair<string, RecordedConsumer> pair in _recordedConsumers)
+            IDictionary<string, RecordedConsumer> recordedConsumersCopy = null;
+            lock (_recordedConsumers)
+            {
+                recordedConsumersCopy = _recordedConsumers.ToDictionary(e => e.Key, e => e.Value);
+            }
+
+            foreach (KeyValuePair<string, RecordedConsumer> pair in recordedConsumersCopy)
             {
                 string tag = pair.Key;
                 RecordedConsumer cons = pair.Value;
@@ -771,7 +818,13 @@ private void RecoverEntities()
 
         private void RecoverExchanges()
         {
-            foreach (RecordedExchange rx in RecordedExchanges.Values)
+            IDictionary<string, RecordedExchange> recordedExchangesCopy = null;
+            lock (_recordedEntitiesLock)
+            {
+                recordedExchangesCopy = _recordedExchanges.ToDictionary(e => e.Key, e => e.Value);
+            }
+
+            foreach (RecordedExchange rx in recordedExchangesCopy.Values)
             {
                 try
                 {
@@ -799,18 +852,24 @@ private void RecoverModels()
 
         private void RecoverQueues()
         {
-            lock (RecordedQueues)
+            IDictionary<string, RecordedQueue> recordedQueuesCopy = null;
+            lock (_recordedEntitiesLock)
+            {
+                recordedQueuesCopy = _recordedQueues.ToDictionary(entry => entry.Key, entry => entry.Value);
+            }
+
+            foreach (KeyValuePair<string, RecordedQueue> pair in recordedQueuesCopy)
             {
-                foreach (KeyValuePair<string, RecordedQueue> pair in RecordedQueues)
+                string oldName = pair.Key;
+                RecordedQueue rq = pair.Value;
+
+                try
                 {
-                    string oldName = pair.Key;
-                    RecordedQueue rq = pair.Value;
+                    rq.Recover();
+                    string newName = rq.Name;
 
-                    try
+                    if (!oldName.Equals(newName))
                     {
-                        rq.Recover();
-                        string newName = rq.Name;
-
                         // Make sure server-named queues are re-added with
                         // their new names.
                         // We only remove old name after we've updated the bindings and consumers,
@@ -840,12 +899,12 @@ private void RecoverQueues()
                             }
                         }
                     }
-                    catch (Exception cause)
-                    {
-                        string s = string.Format("Caught an exception while recovering queue {0}: {1}",
-                            oldName, cause.Message);
-                        HandleTopologyRecoveryException(new TopologyRecoveryException(s, cause));
-                    }
+                }
+                catch (Exception cause)
+                {
+                    string s = string.Format("Caught an exception while recovering queue {0}: {1}",
+                        oldName, cause.Message);
+                    HandleTopologyRecoveryException(new TopologyRecoveryException(s, cause));
                 }
             }
         }

projects/client/Unit/Unit.csproj

@@ -11,7 +11,7 @@
     <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
     <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
     <GenerateAssemblyCopyrightAttribute>false</GenerateAssemblyCopyrightAttribute>
-    <LangVersion>default</LangVersion>
+    <LangVersion>latest</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>