Added tests for topology recovery exception handler

Author: rosca-sabina

projects/RabbitMQ.Client/client/api/TopologyRecoveryExceptionHandler.cs

@@ -16,10 +16,10 @@ public class TopologyRecoveryExceptionHandler
         private Func<IRecordedQueue, Exception, bool> _queueRecoveryExceptionCondition;
         private Func<IRecordedBinding, Exception, bool> _bindingRecoveryExceptionCondition;
         private Func<IRecordedConsumer, Exception, bool> _consumerRecoveryExceptionCondition;
-        private Action<IRecordedExchange, Exception> _exchangeRecoveryExceptionHandler;
-        private Action<IRecordedQueue, Exception> _queueRecoveryExceptionHandler;
-        private Action<IRecordedBinding, Exception> _bindingRecoveryExceptionHandler;
-        private Action<IRecordedConsumer, Exception> _consumerRecoveryExceptionHandler;
+        private Action<IRecordedExchange, Exception, IConnection> _exchangeRecoveryExceptionHandler;
+        private Action<IRecordedQueue, Exception, IConnection> _queueRecoveryExceptionHandler;
+        private Action<IRecordedBinding, Exception, IConnection> _bindingRecoveryExceptionHandler;
+        private Action<IRecordedConsumer, Exception, IConnection> _consumerRecoveryExceptionHandler;
 
         /// <summary>
         /// Decides which exchange recovery exceptions the custom exception handler is applied to.
@@ -92,7 +92,7 @@ public Func<IRecordedConsumer, Exception, bool> ConsumerRecoveryExceptionConditi
         /// <summary>
         /// Retries, or otherwise handles, an exception thrown when attempting to recover an exchange.
         /// </summary>
