Add event for recovering consumer

Related to #1293. In the context of Streams over AMQP,
the consumer must keep track of the consumer offset.
When a consumer subscribes to a stream-type queue, it
may provide a consumer argument to specify a "point"
in the stream to attach to.

This library records consumers, and their arguments,
for topology recovery purposes. When a consumer is
declared, it starts reading at an arbitrary point e.g.
offset=123. After receiving messages, the offset "moves"
forward. In the event of a connection recovery due to
e.g. network error, the consumer is re-declared with
the offset recorded when it was first declared i.e.
offset=123. This is not correct, because the consumer
has received some messages, and the offset value when
it was first declared is not accurate anymore.

This commit adds an event that fire when a consumer is
about to be recovered, but has not started recovering yet.
This event exposes the consumer arguments as a reference
type, allowing an event handler to update the consumer
offset, or any other consumer argument.

Signed-off-by: Aitor Perez Cedres <[email protected]>

Author: Zerpet

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/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()
         {