Merge pull request #1304 from rabbitmq/recovering-consumer-6.x

Add event for recovering consumer

Author: michaelklishin

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

@@ -45,5 +45,6 @@ public interface IAutorecoveringConnection : IConnection
 
         event EventHandler<ConsumerTagChangedAfterRecoveryEventArgs> ConsumerTagChangeAfterRecovery;
         event EventHandler<QueueNameChangedAfterRecoveryEventArgs> QueueNameChangeAfterRecovery;
+        event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer;
     }
 }

projects/RabbitMQ.Client/client/events/RecoveringConsumerEventArgs.cs

@@ -0,0 +1,64 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+//
+// The APL v2.0:
+//
+//---------------------------------------------------------------------------
+//   Copyright (c) 2007-2023 VMware, Inc.
+//
+//   Licensed under the Apache License, Version 2.0 (the "License");
+//   you may not use this file except in compliance with the License.
+//   You may obtain a copy of the License at
+//
+//       https://www.apache.org/licenses/LICENSE-2.0
+//
+//   Unless required by applicable law or agreed to in writing, software
+//   distributed under the License is distributed on an "AS IS" BASIS,
+//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//   See the License for the specific language governing permissions and
+//   limitations under the License.
+//---------------------------------------------------------------------------
+//
+// The MPL v2.0:
+//
+//---------------------------------------------------------------------------
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+//
+//  Copyright (c) 2007-2023 VMware, Inc.  All rights reserved.
+//---------------------------------------------------------------------------
+
+using System;
+using System.Collections.Generic;
+
+namespace RabbitMQ.Client.Events
+{
+    /// <summary>
+    /// Event related to consumer recovery, during automatic recovery.
+    /// </summary>
+    public class RecoveringConsumerEventArgs : EventArgs
+    {
+        /// <summary>
+        /// Constructs an event containing the consumer arguments and consumer
+        /// tag of the consumer this event relates to.
+        /// </summary>
+        /// <param name="consumerArguments">Consumer arguments of the consumer for this event</param>
+        /// <param name="consumerTag">Consumer tag of the consumer for this event</param>
+        public RecoveringConsumerEventArgs(IDictionary<string, object> consumerArguments, string consumerTag)
+        {
+            ConsumerArguments = consumerArguments;
+            ConsumerTag = consumerTag;
+        }
+
+        /// <summary>
+        /// Access the consumer arguments of the consumer this event relates to.
+        /// </summary>
+        public IDictionary<string, object> ConsumerArguments { get; }
+
+        /// <summary>
+        /// Access the consumer tag of the consumer this event relates to.
+        /// </summary>
+        public string ConsumerTag { get; }
+    }
+}

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

@@ -198,6 +198,7 @@ public event EventHandler<EventArgs> ConnectionUnblocked
             }
         }
 
+        public event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer;
         public event EventHandler<ConsumerTagChangedAfterRecoveryEventArgs> ConsumerTagChangeAfterRecovery;
         public event EventHandler<QueueNameChangedAfterRecoveryEventArgs> QueueNameChangeAfterRecovery;
 
