From ea0f1bf98eb07ee430d671d044592bef0f643e40 Mon Sep 17 00:00:00 2001 From: Gabriele Santomaggio Date: Mon, 26 Aug 2024 15:23:14 +0200 Subject: [PATCH 1/3] improve bindings recovery remove the bindings when a queue or an exchange is removed Signed-off-by: Gabriele Santomaggio --- README.md | 4 +- .../Impl/RecordingTopologyListener.cs | 24 ++++ Tests/BindingsTests.cs | 46 +++--- Tests/ConnectionRecoveryTests.cs | 134 ++++++++++++++++++ 4 files changed, 187 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index 3382a375..4c8b2289 100644 --- a/README.md +++ b/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 diff --git a/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs b/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs index c41fc9bb..f23711c7 100644 --- a/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs +++ b/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs @@ -24,6 +24,28 @@ public class RecordingTopologyListener : ITopologyListener private readonly ConcurrentDictionary _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) + { + _bindingSpecifications.TryRemove(binding.Path, out _); + } + } + } + public void QueueDeclared(IQueueSpecification specification) { @@ -33,6 +55,7 @@ public void QueueDeclared(IQueueSpecification specification) public void QueueDeleted(string name) { _queueSpecifications.TryRemove(name, out _); + RemoveBindingsSpecificationFromQueue(name); } public void ExchangeDeclared(IExchangeSpecification specification) @@ -43,6 +66,7 @@ public void ExchangeDeclared(IExchangeSpecification specification) public void ExchangeDeleted(string name) { _exchangeSpecifications.TryRemove(name, out _); + RemoveBindingsSpecificationFromExchange(name); } public void BindingDeclared(IBindingSpecification specification) diff --git a/Tests/BindingsTests.cs b/Tests/BindingsTests.cs index 60ed70ae..7475723c 100644 --- a/Tests/BindingsTests.cs +++ b/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(); var bindingSpecTasks = new List(); for (int i = 0; i < 10; i++) { @@ -219,13 +226,14 @@ public async Task MultiBindingsBetweenExchangeAndQueuesWithArgumentsDifferentVal .DestinationQueue(queueSpec) .Key(key) // single key to use different args .Arguments(new Dictionary() { { $"是英国v_{i}", $"p_{i}" } }); - bindingSpecs.Add(bindingSpec); bindingSpecTasks.Add(bindingSpec.BindAsync()); } + await WhenAllComplete(bindingSpecTasks); bindingSpecTasks.Clear(); - var specialBindArgs = new Dictionary() { { $"v_8", $"p_8" }, { $"v_1", 1 }, { $"v_r", 1000L }, }; + var specialBindArgs = + new Dictionary() { { $"v_8", $"p_8" }, { $"v_1", 1 }, { $"v_r", 1000L }, }; IBindingSpecification specialBindSpec = _management.Binding() .SourceExchange(exchangeSpec) .DestinationQueue(queueSpec) @@ -233,17 +241,20 @@ public async Task MultiBindingsBetweenExchangeAndQueuesWithArgumentsDifferentVal .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() { { $"是英国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(); - */ } } diff --git a/Tests/ConnectionRecoveryTests.cs b/Tests/ConnectionRecoveryTests.cs index 93addade..8ad5ed27 100644 --- a/Tests/ConnectionRecoveryTests.cs +++ b/Tests/ConnectionRecoveryTests.cs @@ -420,4 +420,138 @@ await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistAsync("exchan await connection.CloseAsync(); } + + + /// + /// Test if removed a queue should remove the bindings. + /// In this case there are two queues, one will be deleted to check if the bindings are removed. + /// another queue will not be deleted to check if the bindings are not removed. + /// + /// Only at the end the queue will be removed and bindings should be zero. + /// + [Fact] + public async Task RemoveAQueueShouldRemoveTheBindings() + { + _testOutputHelper.WriteLine("RemoveAQueueShouldRemoveTheBindings"); + const string containerId = "remove-queue-should-remove-binding-connection-name"; + var connection = await AmqpConnection.CreateAsync( + ConnectionSettingBuilder.Create() + .RecoveryConfiguration(RecoveryConfiguration.Create() + .BackOffDelayPolicy(new FakeFastBackOffDelay()) + .Topology(true)) + .ContainerId(containerId) + .Build()); + + var management = connection.Management(); + var exSpec = management.Exchange().Name("e-remove-a-should-remove-binding") + .Type(ExchangeType.DIRECT); + + await exSpec.DeclareAsync(); + + var queueSpec = management.Queue().Name("q-remove-a-should-remove-binding") + .AutoDelete(true).Exclusive(true); + await queueSpec.DeclareAsync(); + + var queueSpecWontDeleted = management.Queue().Name("q-remove-a-should-remove-binding-wont-delete") + .AutoDelete(true).Exclusive(true); + + await queueSpecWontDeleted.DeclareAsync(); + + for (int i = 0; i < 10; i++) + { + await management.Binding().SourceExchange(exSpec) + .DestinationQueue(queueSpec).Key($"key_{i}").BindAsync(); + + await management.Binding().SourceExchange(exSpec) + .DestinationQueue(queueSpecWontDeleted).Key($"key_{i}").BindAsync(); + } + + await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueExistAsync("e-remove-a-should-remove-binding", + "q-remove-a-should-remove-binding"); + + await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueExistAsync("e-remove-a-should-remove-binding", + "q-remove-a-should-remove-binding-wont-delete"); + + + Assert.Equal(20, management.TopologyListener().BindingCount()); + await queueSpec.DeleteAsync(); + + await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistAsync("e-remove-a-should-remove-binding", + "q-remove-a-should-remove-binding"); + + Assert.Equal(10, management.TopologyListener().BindingCount()); + await queueSpecWontDeleted.DeleteAsync(); + + await exSpec.DeleteAsync(); + + Assert.Equal(0, management.TopologyListener().ExchangeCount()); + Assert.Equal(0, management.TopologyListener().QueueCount()); + + await connection.CloseAsync(); + } + + [Fact] + + public async Task RemoveAnExchangeShouldRemoveTheBindings() + { + _testOutputHelper.WriteLine("RemoveAnExchangeShouldRemoveTheBindings"); + const string containerId = "remove-exchange-should-remove-binding-connection-name"; + var connection = await AmqpConnection.CreateAsync( + ConnectionSettingBuilder.Create() + .RecoveryConfiguration(RecoveryConfiguration.Create() + .BackOffDelayPolicy(new FakeFastBackOffDelay()) + .Topology(true)) + .ContainerId(containerId) + .Build()); + + var management = connection.Management(); + var exSpec = management.Exchange().Name("e-remove-exchange-should-remove-binding") + .Type(ExchangeType.DIRECT); + + await exSpec.DeclareAsync(); + + var exSpecWontDeleted = management.Exchange().Name("e-remove-exchange-should-remove-binding-wont-delete") + .Type(ExchangeType.DIRECT); + + await exSpecWontDeleted.DeclareAsync(); + + var queueSpec = management.Queue().Name("q-remove-exchange-should-remove-binding") + .AutoDelete(true).Exclusive(true); + await queueSpec.DeclareAsync(); + + for (int i = 0; i < 10; i++) + { + await management.Binding().SourceExchange(exSpec) + .DestinationQueue(queueSpec).Key($"key_{i}").BindAsync(); + + await management.Binding().SourceExchange(exSpecWontDeleted) + .DestinationQueue(queueSpec).Key($"key_{i}").BindAsync(); + } + + await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueExistAsync("e-remove-exchange-should-remove-binding", + "q-remove-exchange-should-remove-binding"); + + await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueExistAsync( + "e-remove-exchange-should-remove-binding-wont-delete", + "q-remove-exchange-should-remove-binding"); + + Assert.Equal(20, management.TopologyListener().BindingCount()); + await exSpec.DeleteAsync(); + + await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistAsync( + "e-remove-exchange-should-remove-binding", + "q-remove-exchange-should-remove-binding"); + + Assert.Equal(10, management.TopologyListener().BindingCount()); + await exSpecWontDeleted.DeleteAsync(); + + await queueSpec.DeleteAsync(); + + Assert.Equal(0, management.TopologyListener().ExchangeCount()); + Assert.Equal(0, management.TopologyListener().QueueCount()); + + await connection.CloseAsync(); + + + } } From 2c113c5a17c9202df2a7f3af1e3d6727d2a1853e Mon Sep 17 00:00:00 2001 From: Gabriele Santomaggio Date: Mon, 26 Aug 2024 15:25:51 +0200 Subject: [PATCH 2/3] improve bindings recovery remove the bindings when a queue or an exchange is removed Signed-off-by: Gabriele Santomaggio --- RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs b/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs index f23711c7..748216be 100644 --- a/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs +++ b/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs @@ -39,7 +39,8 @@ private void RemoveBindingsSpecificationFromExchange(string exchangeName) { foreach (var binding in _bindingSpecifications.Values) { - if (binding.SourceExchange == exchangeName) + if (binding.SourceExchange == exchangeName + || binding.DestinationExchange == exchangeName) { _bindingSpecifications.TryRemove(binding.Path, out _); } From 15179b2dc483e48fec4126af69848cfe020ebea4 Mon Sep 17 00:00:00 2001 From: Gabriele Santomaggio Date: Mon, 26 Aug 2024 15:56:50 +0200 Subject: [PATCH 3/3] improve bindings recovery remove the bindings when a queue or an exchange is removed Signed-off-by: Gabriele Santomaggio --- RabbitMQ.AMQP.Client/IEntities.cs | 2 +- RabbitMQ.AMQP.Client/ITopologyListener.cs | 8 +-- RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs | 4 +- RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs | 2 +- .../Impl/RecordingTopologyListener.cs | 2 +- Tests/BindingsTests.cs | 4 +- Tests/ConnectionRecoveryTests.cs | 58 ++++++++++++++++++- 7 files changed, 68 insertions(+), 12 deletions(-) diff --git a/RabbitMQ.AMQP.Client/IEntities.cs b/RabbitMQ.AMQP.Client/IEntities.cs index a2a330b7..ebbe392a 100644 --- a/RabbitMQ.AMQP.Client/IEntities.cs +++ b/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(); diff --git a/RabbitMQ.AMQP.Client/ITopologyListener.cs b/RabbitMQ.AMQP.Client/ITopologyListener.cs index 213517cc..4c1ec7b5 100644 --- a/RabbitMQ.AMQP.Client/ITopologyListener.cs +++ b/RabbitMQ.AMQP.Client/ITopologyListener.cs @@ -9,10 +9,10 @@ public interface ITopologyListener void ExchangeDeclared(IExchangeSpecification specification); void ExchangeDeleted(string name); - - + + void BindingDeclared(IBindingSpecification specification); - + void BindingDeleted(string path); void Clear(); @@ -20,6 +20,6 @@ public interface ITopologyListener int QueueCount(); int ExchangeCount(); - + int BindingCount(); } diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs b/RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs index a84d75ef..5f1741c9 100644 --- a/RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs +++ b/RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs @@ -567,8 +567,8 @@ await Management.Exchange(spec).DeclareAsync() } } } - - + + public async Task VisitBindingsAsync(IEnumerable bindingSpec) { // TODO this could be done in parallel diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs b/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs index d1c1428b..a71f026e 100644 --- a/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs +++ b/RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs @@ -106,7 +106,7 @@ public IBindingSpecification Binding() { return new AmqpBindingSpecification(this); } - + public IBindingSpecification Binding(BindingSpec spec) { return Binding() diff --git a/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs b/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs index 748216be..7af86bd9 100644 --- a/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs +++ b/RabbitMQ.AMQP.Client/Impl/RecordingTopologyListener.cs @@ -39,7 +39,7 @@ private void RemoveBindingsSpecificationFromExchange(string exchangeName) { foreach (var binding in _bindingSpecifications.Values) { - if (binding.SourceExchange == exchangeName + if (binding.SourceExchange == exchangeName || binding.DestinationExchange == exchangeName) { _bindingSpecifications.TryRemove(binding.Path, out _); diff --git a/Tests/BindingsTests.cs b/Tests/BindingsTests.cs index 7475723c..839b1f56 100644 --- a/Tests/BindingsTests.cs +++ b/Tests/BindingsTests.cs @@ -109,7 +109,7 @@ public async Task BindBetweenExchangeAndQueueTwoTimes() [InlineData("(~~~!!++@~./.,€€#!!±§##§¶¡€#¢)", "~~~!!++@----.", "==`£!-=+")] - + public async Task SimpleBindingsBetweenExchangeAndExchange(string sourceExchangeName, string destinationExchangeName, string key) @@ -154,7 +154,7 @@ await SystemUtils.WaitUntilBindingsBetweenExchangeAndExchangeDontExistAsync(sour [InlineData("B", 10000L, "H", 0.0001)] [InlineData("是英国", 10000.32, "W", 3.0001)] [InlineData("是英国", "是英国23", "W", 3.0001)] - [InlineData("(~~~!!++@----./.,€€#####§¶¡€#¢)","~~~!!++@----", "==`£!-=+", "===£!-=+")] + [InlineData("(~~~!!++@----./.,€€#####§¶¡€#¢)", "~~~!!++@----", "==`£!-=+", "===£!-=+")] public async Task BindingsBetweenExchangeAndQueuesWithArgumentsDifferentValues(string key1, object value1, string key2, object value2) { diff --git a/Tests/ConnectionRecoveryTests.cs b/Tests/ConnectionRecoveryTests.cs index 8ad5ed27..f47ba012 100644 --- a/Tests/ConnectionRecoveryTests.cs +++ b/Tests/ConnectionRecoveryTests.cs @@ -491,7 +491,6 @@ await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistAsync("e-remo } [Fact] - public async Task RemoveAnExchangeShouldRemoveTheBindings() { _testOutputHelper.WriteLine("RemoveAnExchangeShouldRemoveTheBindings"); @@ -551,6 +550,63 @@ await SystemUtils.WaitUntilBindingsBetweenExchangeAndQueueDontExistAsync( Assert.Equal(0, management.TopologyListener().QueueCount()); await connection.CloseAsync(); + } + + /// + /// This test is specific to check if the bindings are removed when a destination exchange is removed. + /// + [Fact] + public async Task RemoveAnExchangeBoundToAnotherExchangeShouldRemoveTheBindings() + { + _testOutputHelper.WriteLine("RemoveAnExchangeBoundToAnotherExchangeShouldRemoveTheBindings"); + + const string containerId = "remove-exchange-bound-to-another-exchange-should-remove-binding-connection-name"; + var connection = await AmqpConnection.CreateAsync( + ConnectionSettingBuilder.Create() + .RecoveryConfiguration(RecoveryConfiguration.Create() + .BackOffDelayPolicy(new FakeFastBackOffDelay()) + .Topology(true)) + .ContainerId(containerId) + .Build()); + + var management = connection.Management(); + + var exSpec = management.Exchange().Name("e-remove-exchange-bound-to-another-exchange-should-remove-binding") + .Type(ExchangeType.DIRECT); + + await exSpec.DeclareAsync(); + + var exSpecDestination = management.Exchange().Name("e-remove-exchange-bound-to-another-exchange-should-remove-binding-destination") + .Type(ExchangeType.DIRECT); + + await exSpecDestination.DeclareAsync(); + + for (int i = 0; i < 10; i++) + { + await management.Binding().SourceExchange(exSpec) + .DestinationExchange(exSpecDestination).Key($"key_{i}").BindAsync(); + } + + await SystemUtils.WaitUntilBindingsBetweenExchangeAndExchangeExistAsync("e-remove-exchange-bound-to-another-exchange-should-remove-binding", + "e-remove-exchange-bound-to-another-exchange-should-remove-binding-destination"); + + Assert.Equal(10, management.TopologyListener().BindingCount()); + + + + await exSpecDestination.DeleteAsync(); + Assert.Equal(0, management.TopologyListener().BindingCount()); + + await exSpec.DeleteAsync(); + + await SystemUtils.WaitUntilBindingsBetweenExchangeAndExchangeDontExistAsync("e-remove-exchange-bound-to-another-exchange-should-remove-binding", + "e-remove-exchange-bound-to-another-exchange-should-remove-binding-destination"); + + + Assert.Equal(0, management.TopologyListener().ExchangeCount()); + + await connection.CloseAsync(); + }