Implement a super class for producer and consumer (#28)

* Implement a super-class for producer and consumer
* to deal with the reconnection. AbstractReconnectLifeCycle tries to reconnect
internally. This is an edge case.
---------
Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

README.md

@@ -24,15 +24,17 @@ The client is distributed via [NuGet](https://www.nuget.org/packages/RabbitMQ.AM
 - [x] Simple Publish messages
 - [x] Implement backpressure (it is implemented with MaxInflightMessages `MaxInFlight(2000).`)
 - [x] Simple Consume messages
+- [x] Recovery connection on connection lost
+- [x] Recovery management on connection lost
+- [x] Recovery queues on connection lost
+- [x] Recovery publishers on connection lost
+- [x] Recovery consumers on connection lost
 - [ ] 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
 - [ ] Implement metrics ( See `System.Diagnostics.DiagnosticSource` [Link](https://learn.microsoft.com/en-us/dotnet/core/diagnostics/metrics-instrumentation) )
-- [x] Recovery connection on connection lost
-- [x] Recovery management on connection lost
-- [x] Recovery queues on connection lost
-- [x] Recovery publisher on connection lost
-- [x] Recovery consumer on connection lost
-- [ ] Docker image to test in LRE 
+- [ ] Recovery exchanges on connection lost
+- [ ] Recovery bindings on connection lost
+- [ ] Docker image to test in LRE [not mandatory]
 - [ ] Check the TODO in the code
 

RabbitMQ.AMQP.Client/IRecoveryConfiguration.cs

@@ -60,4 +60,6 @@ public interface IBackOffDelayPolicy
     /// or when the user wants to disable the backoff delay policy.
     /// </summary>
     bool IsActive();
+    
+    int CurrentAttempt { get; }
 }

RabbitMQ.AMQP.Client/Impl/AbstractLifeCycle.cs

@@ -1,6 +1,8 @@
+using Amqp;
+
 namespace RabbitMQ.AMQP.Client.Impl;
 
-public class AmqpClosedException(string message) : Exception(message);
+public class AmqpNotOpenException(string message) : Exception(message);
 
 public abstract class AbstractLifeCycle : ILifeCycle
 {
@@ -16,9 +18,18 @@ protected virtual Task OpenAsync()
 
     protected void ThrowIfClosed()
     {
-        if (State == State.Closed)
+        switch (State)
         {
-            throw new AmqpClosedException(GetType().Name);
+            case State.Closed:
+                throw new AmqpNotOpenException("Resource is closed");
+            case State.Closing:
+                throw new AmqpNotOpenException("Resource is closing");
+            case State.Reconnecting:
+                throw new AmqpNotOpenException("Resource is Reconnecting");
+            case State.Open:
+                break;
+            default:
+                throw new ArgumentOutOfRangeException();
         }
     }
 
@@ -41,4 +52,40 @@ protected void OnNewStatus(State newState, Error? error)
     public event LifeCycleCallBack? ChangeState;
 }
 
+public abstract class AbstractReconnectLifeCycle : AbstractLifeCycle
+{
+    private readonly BackOffDelayPolicy _backOffDelayPolicy = BackOffDelayPolicy.Create(2);
+
+    internal void ChangeStatus(State newState, Error? error)
+    {
+        OnNewStatus(newState, error);
+    }
 
+    internal async Task Reconnect()
+    {
+        try
+        {
+            int randomWait = Random.Shared.Next(300, 900);
+            Trace.WriteLine(TraceLevel.Information, $"{ToString()} is reconnecting in {randomWait} ms, " +
+                                                    $"attempt: {_backOffDelayPolicy.CurrentAttempt}");
+            await Task.Delay(randomWait).ConfigureAwait(false);
+            await OpenAsync().ConfigureAwait(false);
+            Trace.WriteLine(TraceLevel.Information,
+                $"{ToString()} is reconnected, attempt: {_backOffDelayPolicy.CurrentAttempt}");
+            _backOffDelayPolicy.Reset();
+        }
+        catch (Exception e)
+        {
+            // Here we give another chance to reconnect
+            // that's an edge case, where the link is not ready for some reason
+            // the backoff policy will be used to delay the reconnection and give just a few attempts
+            Trace.WriteLine(TraceLevel.Error, $"{ToString()} Failed to reconnect, {e.Message}");
+            int delay = _backOffDelayPolicy.Delay();
+            await Task.Delay(delay).ConfigureAwait(false);
+            if (_backOffDelayPolicy.IsActive())
+            {
+                await Reconnect().ConfigureAwait(false);
+            }
+        }
+    }
+}

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -278,7 +278,7 @@ private ClosedCallback MaybeRecoverConnection()
             {
                 // close all the sessions, if the connection is closed the sessions are not valid anymore
                 _nativePubSubSessions.ClearSessions();
-                
+
                 if (error != null)
                 {
                     //  we assume here that the connection is closed unexpectedly, since the error is not null
@@ -299,8 +299,8 @@ private ClosedCallback MaybeRecoverConnection()
                     // to reconnecting and all the events are fired
                     OnNewStatus(State.Reconnecting, Utils.ConvertError(error));
                     ChangeEntitiesStatus(State.Reconnecting, Utils.ConvertError(error));
-                   
-                    
+
+
                     await Task.Run(async () =>
                     {
                         bool connected = false;
@@ -320,8 +320,12 @@ await Task.Run(async () =>
                             try
                             {
                                 int next = _connectionSettings.RecoveryConfiguration.GetBackOffDelayPolicy().Delay();
+
                                 Trace.WriteLine(TraceLevel.Information,
-                                    $"Trying Recovering connection in {next} milliseconds. Info: {ToString()})");
+                                    $"Trying Recovering connection in {next} milliseconds, " +
+                                    $"attempt: {_connectionSettings.RecoveryConfiguration.GetBackOffDelayPolicy().CurrentAttempt}. " +
+                                    $"Info: {ToString()})");
+
                                 await Task.Delay(TimeSpan.FromMilliseconds(next))
                                     .ConfigureAwait(false);
 
@@ -365,8 +369,16 @@ await _recordingTopologyListener.Accept(visitor)
                         OnNewStatus(State.Open, null);
                         // after the connection is recovered we have to reconnect all the publishers and consumers
 
-                        await ReconnectEntities().ConfigureAwait(false);
+                        try
+                        {
+                            await ReconnectEntities().ConfigureAwait(false);
+                        }
+                        catch (Exception e)
+                        {
+                            Trace.WriteLine(TraceLevel.Error, $"Error trying to reconnect entities {e}. Info: {this}");
+                        }
                     }).ConfigureAwait(false);
+
                     return;
                 }
 

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -3,12 +3,12 @@
 
 namespace RabbitMQ.AMQP.Client.Impl;
 
-public class AmqpConsumer : AbstractLifeCycle, IConsumer
+public class AmqpConsumer : AbstractReconnectLifeCycle, IConsumer
 {
     private readonly AmqpConnection _connection;
     private readonly string _address;
     private readonly MessageHandler _messageHandler;
-    private readonly int _initialCredits = 0;
+    private readonly int _initialCredits;
     private readonly Map _filters;
     private ReceiverLink? _receiverLink;
 
@@ -37,7 +37,8 @@ protected sealed override Task OpenAsync()
             attachCompleted.WaitOne(TimeSpan.FromSeconds(5));
             if (_receiverLink.LinkState != LinkState.Attached)
             {
-                throw new ConsumerException("Failed to create receiver link. Link state is not attached, error: " +
+                throw new ConsumerException(
+                    $"{ToString()} Failed to create receiver link. Link state is not attached, error: " +
                     _receiverLink.Error?.ToString() ?? "Unknown error");
             }
 
@@ -46,7 +47,7 @@ protected sealed override Task OpenAsync()
         }
         catch (Exception e)
         {
-            throw new ConsumerException($"Failed to create receiver link, {e}");
+            throw new ConsumerException($"{ToString()} Failed to create receiver link, {e}");
         }
 
         return Task.CompletedTask;
@@ -97,23 +98,8 @@ public override async Task CloseAsync()
         _connection.Consumers.TryRemove(Id, out _);
     }
 
-
-    internal void ChangeStatus(State newState, Error? error)
-    {
-        OnNewStatus(newState, error);
-    }
-
-    internal async Task Reconnect()
+    public override string ToString()
     {
-        int randomWait = Random.Shared.Next(200, 800);
-        Trace.WriteLine(TraceLevel.Information, $"Consumer: {ToString()} is reconnecting in {randomWait} ms");
-        await Task.Delay(randomWait).ConfigureAwait(false);
-
-        if (_receiverLink != null)
-        {
-            await _receiverLink.DetachAsync().ConfigureAwait(false)!;
-        }
-
-        await OpenAsync().ConfigureAwait(false);
+        return $"Consumer{{Address='{_address}', id={Id} ConnectionName='{_connection}', State='{State}'}}";
     }
 }

RabbitMQ.AMQP.Client/Impl/AmqpConsumerBuilder.cs

@@ -5,8 +5,8 @@ namespace RabbitMQ.AMQP.Client.Impl;
 public class AmqpConsumerBuilder(AmqpConnection connection) : IConsumerBuilder
 {
     private string _queue = "";
-    private int _initialCredits = 1;
-    private Map _filters = new Map();
+    private int _initialCredits = 10;
+    private readonly Map _filters = new Map();
 
 
     public IConsumerBuilder Queue(string queue)

RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs

@@ -6,7 +6,7 @@
 
 namespace RabbitMQ.AMQP.Client.Impl;
 
-public class AmqpPublisher : AbstractLifeCycle, IPublisher
+public class AmqpPublisher : AbstractReconnectLifeCycle, IPublisher
 {
     private SenderLink? _senderLink = null;
 
@@ -38,15 +38,15 @@ protected sealed override Task OpenAsync()
             attachCompleted.WaitOne(TimeSpan.FromSeconds(5));
             if (_senderLink.LinkState != LinkState.Attached)
             {
-                throw new PublisherException("Failed to create sender link. Link state is not attached, error: " +
+                throw new PublisherException($"{ToString()} Failed to create sender link. Link state is not attached, error: " +
                     _senderLink.Error?.ToString() ?? "Unknown error");
             }
 
             return base.OpenAsync();
         }
         catch (Exception e)
         {
-            throw new PublisherException($"Failed to create sender link, {e}");
+            throw new PublisherException($"{ToString()} Failed to create sender link, {e}");
         }
     }
 
@@ -127,7 +127,7 @@ public Task Publish(IMessage message, OutcomeDescriptorCallback outcomeCallback)
                     }
                     else
                     {
-                        Trace.WriteLine(TraceLevel.Error, "Message not sent. Killing the process.");
+                        Trace.WriteLine(TraceLevel.Error, $"{ToString()} Message not sent. Killing the process.");
                         Process.GetCurrentProcess().Kill();
                     }
 
@@ -138,7 +138,7 @@ public Task Publish(IMessage message, OutcomeDescriptorCallback outcomeCallback)
         }
         catch (Exception e)
         {
-            throw new PublisherException($"Failed to publish message, {e}");
+            throw new PublisherException($"{ToString()} Failed to publish message, {e}");
         }
 
         return Task.CompletedTask;
@@ -151,6 +151,7 @@ public override async Task CloseAsync()
         {
             return;
         }
+
         OnNewStatus(State.Closing, null);
         try
         {
@@ -173,22 +174,8 @@ await _senderLink.CloseAsync()
     }
 
 
-    internal void ChangeStatus(State newState, Error? error)
-    {
-        OnNewStatus(newState, error);
-    }
-
-    internal async Task Reconnect()
+    public override string ToString()
     {
-        int randomWait = Random.Shared.Next(200, 800);
-        Trace.WriteLine(TraceLevel.Information, $"Publisher: {ToString()} is reconnecting in {randomWait} ms");
-        await Task.Delay(randomWait).ConfigureAwait(false);
-
-        if (_senderLink != null)
-        {
-            await _senderLink.DetachAsync().ConfigureAwait(false)!;
-        }
-
-        await OpenAsync().ConfigureAwait(false);
+        return $"Publisher{{Address='{_address}', id={Id} ConnectionName='{_connection}', State='{State}'}}";
     }
 }

RabbitMQ.AMQP.Client/Impl/ConnectionSettings.cs

@@ -81,10 +81,7 @@ public ConnectionSettings Build()
     {
         var c = new ConnectionSettings(_host, _port, _user,
             _password, _virtualHost,
-            _scheme, _connection)
-        {
-            RecoveryConfiguration = (RecoveryConfiguration)_recoveryConfiguration
-        };
+            _scheme, _connection) { RecoveryConfiguration = (RecoveryConfiguration)_recoveryConfiguration };
 
         return c;
     }
@@ -282,16 +279,27 @@ public static BackOffDelayPolicy Create()
     {
         return new BackOffDelayPolicy();
     }
+    
+    public static BackOffDelayPolicy Create(int maxAttempt)
+    {
+        return new BackOffDelayPolicy(maxAttempt);
+    }
 
     private BackOffDelayPolicy()
     {
     }
 
+    private BackOffDelayPolicy(int maxAttempt)
+    {
+        _maxAttempt = maxAttempt;
+    }
+
     private const int StartRandomMilliseconds = 500;
     private const int EndRandomMilliseconds = 1500;
 
     private int _attempt = 1;
-    private int _totalAttempt = 0;
+    private readonly int _maxAttempt = 12;
+
 
     private void ResetAfterMaxAttempt()
     {
@@ -304,26 +312,28 @@ private void ResetAfterMaxAttempt()
     public int Delay()
     {
         _attempt++;
-        _totalAttempt++;
+        CurrentAttempt++;
         ResetAfterMaxAttempt();
         return Random.Shared.Next(StartRandomMilliseconds, EndRandomMilliseconds) * _attempt;
     }
 
     public void Reset()
     {
         _attempt = 1;
-        _totalAttempt = 0;
+        CurrentAttempt = 0;
     }
 
     public bool IsActive()
     {
-        return _totalAttempt < 12;
+        return CurrentAttempt < _maxAttempt;
     }
 
+    public int CurrentAttempt { get; private set; } = 0;
+
 
     public override string ToString()
     {
-        return $"BackOffDelayPolicy{{ Attempt={_attempt}, TotalAttempt={_totalAttempt}, IsActive={IsActive} }}";
+        return $"BackOffDelayPolicy{{ Attempt={_attempt}, TotalAttempt={CurrentAttempt}, IsActive={IsActive} }}";
     }
 }
 

Tests/ConnectionRecoveryTests.cs

@@ -22,6 +22,7 @@ public void Reset()
     }
 
     public bool IsActive() => false;
+    public int CurrentAttempt => 1;
 }
 
 internal class FakeFastBackOffDelay : IBackOffDelayPolicy
@@ -36,6 +37,7 @@ public void Reset()
     }
 
     public bool IsActive() => true;
+    public int CurrentAttempt => 1;
 }
 
 public class ConnectionRecoveryTests(ITestOutputHelper testOutputHelper)