@@ -1099,6 +1100,20 @@ internal void RecoverConsumers(AutorecoveringModel modelToRecover, IModel channe
                 string tag = pair.Key;
                 try
                 {
+                    foreach (EventHandler<RecoveringConsumerEventArgs> eh in RecoveringConsumer?.GetInvocationList() ?? Array.Empty<Delegate>())
+                    {
+                        try
+                        {
+                            var eventArgs = new RecoveringConsumerEventArgs(cons.Arguments, cons.ConsumerTag);
+                            eh(this, eventArgs);
+                        }
+                        catch (Exception e)
+                        {
+                            var args = new CallbackExceptionEventArgs(e) { Detail = { ["context"] = "OnBeforeRecoveringConsumer" } };
+                            _delegate.OnCallbackException(args);
+                        }
+                    }
+
                     string newTag = cons.Recover(channelToUse);
                     lock (_recordedConsumers)
                     {

projects/Unit/APIApproval.Approve.verified.txt

@@ -243,6 +243,7 @@ namespace RabbitMQ.Client
         event System.EventHandler<RabbitMQ.Client.Events.ConnectionRecoveryErrorEventArgs> ConnectionRecoveryError;
         event System.EventHandler<RabbitMQ.Client.Events.ConsumerTagChangedAfterRecoveryEventArgs> ConsumerTagChangeAfterRecovery;
         event System.EventHandler<RabbitMQ.Client.Events.QueueNameChangedAfterRecoveryEventArgs> QueueNameChangeAfterRecovery;
+        event System.EventHandler<RabbitMQ.Client.Events.RecoveringConsumerEventArgs> RecoveringConsumer;
         event System.EventHandler<System.EventArgs> RecoverySucceeded;
     }
     public interface IBasicConsumer
@@ -689,6 +690,12 @@ namespace RabbitMQ.Client.Events
         public string NameAfter { get; }
         public string NameBefore { get; }
     }
+    public class RecoveringConsumerEventArgs : System.EventArgs
+    {
+        public RecoveringConsumerEventArgs(System.Collections.Generic.IDictionary<string, object> consumerArguments, string consumerTag) { }
+        public System.Collections.Generic.IDictionary<string, object> ConsumerArguments { get; }
+        public string ConsumerTag { get; }
+    }
     public class RecoveryExceptionEventArgs : RabbitMQ.Client.Events.BaseExceptionEventArgs
     {
         public RecoveryExceptionEventArgs(System.Exception e) { }

projects/Unit/TestConnectionRecovery.cs

@@ -30,6 +30,7 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Collections;
 using System.Collections.Generic;
 using System.Threading;
 
@@ -688,6 +689,52 @@ public void TestRecoveryEventHandlersOnConnection()
             Assert.IsTrue(counter >= 3);
         }
 
+        [Test]
+        public void TestRecoveringConsumerHandlerOnConnection()
+        {
+            string q = Model.QueueDeclare(GenerateQueueName(), false, false, false, null).QueueName;
+            var cons = new EventingBasicConsumer(Model);
+            Model.BasicConsume(q, true, cons);
+
+            int counter = 0;
+            ((AutorecoveringConnection)Conn).RecoveringConsumer += (sender, args) => Interlocked.Increment(ref counter);
+
+            CloseAndWaitForRecovery();
+            Assert.IsTrue(Conn.IsOpen, "expected connection to be open");
+            Assert.AreEqual(1, counter);
+
+            CloseAndWaitForRecovery();
+            CloseAndWaitForRecovery();
+            Assert.IsTrue(Conn.IsOpen, "expected connection to be open");
+            Assert.AreEqual(3, counter);
+        }
+
+        [Test]
+        public void TestRecoveringConsumerHandlerOnConnection_EventArgumentsArePassedDown()
+        {
+            var myArgs = new Dictionary<string, object> { { "first-argument", "some-value" } };
+            string q = Model.QueueDeclare(GenerateQueueName(), false, false, false, null).QueueName;
+            var cons = new EventingBasicConsumer(Model);
+            string expectedCTag = Model.BasicConsume(cons, q, arguments: myArgs);
+
+            bool ctagMatches = false;
+            bool consumerArgumentMatches = false;
+            ((AutorecoveringConnection)Conn).RecoveringConsumer += (sender, args) =>
+            {
+                // We cannot assert here because NUnit throws when an assertion fails. This exception is caught and
+                // passed to a CallbackExceptionHandler, instead of failing the test. Instead, we have to do this trick
+                // and assert in the test function.
+                ctagMatches = args.ConsumerTag == expectedCTag;
+                consumerArgumentMatches = (string)args.ConsumerArguments["first-argument"] == "some-value";
+                args.ConsumerArguments["first-argument"] = "event-handler-set-this-value";
+            };
+
+            CloseAndWaitForRecovery();
+            Assert.That(ctagMatches, Is.True, "expected consumer tag to match");
+            Assert.That(consumerArgumentMatches, Is.True, "expected consumer arguments to match");
+            Assert.That(myArgs, Does.ContainKey("first-argument").WithValue("event-handler-set-this-value"));
+        }
+
         [Test]
         public void TestRecoveryEventHandlersOnModel()
         {