-        public Action<IRecordedExchange, Exception> ExchangeRecoveryExceptionHandler
+        public Action<IRecordedExchange, Exception, IConnection> ExchangeRecoveryExceptionHandler
         {
             get => _exchangeRecoveryExceptionHandler;
 
@@ -108,7 +108,7 @@ public Action<IRecordedExchange, Exception> ExchangeRecoveryExceptionHandler
         /// <summary>
         /// Retries, or otherwise handles, an exception thrown when attempting to recover a queue.
         /// </summary>
-        public Action<IRecordedQueue, Exception> QueueRecoveryExceptionHandler
+        public Action<IRecordedQueue, Exception, IConnection> QueueRecoveryExceptionHandler
         {
             get => _queueRecoveryExceptionHandler;
 
@@ -124,7 +124,7 @@ public Action<IRecordedQueue, Exception> QueueRecoveryExceptionHandler
         /// <summary>
         /// Retries, or otherwise handles, an exception thrown when attempting to recover a binding.
         /// </summary>
-        public Action<IRecordedBinding, Exception> BindingRecoveryExceptionHandler
+        public Action<IRecordedBinding, Exception, IConnection> BindingRecoveryExceptionHandler
         {
             get => _bindingRecoveryExceptionHandler;
 
@@ -141,7 +141,7 @@ public Action<IRecordedBinding, Exception> BindingRecoveryExceptionHandler
         /// Retries, or otherwise handles, an exception thrown when attempting to recover a consumer.
         /// Is only called when the exception did not cause the consumer's channel to close.
         /// </summary>
-        public Action<IRecordedConsumer, Exception> ConsumerRecoveryExceptionHandler
+        public Action<IRecordedConsumer, Exception, IConnection> ConsumerRecoveryExceptionHandler
         {
             get => _consumerRecoveryExceptionHandler;
 

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

@@ -224,7 +224,7 @@ private void RecoverExchanges(RecoveryChannelFactory recoveryChannelFactory)
                     if (_config.TopologyRecoveryExceptionHandler.ExchangeRecoveryExceptionHandler != null
                         && _config.TopologyRecoveryExceptionHandler.ExchangeRecoveryExceptionCondition(recordedExchange, ex))
                     {
-                        _config.TopologyRecoveryExceptionHandler.ExchangeRecoveryExceptionHandler(recordedExchange, ex);
+                        _config.TopologyRecoveryExceptionHandler.ExchangeRecoveryExceptionHandler(recordedExchange, ex, this);
                     }
                     else
                     {
@@ -270,7 +270,7 @@ private void RecoverQueues(RecoveryChannelFactory recoveryChannelFactory)
                     if (_config.TopologyRecoveryExceptionHandler.QueueRecoveryExceptionHandler != null
                         && _config.TopologyRecoveryExceptionHandler.QueueRecoveryExceptionCondition(recordedQueue, ex))
                     {
-                        _config.TopologyRecoveryExceptionHandler.QueueRecoveryExceptionHandler(recordedQueue, ex);
+                        _config.TopologyRecoveryExceptionHandler.QueueRecoveryExceptionHandler(recordedQueue, ex, this);
                     }
                     else
                     {
@@ -293,7 +293,7 @@ private void RecoverBindings(RecoveryChannelFactory recoveryChannelFactory)
                     if (_config.TopologyRecoveryExceptionHandler.BindingRecoveryExceptionHandler != null
                         && _config.TopologyRecoveryExceptionHandler.BindingRecoveryExceptionCondition(binding, ex))
                     {
-                        _config.TopologyRecoveryExceptionHandler.BindingRecoveryExceptionHandler(binding, ex);
+                        _config.TopologyRecoveryExceptionHandler.BindingRecoveryExceptionHandler(binding, ex, this);
                     }
                     else
                     {
@@ -330,7 +330,7 @@ internal void RecoverConsumers(AutorecoveringChannel channelToRecover, IChannel
                     if (_config.TopologyRecoveryExceptionHandler.ConsumerRecoveryExceptionHandler != null
                         && _config.TopologyRecoveryExceptionHandler.ConsumerRecoveryExceptionCondition(consumer, ex))
                     {
-                        _config.TopologyRecoveryExceptionHandler.ConsumerRecoveryExceptionHandler(consumer, ex);
+                        _config.TopologyRecoveryExceptionHandler.ConsumerRecoveryExceptionHandler(consumer, ex, this);
                     }
                     else
                     {

projects/Unit/Fixtures.cs

@@ -175,6 +175,18 @@ internal AutorecoveringConnection CreateAutorecoveringConnectionWithTopologyReco
             return (AutorecoveringConnection)cf.CreateConnection($"{_testDisplayName}:{Guid.NewGuid()}");
         }
 
+        internal AutorecoveringConnection CreateAutorecoveringConnectionWithTopologyRecoveryExceptionHandler(TopologyRecoveryExceptionHandler handler)
+        {
+            var cf = new ConnectionFactory
+            {
+                AutomaticRecoveryEnabled = true,
+                TopologyRecoveryEnabled = true,
+                TopologyRecoveryExceptionHandler = handler
+            };
+
+            return (AutorecoveringConnection)cf.CreateConnection($"{_testDisplayName}:{Guid.NewGuid()}");
+        }
+
         internal IConnection CreateConnectionWithContinuationTimeout(bool automaticRecoveryEnabled, TimeSpan continuationTimeout)
         {
             var cf = new ConnectionFactory

projects/Unit/TestConnectionRecovery.cs

@@ -1182,8 +1182,8 @@ public void TestTopologyRecoveryConsumerFilter()
             var exchange = "topology.recovery.exchange";
             var queueWithRecoveredConsumer = "topology.recovery.queue.1";
             var queueWithIgnoredConsumer = "topology.recovery.queue.2";
-            var binding1 = "recovered.binding";
-            var binding2 = "filtered.binding";
+            var binding1 = "recovered.binding.1";
+            var binding2 = "recovered.binding.2";
 
             ch.ExchangeDeclare(exchange, "direct");
             ch.QueueDeclare(queueWithRecoveredConsumer, false, false, false, null);
@@ -1278,6 +1278,209 @@ public void TestTopologyRecoveryDefaultFilterRecoversAllEntities()
             Assert.True(consumerLatch2.Wait(TimeSpan.FromSeconds(5)));
         }
 
+        [Fact]
+        public void TestTopologyRecoveryQueueExceptionHandler()
+        {
+            var changedQueueArguments = new Dictionary<string, object>
+            {
+                { Headers.XMaxPriority, 20 }
+            };
+            var exceptionHandler = new TopologyRecoveryExceptionHandler
+            {
+                QueueRecoveryExceptionCondition = (rq, ex) =>
+                {
+                    return rq.Name.Contains("exception")
+                        && ex is OperationInterruptedException operationInterruptedException
+                        && operationInterruptedException.ShutdownReason.ReplyCode == Constants.PreconditionFailed;
+                },
+                QueueRecoveryExceptionHandler = (rq, ex, connection) =>
+                {
+                    using (var model = connection.CreateModel())
+                    {
+                        model.QueueDeclare(rq.Name, false, false, false, changedQueueArguments);
+                    }
+                }
+            };
+            var latch = new ManualResetEventSlim(false);
+            AutorecoveringConnection conn = CreateAutorecoveringConnectionWithTopologyRecoveryExceptionHandler(exceptionHandler);
+            conn.RecoverySucceeded += (source, ea) => latch.Set();
+            IModel ch = conn.CreateModel();
+
+            var queueToRecoverWithException = "recovery.exception.queue";
+            var queueToRecoverSuccessfully = "successfully.recovered.queue";
+            ch.QueueDeclare(queueToRecoverWithException, false, false, false, null);
+            ch.QueueDeclare(queueToRecoverSuccessfully, false, false, false, null);
+
+            _model.QueueDelete(queueToRecoverSuccessfully);
+            _model.QueueDelete(queueToRecoverWithException);
+            _model.QueueDeclare(queueToRecoverWithException, false, false, false, changedQueueArguments);
+
+            CloseAndWaitForRecovery(conn);
+            Wait(latch);
+
+            Assert.True(ch.IsOpen);
+            AssertQueueRecovery(ch, queueToRecoverSuccessfully, false);
+            AssertQueueRecovery(ch, queueToRecoverWithException, false, changedQueueArguments);
+
+            //Cleanup
+            _model.QueueDelete(queueToRecoverWithException);
+        }
+
+        [Fact]
+        public void TestTopologyRecoveryExchangeExceptionHandler()
+        {
+            var exceptionHandler = new TopologyRecoveryExceptionHandler
+            {
+                ExchangeRecoveryExceptionCondition = (re, ex) =>
+                {
+                    return re.Name.Contains("exception")
+                        && ex is OperationInterruptedException operationInterruptedException
+                        && operationInterruptedException.ShutdownReason.ReplyCode == Constants.PreconditionFailed;
+                },
+                ExchangeRecoveryExceptionHandler = (re, ex, connection) =>
+                {
+                    using (var model = connection.CreateModel())
+                    {
+                        model.ExchangeDeclare(re.Name, "topic", false, false);
+                    }
+                }
+            };
+            var latch = new ManualResetEventSlim(false);
+            AutorecoveringConnection conn = CreateAutorecoveringConnectionWithTopologyRecoveryExceptionHandler(exceptionHandler);
+            conn.RecoverySucceeded += (source, ea) => latch.Set();
+            IModel ch = conn.CreateModel();
+
+            var exchangeToRecoverWithException = "recovery.exception.exchange";
+            var exchangeToRecoverSuccessfully = "successfully.recovered.exchange";
+            ch.ExchangeDeclare(exchangeToRecoverWithException, "direct", false, false);
+            ch.ExchangeDeclare(exchangeToRecoverSuccessfully, "direct", false, false);
+
+            _model.ExchangeDelete(exchangeToRecoverSuccessfully);
+            _model.ExchangeDelete(exchangeToRecoverWithException);
+            _model.ExchangeDeclare(exchangeToRecoverWithException, "topic", false, false);
+
+            CloseAndWaitForRecovery(conn);
+            Wait(latch);
+
+            Assert.True(ch.IsOpen);
+            AssertExchangeRecovery(ch, exchangeToRecoverSuccessfully);
+            AssertExchangeRecovery(ch, exchangeToRecoverWithException);
+
+            //Cleanup
+            _model.ExchangeDelete(exchangeToRecoverWithException);
+        }
+
+        [Fact]
+        public void TestTopologyRecoveryBindingExceptionHandler()
+        {
+            var exchange = "topology.recovery.exchange";
+            var queueWithExceptionBinding = "recovery.exception.queue";
+            var bindingToRecoverWithException = "recovery.exception.binding";
+
+            var exceptionHandler = new TopologyRecoveryExceptionHandler
+            {
+                BindingRecoveryExceptionCondition = (b, ex) =>
+                {
+                    return b.RoutingKey.Contains("exception")
+                        && ex is OperationInterruptedException operationInterruptedException
+                        && operationInterruptedException.ShutdownReason.ReplyCode == Constants.NotFound;
+                },
+                BindingRecoveryExceptionHandler = (b, ex, connection) =>
+                {
+                    using (var model = connection.CreateModel())
+                    {
+                        model.QueueDeclare(queueWithExceptionBinding, false, false, false, null);
+                        model.QueueBind(queueWithExceptionBinding, exchange, bindingToRecoverWithException);
+                    }
+                }
+            };
+            var latch = new ManualResetEventSlim(false);
+            AutorecoveringConnection conn = CreateAutorecoveringConnectionWithTopologyRecoveryExceptionHandler(exceptionHandler);
+            conn.RecoverySucceeded += (source, ea) => latch.Set();
+            IModel ch = conn.CreateModel();
+
+            var queueWithRecoveredBinding = "successfully.recovered.queue";
+            var bindingToRecoverSuccessfully = "successfully.recovered.binding";
+
+            _model.QueueDeclare(queueWithExceptionBinding, false, false, false, null);
+
+            ch.ExchangeDeclare(exchange, "direct");
+            ch.QueueDeclare(queueWithRecoveredBinding, false, false, false, null);
+            ch.QueueBind(queueWithRecoveredBinding, exchange, bindingToRecoverSuccessfully);
+            ch.QueueBind(queueWithExceptionBinding, exchange, bindingToRecoverWithException);
+            ch.QueuePurge(queueWithRecoveredBinding);
+            ch.QueuePurge(queueWithExceptionBinding);
+
+            _model.QueueUnbind(queueWithRecoveredBinding, exchange, bindingToRecoverSuccessfully);
+            _model.QueueUnbind(queueWithExceptionBinding, exchange, bindingToRecoverWithException);
+            _model.QueueDelete(queueWithExceptionBinding);
+
+            CloseAndWaitForRecovery(conn);
+            Wait(latch);
+
+            Assert.True(ch.IsOpen);
+            Assert.True(SendAndConsumeMessage(queueWithRecoveredBinding, exchange, bindingToRecoverSuccessfully));
+            Assert.False(SendAndConsumeMessage(queueWithExceptionBinding, exchange, bindingToRecoverWithException));
+        }
+
+        [Fact]
+        public void TestTopologyRecoveryConsumerExceptionHandler()
+        {
+            var queueWithExceptionConsumer = "recovery.exception.queue";
+
+            var exceptionHandler = new TopologyRecoveryExceptionHandler
+            {
+                ConsumerRecoveryExceptionCondition = (c, ex) =>
+                {
+                    return c.ConsumerTag.Contains("exception")
+                        && ex is OperationInterruptedException operationInterruptedException
+                        && operationInterruptedException.ShutdownReason.ReplyCode == Constants.NotFound;
+                },
+                ConsumerRecoveryExceptionHandler = (c, ex, connection) =>
+                {
+                    using (var model = connection.CreateModel())
+                    {
+                        model.QueueDeclare(queueWithExceptionConsumer, false, false, false, null);
+                        model.BasicConsume(queueWithExceptionConsumer, true, c.ConsumerTag, c.Consumer);
+                    }
+                }
+            };
+            var latch = new ManualResetEventSlim(false);
+            AutorecoveringConnection conn = CreateAutorecoveringConnectionWithTopologyRecoveryExceptionHandler(exceptionHandler);
+            conn.RecoverySucceeded += (source, ea) => latch.Set();
+            IModel ch = conn.CreateModel();
+            ch.ConfirmSelect();
+
+            _model.QueueDeclare(queueWithExceptionConsumer, false, false, false, null);
+            _model.QueuePurge(queueWithExceptionConsumer);
+
+            var recoverLatch = new ManualResetEventSlim(false);
+            var consumerToRecover = new EventingBasicConsumer(ch);
+            consumerToRecover.Received += (source, ea) => recoverLatch.Set();
+            ch.BasicConsume(queueWithExceptionConsumer, true, "exception.consumer", consumerToRecover);
+
+            _model.QueueDelete(queueWithExceptionConsumer);
+
+            CloseAndWaitForRecovery(conn);
+            Wait(latch);
+
+            Assert.True(ch.IsOpen);
+
+            ch.BasicPublish("", queueWithExceptionConsumer, Encoding.UTF8.GetBytes("test message"));
+
+            Assert.True(recoverLatch.Wait(TimeSpan.FromSeconds(5)));
+
+            try
+            {
+                ch.BasicConsume(queueWithExceptionConsumer, true, "exception.consumer", consumerToRecover);
+                Assert.Fail("Expected an exception");
+            }
+            catch (OperationInterruptedException e)
+            {
+                AssertShutdownError(e.ShutdownReason, 530); // NOT_ALLOWED - not allowed to reuse consumer tag
+            }
+        }
+
         internal bool SendAndConsumeMessage(string queue, string exchange, string routingKey)
         {
             using (var ch = _conn.CreateModel())
@@ -1313,15 +1516,15 @@ internal void AssertQueueRecovery(IChannel m, string q)
             AssertQueueRecovery(m, q, true);
         }
 
-        internal void AssertQueueRecovery(IChannel m, string q, bool exclusive)
+        internal void AssertQueueRecovery(IChannel m, string q, bool exclusive, IDictionary<string, object> arguments = null)
         {
             m.ConfirmSelect();
             m.QueueDeclarePassive(q);
-            QueueDeclareOk ok1 = m.QueueDeclare(q, false, exclusive, false, null);
+            QueueDeclareOk ok1 = m.QueueDeclare(q, false, exclusive, false, arguments);
             Assert.Equal(0u, ok1.MessageCount);
             m.BasicPublish("", q, _messageBody);
             Assert.True(WaitForConfirms(m));
-            QueueDeclareOk ok2 = m.QueueDeclare(q, false, exclusive, false, null);
+            QueueDeclareOk ok2 = m.QueueDeclare(q, false, exclusive, false, arguments);
             Assert.Equal(1u, ok2.MessageCount);
         }