Merge pull request #289 from rabbitmq/rabbitmq-dotnet-client-288

Use concurrent collections inside AutorecoveringConnection

Author: michaelklishin

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

@@ -43,8 +43,8 @@
 using RabbitMQ.Client.Impl;
 using System;
 using System.Collections.Generic;
+using System.Collections.Concurrent;
 using System.Linq;
-using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -73,19 +73,20 @@ public class AutorecoveringConnection : IConnection, IRecoverable
 
         protected List<AutorecoveringModel> m_models = new List<AutorecoveringModel>();
 
-        protected HashSet<RecordedBinding> m_recordedBindings = new HashSet<RecordedBinding>();
+        protected IDictionary<RecordedBinding, byte> m_recordedBindings =
+            new ConcurrentDictionary<RecordedBinding, byte>();
 
         protected List<EventHandler<ConnectionBlockedEventArgs>> m_recordedBlockedEventHandlers =
             new List<EventHandler<ConnectionBlockedEventArgs>>();
 
         protected IDictionary<string, RecordedConsumer> m_recordedConsumers =
-            new Dictionary<string, RecordedConsumer>();
+            new ConcurrentDictionary<string, RecordedConsumer>();
 
         protected IDictionary<string, RecordedExchange> m_recordedExchanges =
-            new Dictionary<string, RecordedExchange>();
+            new ConcurrentDictionary<string, RecordedExchange>();
 
         protected IDictionary<string, RecordedQueue> m_recordedQueues =
-            new Dictionary<string, RecordedQueue>();
+            new ConcurrentDictionary<string, RecordedQueue>();
 
         protected List<EventHandler<ShutdownEventArgs>> m_recordedShutdownEventHandlers =
             new List<EventHandler<ShutdownEventArgs>>();
@@ -398,7 +399,7 @@ public void DeleteRecordedBinding(RecordedBinding rb)
         {
             lock (m_recordedEntitiesLock)
             {
-                m_recordedBindings.RemoveWhere(b => b.Equals(rb));
+                m_recordedBindings.Remove(rb);
             }
         }
 
@@ -425,11 +426,10 @@ public void DeleteRecordedExchange(string name)
 
                 // find bindings that need removal, check if some auto-delete exchanges
                 // might need the same
-                List<RecordedBinding> bs = m_recordedBindings.Where(b => name.Equals(b.Destination)).
-                                                              ToList();
-                m_recordedBindings.RemoveWhere(b => name.Equals(b.Destination));
+                var bs = m_recordedBindings.Keys.Where(b => name.Equals(b.Destination));
                 foreach (RecordedBinding b in bs)
                 {
+                    m_recordedBindings.Remove(b);
                     MaybeDeleteRecordedAutoDeleteExchange(b.Source);
                 }
             }
@@ -442,11 +442,10 @@ public void DeleteRecordedQueue(string name)
                 m_recordedQueues.Remove(name);
                 // find bindings that need removal, check if some auto-delete exchanges
                 // might need the same
-                List<RecordedBinding> bs = m_recordedBindings.Where(b => name.Equals(b.Destination)).
-                                                              ToList();
-                m_recordedBindings.RemoveWhere(b => name.Equals(b.Destination));
+                var bs = m_recordedBindings.Keys.Where(b => name.Equals(b.Destination));
                 foreach (RecordedBinding b in bs)
                 {
+                    m_recordedBindings.Remove(b);
                     MaybeDeleteRecordedAutoDeleteExchange(b.Source);
                 }
             }
@@ -468,7 +467,7 @@ public void MaybeDeleteRecordedAutoDeleteExchange(string exchange)
         {
             lock (m_recordedEntitiesLock)
             {
-                if (!HasMoreDestinationsBoundToExchange(m_recordedBindings, exchange))
+                if (!HasMoreDestinationsBoundToExchange(m_recordedBindings.Keys, exchange))
                 {
                     RecordedExchange rx;
                     m_recordedExchanges.TryGetValue(exchange, out rx);
@@ -505,7 +504,7 @@ public void RecordBinding(RecordedBinding rb)
         {
             lock (m_recordedEntitiesLock)
             {
-                m_recordedBindings.Add(rb);
+                m_recordedBindings.Add(rb, 0);
             }
         }
 
@@ -752,8 +751,7 @@ protected void PropagateQueueNameChangeToBindings(string oldName, string newName
         {
             lock (m_recordedBindings)
             {
-                IEnumerable<RecordedBinding> bs = m_recordedBindings.
-                    Where(b => b.Destination.Equals(oldName));
+                var bs = m_recordedBindings.Keys.Where(b => b.Destination.Equals(oldName));
                 foreach (RecordedBinding b in bs)
                 {
                     b.Destination = newName;
@@ -776,7 +774,7 @@ protected void PropagateQueueNameChangeToConsumers(string oldName, string newNam
 
         protected void RecoverBindings()
         {
-            foreach (RecordedBinding b in m_recordedBindings)
+            foreach (var b in m_recordedBindings.Keys)
             {
                 try
                 {
@@ -850,8 +848,7 @@ protected void RecoverConnectionUnblockedHandlers()
 
         protected void RecoverConsumers()
         {
-            var dict = new Dictionary<string, RecordedConsumer>(m_recordedConsumers);
-            foreach (KeyValuePair<string, RecordedConsumer> pair in dict)
+            foreach (KeyValuePair<string, RecordedConsumer> pair in m_recordedConsumers)
             {
                 string tag = pair.Key;
                 RecordedConsumer cons = pair.Value;
@@ -938,8 +935,7 @@ protected void RecoverQueues()
         {
             lock (m_recordedQueues)
             {
-                var rqs = new Dictionary<string, RecordedQueue>(m_recordedQueues);
-                foreach (KeyValuePair<string, RecordedQueue> pair in rqs)
+                foreach (KeyValuePair<string, RecordedQueue> pair in m_recordedQueues)
                 {
                     string oldName = pair.Key;
                     RecordedQueue rq = pair.Value;