Improve bindings recovery (#49)

* improve bindings recovery. Remove the bindings when a queue or an exchange is removed
---------

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

README.md

@@ -31,8 +31,8 @@ The client is distributed via [NuGet](https://www.nuget.org/packages/RabbitMQ.AM
 - [x] Recovery consumers on connection lost
 - [x] Implement Environment to manage the connections
 - [x] Complete the consumer part with `pause` and `unpause`
-- [ ] Complete the binding/unbinding with the special characters
-- [ ] Complete the queues/exchanges name with the special characters
+- [x] Complete the binding/unbinding with the special characters
+- [x] Complete the queues/exchanges name with the special characters
 - [ ] Implement metrics ( See `System.Diagnostics.DiagnosticSource` [Link](https://learn.microsoft.com/en-us/dotnet/core/diagnostics/metrics-instrumentation) )
 - [x] Recovery exchanges on connection lost
 - [x] Recovery bindings on connection lost

RabbitMQ.AMQP.Client/IEntities.cs

@@ -152,7 +152,7 @@ public interface IExchangeSpecification : IEntitySpecification
 public interface IBindingSpecification
 {
     IBindingSpecification SourceExchange(IExchangeSpecification exchangeSpec);
-    
+
     IBindingSpecification SourceExchange(string exchangeName);
     string SourceExchangeName();
 

RabbitMQ.AMQP.Client/ITopologyListener.cs

@@ -9,17 +9,17 @@ public interface ITopologyListener
     void ExchangeDeclared(IExchangeSpecification specification);
 
     void ExchangeDeleted(string name);
-    
-    
+
+
     void BindingDeclared(IBindingSpecification specification);
-        
+
     void BindingDeleted(string path);
 
     void Clear();
 
     int QueueCount();
 
     int ExchangeCount();
-    
+
     int BindingCount();
 }

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -567,8 +567,8 @@ await Management.Exchange(spec).DeclareAsync()
             }
         }
     }
-    
-    
+
+
     public async Task VisitBindingsAsync(IEnumerable<BindingSpec> bindingSpec)
     {
         // TODO this could be done in parallel

RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs

@@ -106,7 +106,7 @@ public IBindingSpecification Binding()
     {
         return new AmqpBindingSpecification(this);
     }
-    
+
     public IBindingSpecification Binding(BindingSpec spec)
     {
         return Binding()

RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs

@@ -24,6 +24,29 @@ public class RecordingTopologyListener : ITopologyListener
 
     private readonly ConcurrentDictionary<string, BindingSpec> _bindingSpecifications = new();
 
+    private void RemoveBindingsSpecificationFromQueue(string queueName)
+    {
+        foreach (var binding in _bindingSpecifications.Values)
+        {
+            if (binding.DestinationQueue == queueName)
+            {
+                _bindingSpecifications.TryRemove(binding.Path, out _);
+            }
+        }
+    }
+
+    private void RemoveBindingsSpecificationFromExchange(string exchangeName)
+    {
+        foreach (var binding in _bindingSpecifications.Values)
+        {
+            if (binding.SourceExchange == exchangeName
+                || binding.DestinationExchange == exchangeName)
+            {
+                _bindingSpecifications.TryRemove(binding.Path, out _);
+            }
+        }
+    }
+
 
     public void QueueDeclared(IQueueSpecification specification)
     {
@@ -33,6 +56,7 @@ public void QueueDeclared(IQueueSpecification specification)
     public void QueueDeleted(string name)
     {
         _queueSpecifications.TryRemove(name, out _);
+        RemoveBindingsSpecificationFromQueue(name);
     }
 
     public void ExchangeDeclared(IExchangeSpecification specification)
@@ -43,6 +67,7 @@ public void ExchangeDeclared(IExchangeSpecification specification)
     public void ExchangeDeleted(string name)
     {
         _exchangeSpecifications.TryRemove(name, out _);
+        RemoveBindingsSpecificationFromExchange(name);
     }
 
     public void BindingDeclared(IBindingSpecification specification)

Tests/BindingsTests.cs

@@ -41,7 +41,8 @@ public async Task SimpleBindingsBetweenExchangeAndQueue(string sourceExchange, s
 
         await bindingSpec.UnbindAsync();
 
-        await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistAsync(sourceExchangeSpec, destinationQueueSpec);
+        await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistAsync(sourceExchangeSpec,
+            destinationQueueSpec);
 
         /*
          * TODO dispose assertions?
@@ -105,7 +106,12 @@ public async Task BindBetweenExchangeAndQueueTwoTimes()
         "[7][8][9] 他被广泛认为是理论计算机科学和人工智能之父。")]
     [InlineData("ήταν Άγγλος μαθηματικός, επιστήμονας υπολογιστών",
         "ήταν Άγγλος μαθηματικός, επιστήμονας", "επι")]
-    public async Task SimpleBindingsBetweenExchangeAndExchange(string sourceExchangeName, string destinationExchangeName,
+    [InlineData("(~~~!!++@~./.,€€#!!±§##§¶¡€#¢)",
+        "~~~!!++@----.", "==`£!-=+")]
+
+
+    public async Task SimpleBindingsBetweenExchangeAndExchange(string sourceExchangeName,
+        string destinationExchangeName,
         string key)
     {
         Assert.NotNull(_connection);
@@ -127,11 +133,13 @@ await WhenAllComplete(
         await bindingSpecification.BindAsync();
 
         await SystemUtils.WaitUntilExchangeExistsAsync(sourceExchangeSpec);
-        await SystemUtils.WaitUntilBindingsBetweenExchangeAndExchangeExistAsync(sourceExchangeSpec, destinationExchangeSpec);
+        await SystemUtils.WaitUntilBindingsBetweenExchangeAndExchangeExistAsync(sourceExchangeSpec,
+            destinationExchangeSpec);
 
         await bindingSpecification.UnbindAsync();
 
-        await SystemUtils.WaitUntilBindingsBetweenExchangeAndExchangeDontExistAsync(sourceExchangeSpec, destinationExchangeSpec);
+        await SystemUtils.WaitUntilBindingsBetweenExchangeAndExchangeDontExistAsync(sourceExchangeSpec,
+            destinationExchangeSpec);
 
         await sourceExchangeSpec.DeleteAsync();
         await destinationExchangeSpec.DeleteAsync();
@@ -146,6 +154,7 @@ await WhenAllComplete(
     [InlineData("B", 10000L, "H", 0.0001)]
     [InlineData("是英国", 10000.32, "W", 3.0001)]
     [InlineData("是英国", "是英国23", "W", 3.0001)]
+    [InlineData("(~~~!!++@----./.,€€#####§¶¡€#¢)", "~~~!!++@----", "==`£!-=+", "===£!-=+")]
     public async Task BindingsBetweenExchangeAndQueuesWithArgumentsDifferentValues(string key1, object value1,
         string key2, object value2)
     {
@@ -196,8 +205,7 @@ await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistWithArgsAsync
     [Theory]
     [InlineData("my_source_exchange_multi_123", "my_destination_789", "myKey")]
     [InlineData("是英国v_", "destination_是英国v_", "μαθηματικός")]
-    // TODO: to validate.  Atm it seems there is a server side problem
-    // [InlineData("(~~~!!++@----./.,€€#####§¶¡€#¢)", ",,~~~!!++@----./.,€€#####§¶¡€#¢@@@", "===£!-=+")]
+    [InlineData("(~~~!!++@----./.,€€#####§¶¡€#¢)", ",,~~~!!++@----./.,€€#####§¶¡€#¢@@@", "===£!-=+")]
     public async Task MultiBindingsBetweenExchangeAndQueuesWithArgumentsDifferentValues(string source,
         string destination, string key)
     {
@@ -210,7 +218,6 @@ public async Task MultiBindingsBetweenExchangeAndQueuesWithArgumentsDifferentVal
         await WhenAllComplete(exchangeSpec.DeclareAsync(), queueSpec.DeclareAsync());
 
         // add 10 bindings to have a list of bindings to find
-        var bindingSpecs = new List<IBindingSpecification>();
         var bindingSpecTasks = new List<Task>();
         for (int i = 0; i < 10; i++)
         {
@@ -219,31 +226,35 @@ public async Task MultiBindingsBetweenExchangeAndQueuesWithArgumentsDifferentVal
                 .DestinationQueue(queueSpec)
                 .Key(key) // single key to use different args
                 .Arguments(new Dictionary<string, object>() { { $"是英国v_{i}", $"p_{i}" } });
-            bindingSpecs.Add(bindingSpec);
             bindingSpecTasks.Add(bindingSpec.BindAsync());
         }
+
         await WhenAllComplete(bindingSpecTasks);
         bindingSpecTasks.Clear();
 
-        var specialBindArgs = new Dictionary<string, object>() { { $"v_8", $"p_8" }, { $"v_1", 1 }, { $"v_r", 1000L }, };
+        var specialBindArgs =
+            new Dictionary<string, object>() { { $"v_8", $"p_8" }, { $"v_1", 1 }, { $"v_r", 1000L }, };
         IBindingSpecification specialBindSpec = _management.Binding()
             .SourceExchange(exchangeSpec)
             .DestinationQueue(queueSpec)
             .Key(key) // single key to use different args
             .Arguments(specialBindArgs);
         await specialBindSpec.BindAsync();
 
-        await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueExistWithArgsAsync(exchangeSpec, queueSpec, specialBindArgs);
+        await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueExistWithArgsAsync(exchangeSpec, queueSpec,
+            specialBindArgs);
 
         await specialBindSpec.UnbindAsync();
 
-        await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistWithArgsAsync(exchangeSpec, queueSpec, specialBindArgs);
+        await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistWithArgsAsync(exchangeSpec, queueSpec,
+            specialBindArgs);
 
         for (int i = 0; i < 10; i++)
         {
             var bindArgs = new Dictionary<string, object>() { { $"是英国v_{i}", $"p_{i}" } };
 
-            await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueExistWithArgsAsync(exchangeSpec, queueSpec, bindArgs);
+            await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueExistWithArgsAsync(exchangeSpec, queueSpec,
+                bindArgs);
 
             await _management.Binding()
                 .SourceExchange(exchangeSpec)
@@ -252,14 +263,11 @@ await _management.Binding()
                 .Arguments(bindArgs)
                 .UnbindAsync();
 
-            await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistWithArgsAsync(exchangeSpec, queueSpec, bindArgs);
+            await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistWithArgsAsync(exchangeSpec, queueSpec,
+                bindArgs);
         }
-
-        /*
-         * NB: test DisposeAsync will do this.
-        await _management.ExchangeDeletion().Delete(source);
-        await _management.QueueDeletion().Delete(destination);
+        await exchangeSpec.DeleteAsync();
+        await queueSpec.DeleteAsync();
         await _connection.CloseAsync();
-        */
     }
 }